簡易版的http websocket

from aiohttp import web
import asyncio


async def index(req):
    return web.json_response(data={"retcode": 0, "stderr":"index"}, status=200)

async def wshandle(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.WSMsgType.text:
            await ws.send_str("Hello, {}".format(msg.data))
        elif msg.type == web.WSMsgType.binary:
            await ws.send_bytes(msg.data)
        elif msg.type == web.WSMsgType.close:
            break

    return ws

async def init(loop_event):
    app = web.Application(loop=loop_event)
    app.router.add_route('GET', '/', index)
    server = await loop.create_server(app.make_handler(), '127.0.0.1', 13000)  # 創(chuàng)建Http服務
    print('Server started at http://127.0.0.1:13000...')
    return server


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(init(loop))
    loop.run_forever()

image.png

復雜點，加上token認證，隊列.....

不過對于提供 簡單http服務器，相對于django , flask , sanic 等來說，這是比較精簡且高性能的。