我的问题:
我想知道Python中是否存在用于从协程端点(又称“接收器”或“消费者”)返回值的“最佳实践”模式.更一般地说,您将如何处理以下情况?
我的情景:
我有我的(制片人)> (过滤器)> (使用者)coroutine管道来处理基于文本的表并从中构建字典列表.我希望将在consumer中构建的对象返回给producer的原始调用者.
我的方法:
我的方法是设置一个独特的完成处理信号,每个协程都会检查.如果它听到信号,那么它将信号传递给它的孩子并产生返回的值.消费者只是产生其当前价值.
替代方法:
我考虑过:
>使用全局来保持所需对象“返回”给调用者.
>具有常规子例程的基于类的方法.
我应该为我的方案重新考虑这些的原因也是受欢迎的.
我的实施:
这是我所做的简化版本,包括所有关键组件.
import uuid
FINISH_PROCESSING_SIGNAL = uuid.uuid4()
def coroutine(func):
def start(*args,**kwargs):
cr = func(*args,**kwargs)
cr.next()
return cr
return start
# Sink
@coroutine
def list_builder():
# accepts objects and adds them to a list
_list = []
try:
while True:
data = (yield)
if data is FINISH_PROCESSING_SIGNAL:
yield _list
break
_list.append(data)
except GeneratorExit:
pass
# Filter
@coroutine
def user_data_filter(target=None):
if target is None:
target = list_builder()
header = "-+-"
footer = "Transfer Packets"
username = "User Name"
fullname = "Full Name"
note = "Description"
try:
while True:
user = {}
data = (yield)
if data is FINISH_PROCESSING_SIGNAL:
yield target.send(FINISH_PROCESSING_SIGNAL)
break
line = data
if header in line:
while True:
line = (yield)
if footer in line:
target.send(user)
break
elif username in line:
user["username"] = line.split('|')[1]
elif fullname in line:
user["fullname"] = line.split('|')[1]
elif note in line:
user["note"] = line.split('|')[1]
except GeneratorExit:
target.close()
# Producer
def process_users_table(table, target=None):
if target is None:
target = user_data_filter()
lines = table.split('\r\n')
for line in lines:
target.send(line)
processed_data = target.send(FINISH_PROCESSING_SIGNAL)
return processed_data
if __name__ == '__main__':
test_users_table = \
"""
Item |Value\r\n
----------------+-----------------------\r\n
User Name |alice\r\n
Full Name |Alice Doe\r\n
Description |\r\n
Transfer Packets|0\r\n
----------------+-----------------------\r\n
User Name |bob\r\n
Full Name |Bob Tables\r\n
Description |\r\n
Transfer Packets|0\r\n
"""
users = process_users_table(test_users_table)
print users
最佳答案 您发信号通知消费者终止的方法很好,并且与使用多处理或线程化队列时的操作相符.但是,生成器还有一种方法可以抛出异常(而不是发送值),而
the purpose of throw
恰好是向生成器发出事件或状态变化的信号.而且,
when an exception is thrown to a generator,
[i]f the
generator catches the exception and yields another value, that is
the return value of g.throw().
这似乎非常适合您的用例.您可以抛出FINISH_PROCESSING_SIGNAL异常,而不是发送FINISH_PROCESSING_SIGNAL值,并使用try..except来产生最终值.
class FINISH_PROCESSING_SIGNAL(Exception): pass
def coroutine(func):
def start(*args,**kwargs):
cr = func(*args,**kwargs)
cr.next()
return cr
return start
# Sink
@coroutine
def list_builder():
# accepts objects and adds them to a list
_list = []
try:
while True:
data = (yield)
_list.append(data)
except FINISH_PROCESSING_SIGNAL:
yield _list
# Filter
@coroutine
def user_data_filter(target=list_builder()):
header = "-+-"
footer = "Transfer Packets"
username = "User Name"
fullname = "Full Name"
note = "Description"
try:
while True:
user = {}
data = (yield)
line = data
if header in line:
while True:
line = (yield)
if footer in line:
target.send(user)
break
elif username in line:
user["username"] = line.split('|')[1]
elif fullname in line:
user["fullname"] = line.split('|')[1]
elif note in line:
user["note"] = line.split('|')[1]
except FINISH_PROCESSING_SIGNAL as err:
# Pass along the Exception to the target, and yield its result back
# to the caller
yield target.throw(err)
# Producer
def process_users_table(table, target=user_data_filter()):
lines = table.split('\r\n')
for line in lines:
target.send(line)
processed_data = target.throw(FINISH_PROCESSING_SIGNAL)
# processed_data = target.close()
return processed_data
if __name__ == '__main__':
test_users_table = \
"""
Item |Value\r\n
----------------+-----------------------\r\n
User Name |alice\r\n
Full Name |Alice Doe\r\n
Description |\r\n
Transfer Packets|0\r\n
----------------+-----------------------\r\n
User Name |bob\r\n
Full Name |Bob Tables\r\n
Description |\r\n
Transfer Packets|0\r\n
"""
users = process_users_table(test_users_table)
print users