在aiohttp应用程序中收听ZeroMQ
问题描述:
我运行aiohttp
应用程序Gunicorn
落后于nginx
。 在我的应用程序的初始化模块中,我不运行使用web.run_app(app)
的应用程序,但只是创建一个将由Gunicorn
导入的实例,以便在每个工作人员Gunicorn
创建中运行该实例。 因此,Gunicorn
创建了一些工作进程,它们中的事件循环,然后在这些循环中创建应用程序的请求处理程序runs。在aiohttp应用程序中收听ZeroMQ
我的aiohttp
应用程序有一个连接的WebSockets
(移动应用程序客户端)的集合,我想通知在由Gunicorn
启动的任何应用程序进程中发生的事件。 我想通知全部WebSockets
连接到所有应用程序。 因此,我使用ZeroMQ
创建某种上游代理,我想使用每个应用程序进程的zmq.SUB
套接字来订阅它。
...所以基本上我想要实现在每个应用程序的工人是这样的:
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect('tcp://localhost:5555')
while True:
event = socket.recv()
for ws in app['websockets']:
ws.send_bytes(event)
# break before app shutdown. How?
我怎么能听aiohttp
应用程序内的ZeroMQ
代理将邮件转发到WebSockets
?
我在哪里可以将此代码在事件循环内的后台运行以及如何在aiohttp
应用程序的生命周期内正确运行和关闭它?
UPDATE
我已经在aiohttp的GitHub的库中创建一个issue描述该问题,并提出可能的解决方案。对于所描述的问题,我会非常感谢在这里或那里提出的意见。
答
好吧,这个问题并在此issue的讨论导致了新的功能,我到aiohttp
贡献,即在版本1.0 ,我们将使用Application.on_startup()
方法注册on_startup
应用信号的能力。
Documentation。
Working example on the master branch。
#!/usr/bin/env python3
"""Example of aiohttp.web.Application.on_startup signal handler"""
import asyncio
import aioredis
from aiohttp.web import Application, WebSocketResponse, run_app
async def websocket_handler(request):
ws = WebSocketResponse()
await ws.prepare(request)
request.app['websockets'].append(ws)
try:
async for msg in ws:
print(msg)
await asyncio.sleep(1)
finally:
request.app['websockets'].remove(ws)
return ws
async def on_shutdown(app):
for ws in app['websockets']:
await ws.close(code=999, message='Server shutdown')
async def listen_to_redis(app):
try:
sub = await aioredis.create_redis(('localhost', 6379), loop=app.loop)
ch, *_ = await sub.subscribe('news')
async for msg in ch.iter(encoding='utf-8'):
# Forward message to all connected websockets:
for ws in app['websockets']:
ws.send_str('{}: {}'.format(ch.name, msg))
print("message in {}: {}".format(ch.name, msg))
except asyncio.CancelledError:
pass
finally:
print('Cancel Redis listener: close connection...')
await sub.unsubscribe(ch.name)
await sub.quit()
print('Redis connection closed.')
async def start_background_tasks(app):
app['redis_listener'] = app.loop.create_task(listen_to_redis(app))
async def cleanup_background_tasks(app):
print('cleanup background tasks...')
app['redis_listener'].cancel()
await app['redis_listener']
async def init(loop):
app = Application(loop=loop)
app['websockets'] = []
app.router.add_get('/news', websocket_handler)
app.on_startup.append(start_background_tasks)
app.on_cleanup.append(cleanup_background_tasks)
app.on_shutdown.append(on_shutdown)
return app
loop = asyncio.get_event_loop()
app = loop.run_until_complete(init(loop))
run_app(app)