python - Redis pub/sub 在订阅中添加额外的 channel

Question

是否可以向 Redis 连接添加额外的订阅？我有一个监听线程，但它似乎不受新 SUBSCRIBE 命令的影响。

如果这是预期的行为，那么如果用户将股票代码提要添加到他们的兴趣或加入聊天室，应该使用什么模式？

我想实现一个类似于以下的 Python 类:

import threading
import redis

class RedisPubSub(object):
    def __init__(self):
        self._redis_pub = redis.Redis(host='localhost', port=6379, db=0)        
        self._redis_sub = redis.Redis(host='localhost', port=6379, db=0)        
        self._sub_thread = threading.Thread(target=self._listen)
        self._sub_thread.setDaemon(True)
        self._sub_thread.start()

    def publish(self, channel, message):
        self._redis_pub.publish(channel, message)

    def subscribe(self, channel):
        self._redis_sub.subscribe(channel)

    def _listen(self):
        for message in self._redis_sub.listen():
            print message

Answer 1

python-redis Redis 和 ConnectionPool 类继承自 threading.local，这会产生您所看到的“神奇”效果。

总结:您的主线程和工作线程的 self._redis_sub 客户端最终使用两个不同的服务器连接，但只有主线程的连接发出了 SUBSCRIBE命令。

详细信息:由于主线程正在创建 self._redis_sub，因此该客户端最终被放入主线程的本地线程存储中。接下来，我假设主线程执行一个 client.subscribe(channel) 调用。现在主线程的客户端在连接 1 上订阅。接下来启动 self._sub_thread 工作线程，它最终将自己的 self._redis_sub 属性设置为 redis.Client 的新实例，它构造一个新的连接池并建立到redis服务器的新连接。

此新连接尚未订阅您的 channel ，因此 listen() 立即返回。因此，对于 python-redis，您无法在线程之间传递具有未完成订阅(或任何其他有状态命令)的已建立连接。

根据您计划如何实现您的应用程序，您可能需要切换到使用不同的客户端，或者想出一些其他方式来将订阅状态传达给工作线程，例如通过队列发送订阅命令。

另一个问题是 python-redis 使用阻塞套接字，这会阻止您的监听线程在等待消息时做其他工作，并且它无法发出希望取消订阅的信号，除非它在之后立即这样做收到消息。