python - 从订阅者调用远程过程并解决 asyncio promise

Question

我在让基于 asyncio 的高速公路 RPC 在事件处理程序中工作时遇到问题:

from autobahn.asyncio import wamp
from autobahn.wamp import register, subscribe

class Foo(wamp.ApplicationSession):
    @subscribe('wamp.metaevent.session.on_join')
    def bar(self):
        baz = yield from self.call('baz')

    @register('baz')
    def baz(self):
        return 'baz'

阅读文档后，我的印象是这应该有效。但是，如果我在 Foo.bar 中使用 yield ，则它的任何代码都不会被执行。我尝试过以各种模式使用 asyncio.coroutine 进行装饰，但根本无法让它运行。

我发现让它发挥作用的唯一方法是“手动”解决返回的 future:

@subscribe('wamp.metaevent.session.on_join')
def bar(self):
    def do_something(f):
        print(f.result())

    f = self.call('baz')
    f.add_done_callback(do_something)

我确信我还没有正确理解异步编程，所以我必须做什么才能编写 baz = self.call('baz') 并获得立即结果(意味着没有额外明确声明的回调)？

Answer 1

the yield keyword的存在在 def 语句的主体中将定义的函数变成 generator function .

当您调用生成器函数(如autobahn)时，函数体不会被执行。相反，会创建一个生成器对象。当您在生成器上调用 next 时，控制将前进到下一个 yield 语句。生成器已在其他 Stack Overflow 帖子、文档和网络上的其他地方进行了广泛讨论。

asyncio 的 API 广泛使用 coroutines和产量。 (对于大多数用途，您可以将“协程”和“生成器”视为同义词。)event loop跟踪生成器集合并根据需要调用 next、send 和 throw。*

看起来 autobahn.subscribe 需要一个常规回调函数，而不是协程，这就是您的代码没有被执行的原因。解决此问题的一种方法是使用 asyncio.async 编写一个调度您的协程的回调函数。 .

class Foo(wamp.ApplicationSession):
    @subscribe('wamp.metaevent.session.on_join')
    def schedule_bar(self):
        coro = self.bar()  # create the coroutine object

        # wrap the coroutine in a Task and register it with the event loop.
        # the loop argument is optional and defaults to asyncio.get_event_loop()
        asyncio.async(coro, loop=my_event_loop)  

    @coroutine
    def bar(self):
        baz = yield from self.call('baz')

    @register('baz')
    def baz(self):
        return 'baz'

如果 autobahn 中没有函数来执行此操作，您可以编写自己的可重用装饰器来订阅 WAMP 主题的协程。

from functools import wraps

def subscribe_coro(uri, loop=None):
    def decorator(f):
        @subscribe(uri)
        @wraps(f)
        def wrapper(*args, **kwargs):
            coro = f(*args, **kwargs)
            asyncio.async(coro, loop=loop)
        return wrapper
    return decorator

现在你的类将如下所示:

class Foo(wamp.ApplicationSession):
    @subscribe_coro('wamp.metaevent.session.on_join')
    @coroutine
    def bar(self):
        baz = yield from self.call('baz')

    @register('baz')
    def baz(self):
        return 'baz'

_{* 这是一种简化。事件循环实际上跟踪 Future s，而不是协程。在生成器上调用适当方法的算法由 Task 实现。 ，它将协程包装成 Future。}