国产探花免费观看_亚洲丰满少妇自慰呻吟_97日韩有码在线_资源在线日韩欧美_一区二区精品毛片,辰东完美世界有声小说,欢乐颂第一季,yy玄幻小说排行榜完本

首頁 > 編程 > Python > 正文

Python操作RabbitMQ服務(wù)器實(shí)現(xiàn)消息隊(duì)列的路由功能

2019-11-25 16:41:01
字體:
供稿:網(wǎng)友

Python使用Pika庫(安裝:sudo pip install pika)可以操作RabbitMQ消息隊(duì)列服務(wù)器(安裝:sudo apt-get install rabbitmq-server),這里我們來看一下MQ相關(guān)的路由功能。

路由鍵的實(shí)現(xiàn)

比如有一個(gè)需要給所有接收端發(fā)送消息的場景,但是如果需要自由定制,有的消息發(fā)給其中一些接收端,有些消息發(fā)送給另外一些接收端,要怎么辦呢?這種情況下就要用到路由鍵了。

路由鍵的工作原理:每個(gè)接收端的消息隊(duì)列在綁定交換機(jī)的時(shí)候,可以設(shè)定相應(yīng)的路由鍵。發(fā)送端通過交換機(jī)發(fā)送信息時(shí),可以指明路由鍵 ,交換機(jī)會(huì)根據(jù)路由鍵把消息發(fā)送到相應(yīng)的消息隊(duì)列,這樣接收端就能接收到消息了。

這邊繼上一篇,還是用send.py和receive.py來模擬實(shí)現(xiàn)路由鍵的功能。send.py表示發(fā)送端,receive.py表示接收端。實(shí)例的功能就是將info、warning、error三種級別的信息發(fā)送到不同的接收端。

send.py代碼分析

#!/usr/bin/env python#coding=utf8import pika connection = pika.BlockingConnection(pika.ConnectionParameters(        'localhost'))channel = connection.channel() #定義交換機(jī),設(shè)置類型為directchannel.exchange_declare(exchange='messages', type='direct') #定義三個(gè)路由鍵routings = ['info', 'warning', 'error'] #將消息依次發(fā)送到交換機(jī),并設(shè)置路由鍵for routing in routings:  message = '%s message.' % routing  channel.basic_publish(exchange='messages',             routing_key=routing,             body=message)  print message connection.close()

receive.py代碼分析

#!/usr/bin/env python#coding=utf8import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(        'localhost'))channel = connection.channel() #定義交換機(jī),設(shè)置類型為directchannel.exchange_declare(exchange='messages', type='direct') #從命令行獲取路由鍵參數(shù),如果沒有,則設(shè)置為inforoutings = sys.argv[1:]if not routings:  routings = ['info'] #生成臨時(shí)隊(duì)列,并綁定到交換機(jī)上,設(shè)置路由鍵result = channel.queue_declare(exclusive=True)queue_name = result.method.queuefor routing in routings:  channel.queue_bind(exchange='messages',            queue=queue_name,            routing_key=routing) def callback(ch, method, properties, body):  print " [x] Received %r" % (body,) channel.basic_consume(callback, queue=queue_name, no_ack=True) print ' [*] Waiting for messages. To exit press CTRL+C'channel.start_consuming()

打開兩個(gè)終端,一個(gè)運(yùn)行代碼python receive.py info warning,表示只接收info和warning的消息。另外一個(gè)終端運(yùn)行send.py,可以觀察到接收終端只接收到了info和warning的消息。如果打開多個(gè)終端運(yùn)行receive.py,并傳入不同的路由鍵參數(shù),可以看到更明顯的效果。

當(dāng)接收端正在運(yùn)行時(shí),可以使用rabbitmqctl list_bindings來查看綁定情況。

路由鍵模糊匹配
路由鍵模糊匹配,就是可以使用正則表達(dá)式,和常用的正則表示式不同,這里的話“#”表示所有、全部的意思;“*”只匹配到一個(gè)詞。看完示例就能明白了。

這邊繼上面的例子,還是用send.py和receive.py來實(shí)現(xiàn)路由鍵模糊匹配的功能。send.py表示發(fā)送端,receive.py表示接收端。實(shí)例的功能大概是這樣:比如你有個(gè)知心好朋友,不管開心、傷心、工作上的還是生活上的事情都可以和她說;還有一些朋友可以分享開心的事情;還有一些朋友,你可以把不開心的事情和她說。

send.py代碼分析

因?yàn)橐M(jìn)行路由鍵模糊匹配,所以交換機(jī)的類型要設(shè)置為topic,設(shè)置為topic,就可以使用#,*的匹配符號(hào)了。

#!/usr/bin/env python#coding=utf8import pika connection = pika.BlockingConnection(pika.ConnectionParameters(        'localhost'))channel = connection.channel() #定義交換機(jī),設(shè)置類型為topicchannel.exchange_declare(exchange='messages', type='topic') #定義路由鍵routings = ['happy.work', 'happy.life', 'sad.work', 'sad.life'] #將消息依次發(fā)送到交換機(jī),并設(shè)定路由鍵for routing in routings:  message = '%s message.' % routing  channel.basic_publish(exchange='messages',             routing_key=routing,             body=message)  print message connection.close()

上例中定義了四種類型的消息,容易理解,就不解釋了,然后依次發(fā)送出去。

receive.py代碼分析

同樣,交換機(jī)的類型要設(shè)定為topic就可以了。從命令行接收參數(shù)的功能稍微調(diào)整了一下,就是沒有參數(shù)時(shí)報(bào)錯(cuò)退出。

#!/usr/bin/env python#coding=utf8import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(        'localhost'))channel = connection.channel() #定義交換機(jī),設(shè)置類型為topicchannel.exchange_declare(exchange='messages', type='topic') #從命令行獲取路由參數(shù),如果沒有,則報(bào)錯(cuò)退出routings = sys.argv[1:]if not routings:  print >> sys.stderr, "Usage: %s [routing_key]..." % (sys.argv[0],)  exit() #生成臨時(shí)隊(duì)列,并綁定到交換機(jī)上,設(shè)置路由鍵result = channel.queue_declare(exclusive=True)queue_name = result.method.queuefor routing in routings:  channel.queue_bind(exchange='messages',            queue=queue_name,            routing_key=routing) def callback(ch, method, properties, body):  print " [x] Received %r" % (body,) channel.basic_consume(callback, queue=queue_name, no_ack=True) print ' [*] Waiting for messages. To exit press CTRL+C'channel.start_consuming()

打開四個(gè)終端,一個(gè)運(yùn)行如下,表示任何事情都可以和她說:

python receive.py "#"

另外一個(gè)終端 運(yùn)行如下,表示可以和她分享開心的事:

python receive.py "happy.*"

第三個(gè)運(yùn)行如下,表示工作上的事情可以和她分享:

python receive.py "*.work"

最后一個(gè)運(yùn)行python send.py。結(jié)果不難想象出來,就不貼出來了。

發(fā)表評論 共有條評論
用戶名: 密碼:
驗(yàn)證碼: 匿名發(fā)表
主站蜘蛛池模板: 藁城市| 汕头市| 农安县| 揭东县| 天长市| 城口县| 大同市| 元江| 阳曲县| 庆元县| 永安市| 鄂尔多斯市| 浑源县| 阿勒泰市| 沙田区| 沙田区| 龙川县| 永修县| 朔州市| 同心县| 公安县| 图木舒克市| 紫阳县| 清水河县| 丹寨县| 孝义市| 万年县| 新田县| 日土县| 合作市| 教育| 拜城县| 祁东县| 丰都县| 蒙山县| 肥城市| 开原市| 固始县| 商都县| 隆安县| 墨脱县|