与 send
和 yield
我们可以与生成器进行双向通信并很好地实现状态机(参见下面的示例)。现在,我们不能(?)发送到 asyncio 协程,那么如何使用 asyncio 协程实现状态机呢?
发电机示例
def lock():
combination = [1, 2, 3]
for digit in combination:
a = (yield True)
while a != digit:
a = (yield False)
yield "You're in"
def main():
l = lock()
next(l)
assert l.send(2) == False
assert l.send(1) == True # correct value 1st digit
assert l.send(1) == False
assert l.send(2) == True # correct value 2nd digit
assert l.send(2) == False
assert l.send(3) == "You're in" # correct value 3rd digit
与 asyncio 类似的东西不太好..有更好的方法吗?
异步提案
import asyncio
class AsyncLock:
class Message:
def __init__(self, value):
self.f = asyncio.Future()
self.value = value
def set_result(self, v):
self.f.set_result(v)
async def result(self):
return await self.f
def __init__(self, msg_q):
self.msg_q = msg_q
self.task = None
async def send(self, value):
msg = AsyncLock.Message(value)
await self.msg_q.put(msg)
return await msg.result()
# all of the above to be able to do this:
async def run(self):
combination = [1, 2, 3]
for digit in combination:
msg = await self.msg_q.get()
while msg.value != digit:
msg.set_result(False)
msg = await self.msg_q.get()
msg.set_result("You're in" if digit == 3 else True)
async def amain():
l = AsyncLock(asyncio.Queue())
l.task = asyncio.ensure_future(l.run())
assert await l.send(2) == False
assert await l.send(1) == True
assert await l.send(1) == False
assert await l.send(2) == True
assert await l.send(2) == False
assert await l.send(3) == "You're in"
asyncio.get_event_loop().run_until_complete(amain())
最佳答案
Python3.6 添加了对异步生成器的支持 ( PEP525 ) 所以 async
函数现在也可以是生成器!
#!/usr/bin/env python3.6
import asyncio
async def lock():
combination = [1, 2, 3]
for digit in combination:
a = (yield True)
while a != digit:
a = (yield False)
yield "You're in!"
async def main():
coro = lock()
await coro.asend(None)
assert (await coro.asend(2)) == False
assert (await coro.asend(1)) == True
assert (await coro.asend(1)) == False
assert (await coro.asend(2)) == True
assert (await coro.asend(2)) == False
assert (await coro.asend(3)) == "You're in!"
print('Got it')
iol = asyncio.get_event_loop()
iol.run_until_complete(main())
在 Python3.6 之前,最好的方法是像以前一样使用消息队列。
关于python-3.x - 与协程的双向通信(asyncio 中的状态机),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51850462/