使用redis.asyncio订阅时回调函数无法触发的问题排查

我在网站中使用Starlette框架和WebSockets，创建了继承自WebSocketEndpoint的类处理WebSocket连接，同时用另一个类管理Redis的Pub/Sub功能。尝试注册订阅时的回调函数，但始终无法触发该回调。查看subscribe方法文档，说明如下：

Subscribe to channels. Channels supplied as keyword arguments expect
a channel name as the key and a callable as the value. A channel's
callable will be invoked automatically when a message is received on
that channel rather than producing a message via listen() or
get_message().

请问我哪里遗漏了？

WSEndpoint类 #

from starlette.endpoints import WebSocketEndpoint
from starlette.websockets import WebSocket
import json
from blueprints import redis_pubsub_manager

class WSEndpoint(WebSocketEndpoint):
    encoding = 'json'

    def __init__(self, scope, receive, send):
        super().__init__(scope, receive, send)

        self.global_chatroom_channel_name = "globalchat"
        self.connected_users = []
        self.redis_manager_sub = redis_pubsub_manager.RedisPubSubManager(room=self.global_chatroom_channel_name)
        self.redis_manager_pub = redis_pubsub_manager.RedisPubSubManager(room=self.global_chatroom_channel_name)

        self.global_subscription = None

        print("done")

    async def on_connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connected_users.append(websocket)
        if self.global_subscription is None:
            await self.redis_manager_sub.connect()
            await self.redis_manager_pub.connect()
            self.global_subscription = await self.redis_manager_sub.subscribe(callback = self.publish_message_to_subscribers)

        print(f"Socket Connected: {websocket}")


    async def on_receive(self, websocket: WebSocket, data):
        await self.redis_manager_pub._publish(json.dumps(data))

    async def publish_message_to_subscribers(msg):
        print("inside publish_message_to_subscribers")
        print(msg)

RedisPubSubManager类 #

import asyncio
import redis.asyncio as aioredis
import json

class RedisPubSubManager:
    def __init__(self, host='localhost', port=6379, room="globalchat"):
        self.redis_host = host
        self.redis_port = port
        self.pubsub = None
        self.room = room

    async def _get_redis_connection(self) -> aioredis.Redis:
        return aioredis.Redis(host=self.redis_host,
                              port=self.redis_port,
                              auto_close_connection_pool=False)
    
    async def connect(self) -> None:
        self.redis_connection = await self._get_redis_connection()
        self.pubsub = self.redis_connection.pubsub()


    async def _publish(self, message) -> None:
        await self.redis_connection.publish(self.room, message)

    async def subscribe(self, callback) -> aioredis.Redis:
        await self.pubsub.subscribe(**{"globalchat": callback})

        return self.pubsub

问题原因及修复方案 #

1. 缺少PubSub监听循环
   aioredis的pubsub回调不会自动触发，必须启动监听循环来处理消息。你只调用了subscribe，但没有启动run()任务来驱动消息处理流程。

2. 回调方法缺少self参数
   publish_message_to_subscribers是类的成员方法，定义时未传入self参数，会导致调用时抛出参数不匹配的错误（隐性问题）。

3. 重复创建Redis连接实例
   分别创建redis_manager_sub和redis_manager_pub完全没必要，一个Redis连接实例即可同时处理发布和订阅操作，重复创建会浪费资源。

4. 订阅频道硬编码
   subscribe方法里硬写了"globalchat"，应该使用类初始化时的self.room变量，保持代码灵活性。

修复后的代码 #

RedisPubSubManager类 #

import asyncio
import redis.asyncio as aioredis
import json

class RedisPubSubManager:
    def __init__(self, host='localhost', port=6379, room="globalchat"):
        self.redis_host = host
        self.redis_port = port
        self.pubsub = None
        self.room = room
        self.redis_connection = None
        self._listen_task = None

    async def _get_redis_connection(self) -> aioredis.Redis:
        return aioredis.Redis(host=self.redis_host,
                              port=self.redis_port,
                              auto_close_connection_pool=False)
    
    async def connect(self) -> None:
        self.redis_connection = await self._get_redis_connection()
        self.pubsub = self.redis_connection.pubsub()

    async def start_listening(self):
        # 启动异步监听任务，触发回调函数
        self._listen_task = asyncio.create_task(self.pubsub.run())

    async def stop_listening(self):
        # 停止监听并关闭连接
        if self._listen_task:
            self._listen_task.cancel()
            await self._listen_task
        await self.pubsub.close()
        await self.redis_connection.close()

    async def _publish(self, message) -> None:
        await self.redis_connection.publish(self.room, message)

    async def subscribe(self, callback) -> aioredis.Redis:
        # 使用实例化时的频道名
        await self.pubsub.subscribe(**{self.room: callback})
        return self.pubsub

WSEndpoint类 #

from starlette.endpoints import WebSocketEndpoint
from starlette.websockets import WebSocket
import json
from blueprints import redis_pubsub_manager

class WSEndpoint(WebSocketEndpoint):
    encoding = 'json'

    def __init__(self, scope, receive, send):
        super().__init__(scope, receive, send)
        self.global_chatroom_channel_name = "globalchat"
        self.connected_users = []
        # 单个实例同时处理发布和订阅
        self.redis_manager = redis_pubsub_manager.RedisPubSubManager(room=self.global_chatroom_channel_name)
        self.global_subscription = None
        print("done")

    async def on_connect(self, websocket: WebSocket):
        await websocket.accept()
        self.connected_users.append(websocket)
        if self.global_subscription is None:
            await self.redis_manager.connect()
            self.global_subscription = await self.redis_manager.subscribe(callback=self.publish_message_to_subscribers)
            # 启动监听循环
            await self.redis_manager.start_listening()
        print(f"Socket Connected: {websocket}")

    async def on_receive(self, websocket: WebSocket, data):
        await self.redis_manager._publish(json.dumps(data))

    async def on_disconnect(self, websocket: WebSocket, close_code):
        self.connected_users.remove(websocket)
        # 无在线用户时关闭Redis连接
        if not self.connected_users and self.redis_manager._listen_task:
            await self.redis_manager.stop_listening()

    # 添加self参数
    async def publish_message_to_subscribers(self, msg):
        print("inside publish_message_to_subscribers")
        print(msg)
        # 可选：将消息推送给所有WebSocket客户端
        for user in self.connected_users:
            await user.send_json(msg)

内容的提问来源于stack exchange，提问作者Asdfg