使用Autbahn.asyncio.Wamp非阻碍地运行多个应用程序

我正在尝试同时在Python中运行两个autobahn.asyncio.wamp.ApplicationSession。以前，我使用Autobahn库的修改进行了此操作，如本文的答案所建议的那样。我现在需要更多专业的解决方案。

搜索了一段时间后，这篇文章似乎很有希望，但是使用twisted库，而不是asyncio。我无法确定autobahn库的asyncio分支的类似解决方案，因为它似乎没有使用Reactors。

我遇到的主要问题是ApplicationRunner.run()正在阻止（这就是为什么我以前将其外包给线程），因此我不能仅在其之后运行第二个ApplicationRunner。

我确实需要同时访问2个WebSocket频道，我似乎无法与单个ApplicationSession一起使用。

到目前为止我的代码：

from autobahn.asyncio.wamp import ApplicationSession
from autobahn.asyncio.wamp import ApplicationRunner
from asyncio import coroutine
import time

channel1 = 'BTC_LTC'
channel2 = 'BTC_XMR'
class LTCComponent(ApplicationSession):
    def onConnect(self):
        self.join(self.config.realm)
    @coroutine
    def onJoin(self, details):
        def onTicker(*args, **kwargs):
            print('LTCComponent', args, kwargs)
        try:
            yield from self.subscribe(onTicker, channel1)
        except Exception as e:
            print("Could not subscribe to topic:", e)
class XMRComponent(ApplicationSession):
    def onConnect(self):
        self.join(self.config.realm)
    @coroutine
    def onJoin(self, details):
        def onTicker(*args, **kwargs):
            print('XMRComponent', args, kwargs)
        try:
            yield from self.subscribe(onTicker, channel2)
        except Exception as e:
            print("Could not subscribe to topic:", e)
def main():
    runner = ApplicationRunner("wss://api.poloniex.com:443", "realm1", extra={})
    runner.run(LTCComponent)
    runner.run(XMRComponent) # <- is not being called

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        quit()
    except Exception as e:
        print(time.time(), e)

我对autobahn库的了解有限，恐怕文档并没有太大改善我的情况。我在这里俯瞰一些东西吗？一个函数，一个参数，它可以使我能够复合我的组件或一次运行它们？

也许是此处提供的类似解决方案，它实现了替代ApplicationRunner？

相关主题

在扭曲的

中运行两个应用程序

在线程中运行AutoBahn ApplicationRunner

autobahn.wamp.papplicationsessionsession

autobahn.wamp.applicationrunner source

根据要求，使用 multithreading代码的 @stovfl的回答的追溯：

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/home/nils/anaconda3/lib/python3.5/threading.py", line     914, in _bootstrap_inner
    self.run()
  File "/home/nils/git/tools/gemini_wss/t2.py", line 27, in run
    self.appRunner.run(self.__ApplicationSession)
  File "/home/nils/anaconda3/lib/python3.5/site-packages/autobahn-    0.14.1-py3.5.egg/autobahn/asyncio/wamp.py", line 143,     in run
    transport_factory = WampWebSocketClientFactory(create,         url=self.url,                 serializers=self.serializers)
  File "/home/nils/anaconda3/lib/python3.5/site-packages/autobahn-    0.14.1-py3.5.egg/autobahn/asyncio/websocket.py", line     319, in __init__
    WebSocketClientFactory.__init__(self, *args, **kwargs)
  File "/home/nils/anaconda3/lib/python3.5/site-packages/autobahn-    0.14.1-py3.5.egg/autobahn/asyncio/websocket.py", line     268, in __init__
    self.loop = loop or asyncio.get_event_loop()
  File "/home/nils/anaconda3/lib/python3.5/asyncio/events.py",     line 626, in get_event_loop
    return get_event_loop_policy().get_event_loop()
  File "/home/nils/anaconda3/lib/python3.5/asyncio/events.py",     line 572, in get_event_loop
    % threading.current_thread().name)
RuntimeError: There is no current event loop in thread 'Thread-2'.
Exception in thread Thread-1:
**Same as in Thread-2**
...
RuntimeError: There is no current event loop in thread 'Thread-1'.

正如我从traceback所看到的那样，我们仅到达4个步骤2 of 4

来自 asyncio docs：
该模块提供了用于使用Coroutines编写单线程并发代码的基础架构，多路复用I/O访问插座和其他资源

所以我使用multithreading放弃了我的第一个建议。
我可以想象以下三个选项：

使用multiprocessing而不是multithreading
在asyncio loop中使用coroutine进行
def onJoin(self, details)中的channels之间的切换

第二个建议，使用multiprocessing的第一个选项。
我可以启动两个asyncio loops，因此appRunner.run(...)应该起作用。

您可以使用一个 class ApplicationSession如果channel是唯一的不同。如果您需要通过其他class ApplicationSession将其添加到args=

class __ApplicationSession(ApplicationSession):
        # ...
        try:
            yield from self.subscribe(onTicker, self.config.extra['channel'])
        except Exception as e:
            # ...
import multiprocessing as mp
import time
def ApplicationRunner_process(realm, channel):
        appRunner = ApplicationRunner("wss://api.poloniex.com:443", realm, extra={'channel': channel})
        appRunner.run(__ApplicationSession)
if __name__ == "__main__":
    AppRun = [{'process':None, 'channel':'BTC_LTC'},
              {'process': None, 'channel': 'BTC_XMR'}]
    for app in AppRun:
        app['process'] =  mp.Process(target = ApplicationRunner_process, args = ('realm1', app['channel'] ))
        app['process'].start()
        time.sleep(0.1)
    AppRun[0]['process'].join()
    AppRun[1]['process'].join()

按照您链接的扭曲方法，我设法使用asyncio设置获得了相同的行为 start_loop = false

import asyncio
from autobahn.asyncio.wamp import ApplicationSession, ApplicationRunner
runner1 = ApplicationRunner(url, realm, extra={'cli_id': 1})
coro1 = runner1.run(MyApplicationSession, start_loop=False)
runner2 = ApplicationRunner(url, realm, extra={'cli_id': 2})
coro2 = runner2.run(MyApplicationSession, start_loop=False)
asyncio.get_event_loop().run_until_complete(coro1)
asyncio.get_event_loop().run_until_complete(coro2)
asyncio.get_event_loop().run_forever()
class MyApplicationSession(ApplicationSession):
    def __init__(self, cfg):
        super().__init__(cfg)
        self.cli_id = cfg.extra['cli_id']
   def onJoin(self, details):
        print("session attached", self.cli_id)

相关内容

最新更新

热门标签：