python - aiohttp应用进程中监听ZeroMQ

Question

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

我的 aiohttp 应用程序有一组已连接的 WebSocket(移动应用程序客户端)，我希望在由 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 应用程序的生命周期内正确运行和关闭它？

---

更新

我已经创建了一个 issue在 aiohttp 的 GitHub 存储库中描述问题并提出可能的解决方案。我非常感谢在这里或那里就所描述的问题提供意见。

Answer 1

好的，关于这个的问题和讨论 issue带来了我为 aiohttp 贡献的新功能，即在 1.0 版本中，我们将能够使用以下方式注册 on_startup 应用程序信号Application.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)