在aiohttp应用程序中收听ZeroMQ

Question

我运行aiohttp应用程序Gunicorn落后于nginx。 在我的应用程序的初始化模块中，我不运行使用web.run_app(app)的应用程序，但只是创建一个将由Gunicorn导入的实例，以便在每个工作人员Gunicorn创建中运行该实例。 因此，Gunicorn创建了一些工作进程，它们中的事件循环，然后在这些循环中创建应用程序的请求处理程序runs。

我的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描述该问题，并提出可能的解决方案。对于所描述的问题，我会非常感谢在这里或那里提出的意见。

Answer 1

好吧，这个问题并在此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)