我的问题:
我想知道Python中是否存在从协程终结点(也称为“接收器”或“消费者”)返回值的“最佳实践”模式。一般来说,您将如何处理以下情况?
我的情况:
我有(producer) > (filter) > (consumer)
协程管道来处理基于文本的表并从中构建字典列表。我希望将内置的对象consumer
返回给的原始调用方producer
。
我的方法:
我的方法是设置每个协程检查的独特的后处理信号。如果它听到信号,则将信号传递给它的子信号并产生返回值。在consumer
刚刚产生的电流值。
替代方法:
我考虑过:
我为什么应该为我的情况重新考虑这些原因也将受到欢迎。
我的实现:
这是我所做工作的简化版本,包括所有关键组件。
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
用信号通知使用者终止的方法很好,并且与使用多处理或线程化队列时的操作保持一致。但是,生成器还可以抛出异常(而不是发送值),并且的目的throw
正是向生成器发出事件或状态更改的信号。此外,当将异常抛出给生成器时,
[i] f生成器捕获异常并产生另一个值,即g.throw()的返回值。
这似乎非常适合您的用例。除了发送FINISH_PROCESSING_SIGNAL
值外,还可以引发FINISH_PROCESSING_SIGNAL
Exception并使用它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
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句