关于 Redis的订阅发布

163次阅读

没有评论

共计 2843 个字符，预计需要花费 8 分钟才能阅读完成。

文章目录

1. 为什么做订阅分布?
2. Redis 中的订阅发布
3. Redis 生产者消费者
4. Redis 中订阅发布
5. Java Jedis 踩过的坑

为什么做订阅分布?

随着业务复杂, 业务的项目依赖关系增强, 使用消息队列帮助系统 降低耦合度.

订阅分布本身也是一种生产者消费者模式, 订阅者是消费者, 发布者是生产者.
订阅发布模式, 发布者发布消息后, 只要有订阅方, 则多个订阅方会收到同样的消息
生产者消费者模式, 生产者往队列里放入消息, 由多个消费者对一条消息进行抢占.
订阅分布模式可以将一些不着急完成的工作放到其他进程或者线程中进行离线处理.

Redis 中的订阅发布

Redis 中的订阅发布模式, 当没有订阅者时, 消息会被直接丢弃(Redis 不会持久化保存消息)

Redis 生产者消费者

生产者使用 Redis 中的 list 数据结构进行实现, 将待处理的消息塞入到消息队列中.

class Producer(object):
 
def __init__(self, host=“localhost”, port=6379):
self._conn = redis.StrictRedis(host=host, port=port)
self.key = “test_key”
self.value = “test_value_{id}”
 
def produce(self):
for id in xrange(5):
msg = self.value.format(id=id)
self._conn.lpush(self.key, msg)

消费者使用 redis 中 brpop 进行实现, brpop 会从 list 头部消息, 并能够设置超时等待时间.

class Consumer(object):
 
def __init__(self, host=“localhost”, port=6379):
self._conn = redis.StrictRedis(host=host, port=port)
self.key = “test_key”
 
def consume(self, timeout=0):
# timeout=0 表示会无线阻塞, 直到获得消息
while True:
msg = self._conn.brpop(self.key, timeout=timeout)
process(msg)
 
 
def process(msg):
print msg
 
if __name__ == ‘__main__’:
consumer = Consumer()
consumer.consume()
# 输出结果
(‘test_key’, ‘test_value_1’)
(‘test_key’, ‘test_value_2’)
(‘test_key’, ‘test_value_3’)
(‘test_key’, ‘test_value_4’)
(‘test_key’, ‘test_value_5’)

Redis 中订阅发布

在 Redis Pubsub 中, 一个频道 (channel) 相当于一个消息队列

class Publisher(object):
 
def __init__(self, host, port):
self._conn = redis.StrictRedis(host=host, port=port)
self.channel = “test_channel”
self.value = “test_value_{id}”
 
def pub(self):
for id in xrange(5):
msg = self.value.format(id=id)
self._conn.publish(self.channel, msg)

其中 get_message 使用了select IO 多路复用来检查 socket 连接是否是否可读.

class Subscriber(object):
 
def __init__(self, host=“localhost”, port=6379):
self._conn = redis.StrictRedis(host=host, port=port)
self._pubsub = self._conn.pubsub() # 生成 pubsub 对象
self.channel = “test_channel”
self._pubsub.subscribe(self.channel)
 
def sub(self):
while True:
msg = self._pubsub.get_message()
if msg and isinstance(msg.get(“data”), basestring):
process(msg.get(“data”))
 
def close(self):
self._pubsub.close()
 
# 输出结果
test_value_1
test_value_2
test_value_3
test_value_4
test_value_5