Publish / Subscribe

redis-py包含一个PubSub对象,来订阅频道和监听消息,创建PubSub对象很简单

>>> r = redis.StrictRedis(...) >>> p = r.pubsub()

一旦一个PubSub对象被创建,频道channel和匹配模式(基于正则表达式的channel)就能够订阅了

>>> p.subscribe('my-first-channel', 'my-second-channel', ...) >>> p.psubscribe('my-*', ...)现在PubSub对象可以订阅这些频道了,可以从PubSub对象读取消息来确认是否订阅成功>>> p.get_message() {'pattern': None, 'type': 'subscribe', 'channel': 'my-second-channel', 'data': 1L} >>> p.get_message() {'pattern': None, 'type': 'subscribe', 'channel': 'my-first-channel', 'data': 2L} >>> p.get_message() {'pattern': None, 'type': 'psubscribe', 'channel': 'my-*', 'data': 3L}

从PubSub对象读取的消息时一个包含一下键的字典

type:可以是以下值中的一个

 ‘subscribe’, ‘unsubscribe’, ‘psubscribe’, ‘punsubscribe’, ‘message’, ‘pmessage’

channel:订阅或取消订阅的频道或者消息要发送的频道

pattern: 匹配一个信息频道的模式,除了'pmessage'其他情况下都是none

data:消息数据,对于(非)订阅消息,这个值会是当前订阅的channel和匹配模式连接的数量,对于[p]message,这个值就是发送的消息

现在就可以发送消息了

 发送方法返回channel匹配和模型pattern匹配的数量# 'my-first-channel' 匹配 'my-first-channel' channel订阅和'my-*' pattern订阅#所以这些消息会被传送给2个channel或pattern>>> r.publish('my-first-channel', 'some data') 2 >>> p.get_message() {'channel': 'my-first-channel', 'data': 'some data', 'pattern': None, 'type': 'message'} >>> p.get_message() {'channel': 'my-first-channel', 'data': 'some data', 'pattern': 'my-*', 'type': 'pmessage'}

对于取消订阅,和订阅一样,如果没有传递参数,会取消所有订阅

>>> p.unsubscribe()>>> p.punsubscribe('my-*')>>> p.get_message(){'channel': 'my-second-channel', 'data': 2L, 'pattern': None, 'type': 'unsubscribe'}>>> p.get_message(){'channel': 'my-first-channel', 'data': 1L, 'pattern': None, 'type': 'unsubscribe'}>>> p.get_message(){'channel': 'my-*', 'data': 0L, 'pattern': None, 'type': 'punsubscribe'}

redis-py 也允许你注册一个回调功能来控制消息发布.消息控制器只有一个参数,message,就像上面例子一样是一个字典.用消息控制器订阅频道channel或者匹配样式pattern,传送channel或pattern作为关键字参数,值作为回调功能

当使用消息控制器从channel或pattern读取消息时,消息字典被创建并传递给消息控制器.这种情况下,由于消息已经被处理,get_message()返回一个None值

>>> def my_handler(message):...     print 'MY HANDLER: ', message['data']>>> p.subscribe(**{'my-channel': my_handler})# 读取订阅确认信息>>> p.get_message(){'pattern': None, 'type': 'subscribe', 'channel': 'my-channel', 'data': 1L}>>> r.publish('my-channel', 'awesome data')1#由于消息控制器的作用,我们需要告诉实例读取数据,可以有多种方式处理,#这里我们只使用get_message()>>> message = p.get_message()MY HANDLER:  awesome data#注意这里my_handler回调打印了上面的字符串 # `message`是 None 因为消息被控制器控制了>>> print messageNone

如果你的应用不关心订阅/取消订阅确认消息(有时候是噪音),你可以传一个 ignore_subscribe_messages=True给 r.pubsub().这会引起所有的订阅非订阅消息读取,但不会出现在你的应用中

>>> p = r.pubsub(ignore_subscribe_messages=True)>>> p.subscribe('my-channel')>>> p.get_message()  #隐藏了订阅消息,返回None >>> r.publish('my-channel') 1 >>> p.get_message() {'channel': 'my-channel', data': 'my data', 'pattern': None, 'type': 'message'}

有三种不同的读取消息的策略

上面的例子使用pubsub.get_message().在这种场景,get_message()使用系统的'select'模式快速测试连接的socket.如果有数据可以被读取,get_message()会读取它,处理后返回或者传递给消息处理器.如果没有数据读取,get_message()会立刻返回None.这使得整合到你的应用中一个已存的事件循环并不重要

>>> while True:>>>     message = p.get_message()>>>     if message:>>>        # do something with the message>>>     time.sleep(0.001)  # be nice to the system :)

redis-py更老的版本只能用 pubsub.listen()读取消息,listen()是一个生成器,会阻塞直到有消息可以获得.如果你的应用不需要做任何事除了从redis接收消息,并对消息做出反应,listen()是一个简单的运行方式

>>> for message in p.listen():...     # do something with the message

第三种选择是在单独的线程里运行一个事件循环, pubsub.run_in_thread() 创建一个新的线程并启动事件循环.线程对象被返回给调用者run_in_thread().调用者可以使用 thread.stop() 来关闭事件循环和线程.在这种场景下,运行线程的只是一个简单的对get_message()的包装器,尤其是你创建一个小的非阻塞的事件循环. run_in_thread() 有一个可选择的 sleep_time参数.如果被指定,事件循环会在每次循环迭代时用指定的值调用time.sleep()

注意,由于我们运行了一个单独的线程,没有办法控制不是由注册的消息控制器自动控制的消息.因此,如果你正在订阅没有消息控制器关联的pattern或channel,redis-p会阻止你调用 run_in_thread()

>>> p.subscribe(**{'my-channel': my_handler})>>> thread = p.run_in_thread(sleep_time=0.001)# 现在事件循环在后台运行处理消息# 当要关闭该线程时>>> thread.stop()

一个PubSub对象绑定到同样编码的语义作为它创建的客户端实例.任何采用unicode的pattern和channel在发给Redis之前会被编码为指定的字符集.如果客户端的解码flag decode_responses被设定为False(默认值),消息字典中的 ‘channel’, ‘pattern’ 和 ‘data’会变成byte字符串((Python 2时str,  Python 3时byte).如果客户端decode_responses 是True,‘channel’, ‘pattern’ 和 ‘data’值会使用客户端的字符集自动解码为unicode字符

PubSub对象保存了他们订阅的channel和pattern.在无法连接的事件中,如网络错误或超时,当重新连接时PubSub对象会重新订阅所有的先前的channel和pattern.无法连接期间发布的消息无法再呈现.当你要结束一个PubSub对象时,调用close()方法关闭连接

>>> p = r.pubsub()>>> ...>>> p.close()

Redis官方文档