python中使用redis发布订阅者模型

Posted 2020-11-27 小学弟-

python中使用redis发布订阅者模型

 

redis发布订阅者模型：

　　Redis提供了发布订阅功能，可以用于消息的传输，Redis的发布订阅机制包括三个部分，发布者，订阅者和Channel。发布者和订阅者都是Redis客户端，Channel则为Redis服务器端，发布者将消息发送到某个的频道，订阅了这个频道的订阅者就能接收到这条消息。Redis的这种发布订阅机制与基于主题的发布订阅类似，Channel相当于主题。

  　　　　　　　　　　

发布者：

　　pub.py

import redis

conn = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)


conn.publish("333", "18")

订阅者：

　　sub.py

import redis

conn = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

# 第一步 生成一个订阅者对象
pubsub = conn.pubsub()

# 第二步 订阅一个消息

pubsub.subscribe("gaoxin333")

# 创建一个接收

while True:
    print("working~~~")
    msg = pubsub.parse_response()
    print(msg)

 

（1）发送消息 
Redis采用PUBLISH命令发送消息，其返回值为接收到该消息的订阅者的数量。

（2）订阅某个频道 

 

Redis采用SUBSCRIBE命令订阅某个频道，其返回值包括客户端订阅的频道，目前已订阅的频道数量，以及接收到的消息，其中subscribe表示已经成功订阅了某个频道。

（3）模式匹配 

 

模式匹配功能允许客户端订阅符合某个模式的频道，Redis采用PSUBSCRIBE订阅符合某个模式所有频道，用“”表示模式，“”可以被任意值代替。假设客户端同时订阅了某种模式和符合该模式的某个频道，那么发送给这个频道的消息将被客户端接收到两次，只不过这两条消息的类型不同，一个是message类型，一个是pmessage类型，但其内容相同。 

（4）取消订阅 
Redis采用UNSUBSCRIBE和PUNSUBSCRIBE命令取消订阅，其返回值与订阅类似。 
由于Redis的订阅操作是阻塞式的，因此一旦客户端订阅了某个频道或模式，就将会一直处于订阅状态直到退出。在SUBSCRIBE，PSUBSCRIBE，UNSUBSCRIBE和PUNSUBSCRIBE命令中，其返回值都包含了该客户端当前订阅的频道和模式的数量，当这个数量变为0时，该客户端会自动退出订阅状态。

 

import redis
conn = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

conn.publish("gaoxin333", "18")

---------------------------------------------------------------------------------------------------------------------------------------------------------

---------------------------------------------------------------------------------------------------------------------------------------------------------

python开发-实现redis中的发布订阅功能

 

Python3学习（二十七）：python实现Redis的订阅与发布（sub-pub机制）

先介绍一下redis的pub/sub功能：

Pub/Sub功能（means Publish, Subscribe）即发布及订阅功能。基于事件的系统中，Pub/Sub是目前广泛使用的通信模型，它采用事件作为基本的通信机制，提供大规模系统所要求的松散耦合的交互模式：订阅者(如客户端)以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件；发布者(如服务器)可将订阅者感兴趣的事件随时通知相关订阅者。

通俗来讲，就是说我sub端（订阅者）一直监听着，一旦pub端（发布者）发布了消息，那么我就接收过来，举个例子，先是发布者：

 

#coding:utf-8
import time
import redis
 
number_list = [\'300033\', \'300032\', \'300031\', \'300030\']
signal = [\'1\', \'-1\', \'1\', \'-1\']
 
rc = redis.StrictRedis(host=\'***\', port=\'6379\', db=3, password=\'********\')
for i in range(len(number_list)):
    value_new = str(number_list[i]) + \' \' + str(signal[i])
    rc.publish("liao", value_new)  #发布消息到liao

 

 

接着我们来看看订阅者：

 

 

#coding:utf-8
import time
import redis
 
rc = redis.StrictRedis(host=\'****\', port=\'6379\', db=3, password=\'******\')
ps = rc.pubsub()
ps.subscribe(\'liao\')  #从liao订阅消息
for item in ps.listen():        #监听状态：有消息发布了就拿过来
    if item[\'type\'] == \'message\':
        print item[\'channel\']
        print item[\'data\']

 

 

关于数据结构，也就是item，是类似于：{\'pattern\': None, \'type\': \'message\', \'channel\': \'liao\', \'data\': \'300033 1\'}这样的，所以可以通过channel来判断这个消息是属于哪一个队列里的。（运行程序的时候，先运行订阅者，在运行发布者程序）

总结，要点有两个：

• 一是连接方式。使用python连接redis有三种方式：①使用库中的Redis类（或StrictRedis类，其实差不多）；②使用ConnectionPool连接池（可保持长连接）；③使用Sentinel类（如果有多个redis做集群时，程序会自己选择一个合适的连接）。
• 二是订阅方法。这里使用的是StrictRedis类中的pubsub方法。连接好之后，可使用subscribe或psubscribe方法来订阅redis消息。其中subscribe是订阅一个频道，psubscribe可订阅多个频道（这样写的时候，作为参数的频道应该是一个列表）。之后就可以开始监听了。

redis中的发布/订阅模型是一种消息通信模式，今天聊一下在python中实现简单的发布订阅功能。

实现方式一:
redis_helper.py: 封装发布订阅方法

import redis
                
                
class RedisHelper(object):
    
    def __init__(self):
        self.__conn = redis.Redis(host="localhost")
        # 订阅频道
        self.chan_sub = "fm104.5"
        
    def public(self, msg):
        """
        在指定频道上发布消息
        :param msg:
        :return:
        """
        # publish(): 在指定频道上发布消息，返回订阅者的数量
        self.__conn.publish(self.chan_sub, msg)
        return True

    def subscribe(self):
        # 返回发布订阅对象，通过这个对象你能1）订阅频道 2）监听频道中的消息
        pub = self.__conn.pubsub()
        # 订阅频道，与publish()中指定的频道一样。消息会发布到这个频道中
        pub.subscribe(self.chan_sub)
        ret = pub.parse_response()  # [b\'subscribe\', b\'fm86\', 1]
        print("ret:%s" % ret)
        return pub

redis_pub.py: 发布者

from redis_helper import RedisHelper
                
                
obj = RedisHelper()
for i in range(5):
    obj.public("hello_%s" % i)

redis_sub.py: 订阅者

from redis_helper import RedisHelper
                
                
obj = RedisHelper()
redis_sub = obj.subscribe()
while True:
    msg = redis_sub.parse_response()
    print(msg)

实现方式二:
redis_helper.py: 封装发布订阅方法

import redis
                
                
class RedisHelper(object):
    
    def __init__(self):
        self.__conn = redis.Redis(host="localhost")
        # 频道名称
        self.chan_sub = "orders"
        
    def public(self, msg):
        """
        在指定频道上发布消息
        :param msg:
        :return:
        """
        # publish(): 在指定频道上发布消息，返回订阅者的数量
        self.__conn.publish(self.chan_sub, msg)
        return True

    def subscribe(self):
        # 返回发布订阅对象，通过这个对象你能1）订阅频道 2）监听频道中的消息
        pub = self.__conn.pubsub()
        # 订阅某个频道，与publish()中指定的频道一样。消息会发布到这个频道中
        pub.subscribe(self.chan_sub)
        return pub

redis_pub.py:

from redis_helper import RedisHelper


obj = RedisHelper()
for i in range(5):
    obj.public("hello_%s" % i)

redis_sub.py:

from redis_helper import RedisHelper

obj = RedisHelper()
redis_sub = obj.subscribe()
while True:
    # listen()函数封装了parse_response()函数
    msg = redis_sub.listen()
    for i in msg:
        if i["type"] == "message":
            print(str(i["channel"], encoding="utf-8") + ":" + str(i["data"], encoding="utf-8"))
        elif i["type"] == "subscrube":
            print(str(i["chennel"], encoding="utf-8"))

以上两种方式的不同之处在于，方式一使用发布订阅对象的parse_response()方法获取订阅信息，方式二使用发布订阅对象的listen()方法获取订阅信息。listen()方法是对parse_response()方法的封装，加入了阻塞，并将parse_response()返回的结果进行了处理，使结果更加简单。