websocket - Quart(异步 Flask)应用程序中的 Autobahn websocket 客户端
问题描述
各位晚上好。我对这个地方并不陌生,但最终决定注册并寻求帮助。我使用 Quart 框架(异步 Flask)开发了一个 Web 应用程序。现在随着应用程序变得更大更复杂,我决定将不同的程序分离到不同的服务器实例,这主要是因为我想保持 Web 服务器的清洁、更抽象和无计算负载。
因此,我计划将一台 Web 服务器与几个(如果需要)相同的过程服务器一起使用。所有服务器都基于quart框架,现在只是为了简化开发。我决定使用 Crossbar.io 路由器和高速公路将所有服务器连接在一起。
问题就在这里发生了。我关注了这个帖子:
使用 autbahn.asyncio.wamp 以非阻塞方式运行多个 ApplicationSession
如何使用高速公路异步实现交互式 websocket 客户端?
我如何将交叉开关客户端(python3,asyncio)与 tkinter 集成
如何从协议外部发送 Autobahn/Twisted WAMP 消息?
似乎我尝试了所有可能的方法来在我的 quart 应用程序中实现高速公路 websocket 客户端。我不知道如何使它成为可能,所以两件事都可以正常工作,无论 Quart 应用程序是否有效,但高速公路 WS 客户端是否无效,反之亦然。
简化我的夸脱应用程序如下所示:
from quart import Quart, request, current_app
from config import Config
# Autobahn
import asyncio
from autobahn import wamp
from autobahn.asyncio.wamp import ApplicationSession, ApplicationRunner
import concurrent.futures
class Component(ApplicationSession):
"""
An application component registering RPC endpoints using decorators.
"""
async def onJoin(self, details):
# register all methods on this object decorated with "@wamp.register"
# as a RPC endpoint
##
results = await self.register(self)
for res in results:
if isinstance(res, wamp.protocol.Registration):
# res is an Registration instance
print("Ok, registered procedure with registration ID {}".format(res.id))
else:
# res is an Failure instance
print("Failed to register procedure: {}".format(res))
@wamp.register(u'com.mathservice.add2')
def add2(self, x, y):
return x + y
def create_app(config_class=Config):
app = Quart(__name__)
app.config.from_object(config_class)
# Blueprint registration
from app.main import bp as main_bp
app.register_blueprint(main_bp)
print ("before autobahn start")
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
runner = ApplicationRunner('ws://127.0.0.1:8080 /ws', 'realm1')
future = executor.submit(runner.run(Component))
print ("after autobahn started")
return app
from app import models
在这种情况下,应用程序卡在跑步者循环中并且整个应用程序无法工作(无法服务请求),只有当我通过 Ctrl-C 中断跑步者(高速公路)循环时才有可能。
启动后的CMD:
(quart-app) user@car:~/quart-app$ hypercorn --debug --error-log - --access-log - -b 0.0.0.0:8001 tengine:app
Running on 0.0.0.0:8001 over http (CTRL + C to quit)
before autobahn start
Ok, registered procedure with registration ID 4605315769796303
按 ctrl-C 后:
...
^Cafter autobahn started
2019-03-29T01:06:52 <Server sockets=[<socket.socket fd=11, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('0.0.0.0', 8001)>]> is serving
如何使 quart 应用程序与高速公路客户端以非阻塞方式一起工作成为可能?因此高速公路打开并保持与 Crossbar 路由器的 websocket 连接,并在后台静默监听。
解决方案
好吧,经过许多不眠之夜,我终于找到了解决这个难题的好方法。
感谢这篇文章C-Python asyncio: running discord.py in a thread
因此,我像这样重写了我的代码,并且能够在内部运行带有高速公路客户端的 Quart 应用程序,并且两者都以非阻塞方式积极工作。整体__init__.py
看起来像:
from quart import Quart, request, current_app
from config import Config
def create_app(config_class=Config):
app = Quart(__name__)
app.config.from_object(config_class)
# Blueprint registration
from app.main import bp as main_bp
app.register_blueprint(main_bp)
return app
# Autobahn
import asyncio
from autobahn import wamp
from autobahn.asyncio.wamp import ApplicationSession, ApplicationRunner
import threading
class Component(ApplicationSession):
"""
An application component registering RPC endpoints using decorators.
"""
async def onJoin(self, details):
# register all methods on this object decorated with "@wamp.register"
# as a RPC endpoint
##
results = await self.register(self)
for res in results:
if isinstance(res, wamp.protocol.Registration):
# res is an Registration instance
print("Ok, registered procedure with registration ID {}".format(res.id))
else:
# res is an Failure instance
print("Failed to register procedure: {}".format(res))
def onDisconnect(self):
print('Autobahn disconnected')
@wamp.register(u'com.mathservice.add2')
def add2(self, x, y):
return x + y
async def start():
runner = ApplicationRunner('ws://127.0.0.1:8080/ws', 'realm1')
await runner.run(Component) # use client.start instead of client.run
def run_it_forever(loop):
loop.run_forever()
asyncio.get_child_watcher() # I still don't know if I need this method. It works without it.
loop = asyncio.get_event_loop()
loop.create_task(start())
print('Starting thread for Autobahn...')
thread = threading.Thread(target=run_it_forever, args=(loop,))
thread.start()
print ("Thread for Autobahn has been started...")
from app import models
在这种情况下,我们使用高速公路的 runner.run 创建任务并将其附加到当前循环,然后在新线程中永远运行此循环。
我对当前的解决方案非常满意....但后来发现这个解决方案有一些缺点,这对我来说至关重要,例如:如果连接断开(即交叉开关路由器不可用),则重新连接。使用这种方法,如果连接未能初始化或在一段时间后断开,它将不会尝试重新连接。此外,对我来说,如何使用 ApplicationSession API 并不明显,即从我的 quart 应用程序中的代码注册/调用 RPC。
幸运的是,我发现了 autobahn 在他们的文档中使用的另一个新组件 API: https ://autobahn.readthedocs.io/en/latest/wamp/programming.html#registering-procedures https://github.com/crossbario/autobahn-python /blob/master/examples/asyncio/wamp/component/backend.py
它具有自动重新连接功能,并且很容易使用装饰器为 RPC 注册功能@component.register('com.something.do')
,您只需要import component
之前。
所以这是__init__.py
解决方案的最终视图:
from quart import Quart, request, current_app
from config import Config
def create_app(config_class=Config):
...
return app
from autobahn.asyncio.component import Component, run
from autobahn.wamp.types import RegisterOptions
import asyncio
import ssl
import threading
component = Component(
transports=[
{
"type": "websocket",
"url": u"ws://localhost:8080/ws",
"endpoint": {
"type": "tcp",
"host": "localhost",
"port": 8080,
},
"options": {
"open_handshake_timeout": 100,
}
},
],
realm=u"realm1",
)
@component.on_join
def join(session, details):
print("joined {}".format(details))
async def start():
await component.start() #used component.start() instead of run([component]) as it's async function
def run_it_forever(loop):
loop.run_forever()
loop = asyncio.get_event_loop()
#asyncio.get_child_watcher() # I still don't know if I need this method. It works without it.
asyncio.get_child_watcher().attach_loop(loop)
loop.create_task(start())
print('Starting thread for Autobahn...')
thread = threading.Thread(target=run_it_forever, args=(loop,))
thread.start()
print ("Thread for Autobahn has been started...")
from app import models
我希望它会帮助某人。干杯!
推荐阅读
- css - CSS 集中元素
- java - Unity Android 方向变化动画显示最后的方向快照
- c# - 如何从 JsonNet 序列化 DateTimeOffset?
- ruby-on-rails - 如何防止工作每小时发送邮件超过一次?
- reactjs - Webpack:跨 html 页面的捆绑拆分与跨客户端路由的代码拆分
- php - 如何在作曲家中丢弃缓存中的负载
- eclipse - Eclipse 4.4 在 MacOS 上意外退出 - 从终端运行时有效
- xml - 在 XML 中,如何/为什么在定义名称空间之前使用它?
- ios - 在IOS中使用NSPredicate通过数组的一个元素过滤数组的NSArray
- python - 如何使用变量字符串数据在python中动态创建列表字典