Publish / Subscribe

redis-py包含一個(gè)PubSub對(duì)象,來訂閱頻道和監(jiān)聽消息,創(chuàng)建PubSub對(duì)象很簡(jiǎn)單

>>>?r?=?redis.StrictRedis(...)

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

一旦一個(gè)PubSub對(duì)象被創(chuàng)建,頻道channel和匹配模式(基于正則表達(dá)式的channel)就能夠訂閱了

>>>?p.subscribe('my-first-channel',?'my-second-channel',?...)

>>>?p.psubscribe('my-*',?...)

現(xiàn)在PubSub對(duì)象可以訂閱這些頻道了,可以從PubSub對(duì)象讀取消息來確認(rèn)是否訂閱成功

>>>?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對(duì)象讀取的消息時(shí)一個(gè)包含一下鍵的字典

type:可以是以下值中的一個(gè)

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

channel:訂閱或取消訂閱的頻道或者消息要發(fā)送的頻道

pattern: 匹配一個(gè)信息頻道的模式,除了'pmessage'其他情況下都是none

data:消息數(shù)據(jù),對(duì)于(非)訂閱消息,這個(gè)值會(huì)是當(dāng)前訂閱的channel和匹配模式連接的數(shù)量,對(duì)于[p]message,這個(gè)值就是發(fā)送的消息

現(xiàn)在就可以發(fā)送消息了

發(fā)送方法返回channel匹配和模型pattern匹配的數(shù)量

#?'my-first-channel'?匹配?'my-first-channel'?channel訂閱和'my-*'?pattern訂閱

#所以這些消息會(huì)被傳送給2個(gè)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'}

對(duì)于取消訂閱,和訂閱一樣,如果沒有傳遞參數(shù),會(huì)取消所有訂閱

>>>?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 也允許你注冊(cè)一個(gè)回調(diào)功能來控制消息發(fā)布.消息控制器只有一個(gè)參數(shù),message,就像上面例子一樣是一個(gè)字典.用消息控制器訂閱頻道channel或者匹配樣式pattern,傳送channel或pattern作為關(guān)鍵字參數(shù),值作為回調(diào)功能

當(dāng)使用消息控制器從channel或pattern讀取消息時(shí),消息字典被創(chuàng)建并傳遞給消息控制器.這種情況下,由于消息已經(jīng)被處理,get_message()返回一個(gè)None值

>>>?def?my_handler(message):

...?????print?'MY?HANDLER:?',?message['data']

>>>?p.subscribe(**{'my-channel':?my_handler})

#?讀取訂閱確認(rèn)信息

>>>?p.get_message()

{'pattern':?None,?'type':?'subscribe',?'channel':?'my-channel',?'data':?1L}

>>>?r.publish('my-channel',?'awesome?data')

1

#由于消息控制器的作用,我們需要告訴實(shí)例讀取數(shù)據(jù),可以有多種方式處理,

#這里我們只使用get_message()

>>>?message?=?p.get_message()

MY?HANDLER:??awesome?data

#注意這里my_handler回調(diào)打印了上面的字符串

#?`message`是?None?因?yàn)橄⒈豢刂破骺刂屏?/p>

>>>?print?message

None

如果你的應(yīng)用不關(guān)心訂閱/取消訂閱確認(rèn)消息(有時(shí)候是噪音),你可以傳一個(gè)ignore_subscribe_messages=True給r.pubsub().這會(huì)引起所有的訂閱非訂閱消息讀取,但不會(huì)出現(xiàn)在你的應(yīng)用中

>>>?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().在這種場(chǎng)景,get_message()使用系統(tǒng)的'select'模式快速測(cè)試連接的socket.如果有數(shù)據(jù)可以被讀取,get_message()會(huì)讀取它,處理后返回或者傳遞給消息處理器.如果沒有數(shù)據(jù)讀取,get_message()會(huì)立刻返回None.這使得整合到你的應(yīng)用中一個(gè)已存的事件循環(huán)并不重要

>>>?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()是一個(gè)生成器,會(huì)阻塞直到有消息可以獲得.如果你的應(yīng)用不需要做任何事除了從redis接收消息,并對(duì)消息做出反應(yīng),listen()是一個(gè)簡(jiǎn)單的運(yùn)行方式

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

第三種選擇是在單獨(dú)的線程里運(yùn)行一個(gè)事件循環(huán),pubsub.run_in_thread()創(chuàng)建一個(gè)新的線程并啟動(dòng)事件循環(huán).線程對(duì)象被返回給調(diào)用者run_in_thread().調(diào)用者可以使用thread.stop()來關(guān)閉事件循環(huán)和線程.在這種場(chǎng)景下,運(yùn)行線程的只是一個(gè)簡(jiǎn)單的對(duì)get_message()的包裝器,尤其是你創(chuàng)建一個(gè)小的非阻塞的事件循環(huán).run_in_thread()有一個(gè)可選擇的sleep_time參數(shù).如果被指定,事件循環(huán)會(huì)在每次循環(huán)迭代時(shí)用指定的值調(diào)用time.sleep()

注意,由于我們運(yùn)行了一個(gè)單獨(dú)的線程,沒有辦法控制不是由注冊(cè)的消息控制器自動(dòng)控制的消息.因此,如果你正在訂閱沒有消息控制器關(guān)聯(lián)的pattern或channel,redis-p會(huì)阻止你調(diào)用run_in_thread()

>>>?p.subscribe(**{'my-channel':?my_handler})

>>>?thread?=?p.run_in_thread(sleep_time=0.001)

#?現(xiàn)在事件循環(huán)在后臺(tái)運(yùn)行處理消息

#?當(dāng)要關(guān)閉該線程時(shí)

>>>?thread.stop()

一個(gè)PubSub對(duì)象綁定到同樣編碼的語(yǔ)義作為它創(chuàng)建的客戶端實(shí)例.任何采用unicode的pattern和channel在發(fā)給Redis之前會(huì)被編碼為指定的字符集.如果客戶端的解碼flagdecode_responses被設(shè)定為False(默認(rèn)值),消息字典中的‘channel’, ‘pattern’ 和 ‘data’會(huì)變成byte字符串((Python 2時(shí)str, ?Python 3時(shí)byte).如果客戶端decode_responses是True,‘channel’, ‘pattern’ 和 ‘data’值會(huì)使用客戶端的字符集自動(dòng)解碼為unicode字符

PubSub對(duì)象保存了他們訂閱的channel和pattern.在無法連接的事件中,如網(wǎng)絡(luò)錯(cuò)誤或超時(shí),當(dāng)重新連接時(shí)PubSub對(duì)象會(huì)重新訂閱所有的先前的channel和pattern.無法連接期間發(fā)布的消息無法再呈現(xiàn).當(dāng)你要結(jié)束一個(gè)PubSub對(duì)象時(shí),調(diào)用close()方法關(guān)閉連接

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

>>>?...

>>>?p.close()

總結(jié)

以上是生活随笔為你收集整理的python redis模块_python redis 模块 官方文档(中)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。