Twisted - 如何让大量 Python 代码变成非阻塞的答案

【问题标题】：Twisted - how to make lots of Python code non-blockingTwisted - 如何让大量 Python 代码变成非阻塞的
【发布时间】：2018-01-12 01:08:46
【问题描述】：

我一直试图让这个脚本以书面顺序执行hub() 中的代码。

hub() 包含标准 Python 代码和使用 Twisted 和 Crossbar 执行 I/O 的请求。

但是，由于 Python 代码是阻塞的，reactor 没有任何机会执行那些“发布”任务。 我的前端在最后收到所有发布的消息。

此代码是我实际处理的代码的大幅简化版本。真正的脚本（hub() 和它调用的其他方法）超过 1500 行。修改所有这些函数以使其成为非阻塞的并不理想。如果可以解决此问题，我宁愿将更改隔离到 publish() 等几个方法。
我玩过async、await、deferLater、loopingCall 等术语。我还没有找到对我的情况有帮助的例子。

有没有办法修改publish()（或hub()），让它们按顺序发送消息？

from autobahn.twisted.component import Component, run
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.internet import reactor, defer

component = Component(
    transports=[
        {
            u"type": u"websocket",
            u"url": u"ws://127.0.0.1:8080/ws",
            u"endpoint": {
                u"type": u"tcp",
                u"host": u"localhost",
                u"port": 8080,
            },
            u"options": {
                u"open_handshake_timeout": 100,
            }
        },
    ],
    realm=u"realm1",
)

@component.on_join
@inlineCallbacks
def join(session, details):
    print("joined {}: {}".format(session, details))

    def publish(context='output', value='default'):
        """ Publish a message. """
        print('publish', value)
        session.publish(u'com.myapp.universal_feedback', {"id": context, "value": value})

    def hub(thing):
        """ Main script. """
        do_things
        publish('output', 'some data for you')
        do_more_things
        publish('status', 'a progress message')
        do_even_more_things
        publish('status', 'some more data')
        do_all_the_things
        publish('other', 'something else')

    try:
        yield session.register(hub, u'com.myapp.hello')
        print("procedure registered")
    except Exception as e:
        print("could not register procedure: {0}".format(e))


if __name__ == "__main__":
    run([component])
    reactor.run()

【问题讨论】：

标签： wamp twisted autobahn twisted.internet crossbar

【解决方案1】：

你的join() 函数是异步的（用@inlineCallbacks 装饰并且在主体中至少包含一个yield）。

它在内部将函数 hub() 注册为 WAMP RPC； hub() 但不是异步的。

对session.publish() 的调用也不会像异步调用那样产生。

结果：您将一堆事件添加到事件循环中，但在应用程序关闭时刷新事件循环之前不要等待它们。

【讨论】：

【解决方案2】：

您需要使您的功能中心和发布异步。

@inlineCallbacks
def publish(context='output', value='default'):
    """ Publish a message. """
    print('publish', value)
    yield session.publish(u'com.myapp.universal_feedback', {"id": context, "value": value})

@inlineCallbacks
def hub(thing):
    """ Main script. """
    do_things
    yield publish('output', 'some data for you')
    do_more_things
    yield publish('status', 'a progress message')
    do_even_more_things
    yield publish('status', 'some more data')
    do_all_the_things
    yield publish('other', 'something else')

【讨论】：