aioredis.2.X版本下异步方式的订阅和发布

目前部分的业务涉及到跨服通讯，设计上所以采取了生产者和消费者的模式来进行处理。考虑当前已都存在aioredis的使用，所以直接基于aioredis来设置。但是由于大部分的接口都是异步的方式，所以也需要在aioredis进行异步的使用。于是有下面简单小案例示例。

完整示例代码：

from fastapi import FastAPI,Request
import asyncio
import async_timeout
import aioredis
from aioredis.client import Redis

app = FastAPI()  # pylint: disable=invalid-name
STOPWORD = "STOP"
from pydantic import BaseModel


class MessageEvent(BaseModel):
    username: str
    message: dict


async def reader(channel: aioredis.client.PubSub):
    while True:
        try:
            async with async_timeout.timeout(1):
                # 执行接收订阅消息
                message = await channel.get_message(ignore_subscribe_messages=True)
                if message is not None:
                    pass
                    message_event = MessageEvent.parse_raw(message["data"].decode('utf-8'))
                    print("订阅接收到消息为：",message_event)
                await asyncio.sleep(0.01)
        except asyncio.TimeoutError:
            pass


@app.on_event("startup")
async def startup_event():


    redis:Redis = aioredis.from_url("redis://localhost")
    pubsub = redis.pubsub()
    app.state.redis = redis
    app.state.pubsub = pubsub
    await pubsub.subscribe("channel:1", "channel:2")
    event = MessageEvent(username="asdsd", message={"sdfffff": "asdasd"})
    await redis.publish(channel="channel:1", message=event.json())
    future = asyncio.create_task(reader(pubsub))
    # 不能加这个等待结果返回，因为里面是一个while 循环，等待返回就会一直阻塞
    # await future


@app.on_event("shutdown")
async def shutdown_event():
    pass
    app.state.pubsub.unsubscribe("channel:1", "channel:2")
    app.state.redis.close()

@app.get('/index')
async def get(re:Request):
    event = MessageEvent(username="我是你大爷", message={"今晚吃什么": "吃吃吃！哈哈！就知道吃！"})
    await re.app.state.redis.publish(channel="channel:1", message=event.json())
    return "shouey"


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='run:app', host="127.0.0.1", port=8000, reload=True)

代码解析说明：

• 首先在服务启动事件中启动开始进行redis实例的创建，并且把它放到我们的全局app上下文中。
• 然后开始创建通过使用redis.pubsub()返回发布/订阅对象,使用pubsub才可以订阅频道并收听发布到的消息。
• 使用pubsub对象开始进行消息频道的定义 pubsub.subscribe("channel:1", "channel:2")（消息频道可以用于消息隔离，订阅不同的频道可以获取到不同的消息）
• 测试消息发布示例，通过await redis.publish进行一次消息的发布，并且发布到的频道是channel="channel:1"

  event = MessageEvent(username="asdsd", message={"sdfffff": "asdasd"}) await redis.publish(channel="channel:1", message=event.json())

• 开始启动频道消息的订阅循环 asyncio.create_task(reader(pubsub))，这里需要注意的是，由于我们的这里是集成在框架中，不能使用# await future等待结果返回，否则会无法往下执行。这里只是创建了协程任务，然后会自动丢到事件循环中去执行。

定义测试消息发布接口：

@app.get('/index')
async def get(re:Request):
    event = MessageEvent(username="我是你大爷", message={"今晚吃什么": "吃吃吃！哈哈！就知道吃！"})
    await re.app.state.redis.publish(channel="channel:1", message=event.json())
    return "shouey"

代码说明：

• 上面通过封装MessageEvent消息体信息进行序列化转为.json()格式，然后发布到了redis频道中。然后我们可以通过：reader接收到具体消息频道，然后进行消息的反序列：

async def reader(channel: aioredis.client.PubSub):
    while True:
        try:
            async with async_timeout.timeout(1):
                # 执行接收订阅消息
                message = await channel.get_message(ignore_subscribe_messages=True)
                if message is not None:
                    pass
                    message_event = MessageEvent.parse_raw(message["data"].decode('utf-8'))
                    print("订阅接收到消息为：",message_event)
                await asyncio.sleep(0.01)
        except asyncio.TimeoutError:
            pass

需要注意的是反序化的过程是需要进行解码：

  message_event = MessageEvent.parse_raw(message["data"].decode('utf-8'))

小技巧：

• 消息实体的序列化和反序列化处理 如：

序列化
 event = MessageEvent(username="我是你大爷", message={"今晚吃什么": "吃吃吃！哈哈！就知道吃！"})
event.json()
反序列化
message_event = MessageEvent.parse_raw(message["data"].decode('utf-8'))

然后还有其他序列化库，如：

• dataclasses-json

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822