Asyncio+aiohttp-redis Pub/Sub和websocket单手读/写

Question

我目前正在使用aiohttp来查看它作为带有websocket连接的移动应用程序的服务器应用程序将如何运行。

下面是一个简单的“Hello world”示例（as gist here）：

import asyncio
import aiohttp
from aiohttp import web


class WebsocketEchoHandler:

    @asyncio.coroutine
    def __call__(self, request):
        ws = web.WebSocketResponse()
        ws.start(request)

        print('Connection opened')
        try:
            while True:
                msg = yield from ws.receive()
                ws.send_str(msg.data + '/answer')
        except:
            pass
        finally:
            print('Connection closed')
        return ws


if __name__ == "__main__":
    app = aiohttp.web.Application()
    app.router.add_route('GET', '/ws', WebsocketEchoHandler())

    loop = asyncio.get_event_loop()
    handler = app.make_handler()

    f = loop.create_server(
        handler,
        '127.0.0.1',
        8080,
    )

    srv = loop.run_until_complete(f)
    print("Server started at {sock[0]}:{sock[1]}".format(
        sock=srv.sockets[0].getsockname()
    ))
    try:
        loop.run_forever()
    except keyboardInterrupt:
        pass
    finally:
        loop.run_until_complete(handler.finish_connections(1.0))
        srv.close()
        loop.run_until_complete(srv.wait_closed())
        loop.run_until_complete(app.finish())
    loop.close()

问题

现在我想使用下面描述的结构（node server=python aiohttp）。更具体地说，使用Redis Pub/Sub机制和asyncio-redis来读写my WebsocketEchoHandler中的websocket connection和Redis。

WebsocketEchoHandler是一个死的简单循环，所以我不确定该如何完成。使用Tornado和brükva我只需要使用回调。

额外（可能是非主题）问题

因为我已经在使用Redis，所以我应该采取两种方法中的哪一种：

1. 就像在“经典”web应用程序中一样，对所有内容都有一个控制器/视图，只对消息传递使用Redis
2. Web应用程序应该只是客户端和任务队列（最简单的Python RQ）之间的一层。每个请求都应该委托给工人。

编辑

来自http://goldfirestudios.com/blog/136/Horizontally-Scaling-Node.js-and-WebSockets-with-Redis的图像

编辑2

看来我需要澄清一下。

• 仅限Websocket处理程序如上图所示

• Redis发布/子处理程序可能是这样的：

  class WebsocketEchoHandler:
  
      @asyncio.coroutine
      def __call__(self, request):
          ws = web.WebSocketResponse()
          ws.start(request)
  
          connection = yield from asyncio_redis.Connection.create(host='127.0.0.1', port=6379)
          subscriber = yield from connection.start_subscribe()
          yield from subscriber.subscribe(['ch1', 'ch2'])
  
          print('Connection opened')
          try:
              while True:
                  msg = yield from subscriber.next_published()
                  ws.send_str(msg.value + '/answer')
          except:
              pass
          finally:
              print('Connection closed')
          return ws
  

  这个处理程序只订阅Redis channelch1和ch2并将从这些channel接收到的所有消息发送到websocket。

• 我想要这个处理程序：

  class WebsocketEchoHandler:
  
      @asyncio.coroutine
      def __call__(self, request):
          ws = web.WebSocketResponse()
          ws.start(request)
  
          connection = yield from asyncio_redis.Connection.create(host='127.0.0.1', port=6379)
          subscriber = yield from connection.start_subscribe()
          yield from subscriber.subscribe(['ch1', 'ch2'])
  
          print('Connection opened')
          try:
              while True:
                  # If message recived from redis OR from websocket
                  msg_ws = yield from ws.receive()
                  msg_redis = yield from subscriber.next_published()
                  if msg_ws:
                      # push to redis / do something else
                      self.on_msg_from_ws(msg_ws)
                  if msg_redis:
                      self.on_msg_from_redis(msg_redis)
          except:
              pass
          finally:
              print('Connection closed')
          return ws
  

  但是以下代码总是按顺序调用的，因此从websocket读取会阻止从Redis读取：

  msg_ws = yield from ws.receive()
  msg_redis = yield from subscriber.next_published()
  

我希望在event上完成读取，其中event是从两个源之一接收的消息。

Answer 1

最近我们可以在Python3.5及更高版本中使用async await。。

async def task1(ws):
    async for msg in ws:
        if msg.type == WSMsgType.TEXT:
            data = msg.data
            print(data)
            if data:
                await ws.send_str('pong')
## ch is a redis channel
async def task2(ch):
    async for msg in ch1.iter(encoding="utf-8", decoder=json.loads):
        print("receving", msg)
        user_token = msg['token']
        if user_token in r_cons.keys():
            _ws = r_cons[user_token]
            await  _ws.send_json(msg)

coroutines = list()
coroutines.append(task1(ws))
coroutines.append(task2(ch1))

await asyncio.gather(*coroutines)

这就是我所做的。当websockets需要等待来自多个源的消息时。

这里的要点是使用asyncio.gather将两个corotine一起运行，就像 @达诺提到。