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

时间：2023-03-09 08:54:36
python开发-实现redis中的发布订阅功能

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可订阅多个频道（这样写的时候，作为参数的频道应该是一个列表）。之后就可以开始监听了。

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

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()返回的结果进行了处理，使结果更加简单。

相关文章