引入
rabbitmq 是一個(gè)由 erlang 語言開發(fā)的 amqp 的開源實(shí)現(xiàn)。
rabbitmq是一款基于amqp協(xié)議的消息中間件，它能夠在應(yīng)用之間提供可靠的消息傳輸。在易用性，擴(kuò)展性，高可用性上表現(xiàn)優(yōu)秀。使用消息中間件利于應(yīng)用之間的解耦，生產(chǎn)者（客戶端）無需知道消費(fèi)者（服務(wù)端）的存在。而且兩端可以使用不同的語言編寫，大大提供了靈活性。
安裝
|
1
2
3
4
5
6
7
8
9
10
11
|
# 安裝配置epel源 rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm # 安裝erlang yum -y install erlang # 安裝rabbitmq yum -y install rabbitmq-server# 啟動(dòng)/停止 service rabbitmq-server start/stop |
rabbitmq工作模型
簡單模式
生產(chǎn)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import pikaconnection = pika.blockingconnection(pika.connectionparameters( host='localhost'))channel = connection.channel()channel.queue_declare(queue='hello')channel.basic_publish(exchange='', routing_key='hello', body='hello world!')print(" [x] sent 'hello world!'")connection.close() |
消費(fèi)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
connection = pika.blockingconnection(pika.connectionparameters(host='localhost'))channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] received %r" % body) channel.basic_consume( callback, queue='hello', no_ack=true) print(' [*] waiting for messages. to exit press ctrl+c')channel.start_consuming() |
相關(guān)參數(shù)
1，no-ack ＝ false
如果消費(fèi)者遇到情況(its channel is closed, connection is closed, or tcp connection is lost)掛掉了，那么，rabbitmq會(huì)重新將該任務(wù)添加到隊(duì)列中。
- 回調(diào)函數(shù)中的 ch.basic_ack(delivery_tag=method.delivery_tag)
- basic_comsume中的no_ack=false
接收消息端應(yīng)該這么寫：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
import pikaconnection = pika.blockingconnection(pika.connectionparameters( host='10.211.55.4'))channel = connection.channel()channel.queue_declare(queue='hello')def callback(ch, method, properties, body): print(" [x] received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_consume(callback, queue='hello', no_ack=false)print(' [*] waiting for messages. to exit press ctrl+c')channel.start_consuming() |
2，durable ：消息不丟失
生產(chǎn)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
import pikaconnection = pika.blockingconnection(pika.connectionparameters(host='10.211.55.4'))channel = connection.channel()# make message persistentchannel.queue_declare(queue='hello', durable=true)channel.basic_publish(exchange='', routing_key='hello', body='hello world!', properties=pika.basicproperties( delivery_mode=2, # make message persistent ))print(" [x] sent 'hello world!'")connection.close() |
3，消息獲取順序
默認(rèn)消息隊(duì)列里的數(shù)據(jù)是按照順序被消費(fèi)者拿走，例如：消費(fèi)者1 去隊(duì)列中獲取 奇數(shù) 序列的任務(wù)，消費(fèi)者1去隊(duì)列中獲取 偶數(shù) 序列的任務(wù)。
channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數(shù)排列
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
import pikaconnection = pika.blockingconnection(pika.connectionparameters(host='10.211.55.4'))channel = connection.channel()# make message persistentchannel.queue_declare(queue='hello')def callback(ch, method, properties, body): print(" [x] received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_qos(prefetch_count=1)channel.basic_consume(callback, queue='hello', no_ack=false)print(' [*] waiting for messages. to exit press ctrl+c')channel.start_consuming() |
exchange模型
1，發(fā)布訂閱
發(fā)布訂閱和簡單的消息隊(duì)列區(qū)別在于，發(fā)布訂閱會(huì)將消息發(fā)送給所有的訂閱者，而消息隊(duì)列中的數(shù)據(jù)被消費(fèi)一次便消失。所以，rabbitmq實(shí)現(xiàn)發(fā)布和訂閱時(shí)，會(huì)為每一個(gè)訂閱者創(chuàng)建一個(gè)隊(duì)列，而發(fā)布者發(fā)布消息時(shí)，會(huì)將消息放置在所有相關(guān)隊(duì)列中。
exchange type = fanout
生產(chǎn)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
import pikaimport sysconnection = pika.blockingconnection(pika.connectionparameters( host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs', type='fanout')message = ' '.join(sys.argv[1:]) or "info: hello world!"channel.basic_publish(exchange='logs', routing_key='', body=message)print(" [x] sent %r" % message)connection.close() |
消費(fèi)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
import pikaconnection = pika.blockingconnection(pika.connectionparameters( host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='logs', type='fanout')result = channel.queue_declare(exclusive=true)queue_name = result.method.queuechannel.queue_bind(exchange='logs', queue=queue_name)print(' [*] waiting for logs. to exit press ctrl+c')def callback(ch, method, properties, body): print(" [x] %r" % body)channel.basic_consume(callback, queue=queue_name, no_ack=true)channel.start_consuming() |
2，關(guān)鍵字發(fā)送
之前事例，發(fā)送消息時(shí)明確指定某個(gè)隊(duì)列并向其中發(fā)送消息，rabbitmq還支持根據(jù)關(guān)鍵字發(fā)送，即：隊(duì)列綁定關(guān)鍵字，發(fā)送者將數(shù)據(jù)根據(jù)關(guān)鍵字發(fā)送到消息exchange，exchange根據(jù) 關(guān)鍵字 判定應(yīng)該將數(shù)據(jù)發(fā)送至指定隊(duì)列。
exchange type = direct
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
import pikaimport sysconnection = pika.blockingconnection(pika.connectionparameters( host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='direct_logs', type='direct')result = channel.queue_declare(exclusive=true)queue_name = result.method.queueseverities = sys.argv[1:]if not severities: sys.stderr.write("usage: %s [info] [warning] [error]\n" % sys.argv[0]) sys.exit(1)for severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)print(' [*] waiting for logs. to exit press ctrl+c')def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback, queue=queue_name, no_ack=true)channel.start_consuming() |
3，模糊匹配
exchange type = topic
發(fā)送者路由值 隊(duì)列中
old.boy.python old.* -- 不匹配
old.boy.python old.# -- 匹配
在topic類型下，可以讓隊(duì)列綁定幾個(gè)模糊的關(guān)鍵字，之后發(fā)送者將數(shù)據(jù)發(fā)送到exchange，exchange將傳入”路由值“和 ”關(guān)鍵字“進(jìn)行匹配，匹配成功，則將數(shù)據(jù)發(fā)送到指定隊(duì)列。
- # 表示可以匹配 0 個(gè) 或 多個(gè) 單詞
- * 表示只能匹配 一個(gè) 單詞
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
import pikaimport sysconnection = pika.blockingconnection(pika.connectionparameters( host='localhost'))channel = connection.channel()channel.exchange_declare(exchange='topic_logs', type='topic')result = channel.queue_declare(exclusive=true)queue_name = result.method.queuebinding_keys = sys.argv[1:]if not binding_keys: sys.stderr.write("usage: %s [binding_key]...\n" % sys.argv[0]) sys.exit(1)for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)print(' [*] waiting for logs. to exit press ctrl+c')def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body))channel.basic_consume(callback, queue=queue_name, no_ack=true)channel.start_consuming() |
基于rabbitmq的rpc
callback queue 回調(diào)隊(duì)列
一個(gè)客戶端向服務(wù)器發(fā)送請求，服務(wù)器端處理請求后，將其處理結(jié)果保存在一個(gè)存儲(chǔ)體中。而客戶端為了獲得處理結(jié)果，那么客戶在向服務(wù)器發(fā)送請求時(shí)，同時(shí)發(fā)送一個(gè)回調(diào)隊(duì)列地址 reply_to 。
correlation id 關(guān)聯(lián)標(biāo)識(shí)
一個(gè)客戶端可能會(huì)發(fā)送多個(gè)請求給服務(wù)器，當(dāng)服務(wù)器處理完后，客戶端無法辨別在回調(diào)隊(duì)列中的響應(yīng)具體和那個(gè)請求時(shí)對應(yīng)的。為了處理這種情況，客戶端在發(fā)送每個(gè)請求時(shí)，同時(shí)會(huì)附帶一個(gè)獨(dú)有 correlation_id 屬性，這樣客戶端在回調(diào)隊(duì)列中根據(jù) correlation_id 字段的值就可以分辨此響應(yīng)屬于哪個(gè)請求。
客戶端發(fā)送請求：
某個(gè)應(yīng)用將請求信息交給客戶端，然后客戶端發(fā)送rpc請求，在發(fā)送rpc請求到rpc請求隊(duì)列時(shí)，客戶端至少發(fā)送帶有reply_to以及correlation_id兩個(gè)屬性的信息
服務(wù)端工作流：
等待接受客戶端發(fā)來rpc請求，當(dāng)請求出現(xiàn)的時(shí)候，服務(wù)器從rpc請求隊(duì)列中取出請求，然后處理后，將響應(yīng)發(fā)送到reply_to指定的回調(diào)隊(duì)列中
客戶端接受處理結(jié)果：
客戶端等待回調(diào)隊(duì)列中出現(xiàn)響應(yīng)，當(dāng)響應(yīng)出現(xiàn)時(shí)，它會(huì)根據(jù)響應(yīng)中correlation_id字段的值，將其返回給對應(yīng)的應(yīng)用
服務(wù)者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
import pika# 建立連接，服務(wù)器地址為localhost，可指定ip地址connection = pika.blockingconnection(pika.connectionparameters( host='localhost'))# 建立會(huì)話channel = connection.channel()# 聲明rpc請求隊(duì)列channel.queue_declare(queue='rpc_queue')# 數(shù)據(jù)處理方法def fib(n): if n == 0: return 0 elif n == 1: return 1 else: return fib(n-1) + fib(n-2)# 對rpc請求隊(duì)列中的請求進(jìn)行處理def on_request(ch, method, props, body): n = int(body) print(" [.] fib(%s)" % n) # 調(diào)用數(shù)據(jù)處理方法 response = fib(n) # 將處理結(jié)果(響應(yīng))發(fā)送到回調(diào)隊(duì)列 ch.basic_publish(exchange='', routing_key=props.reply_to, properties=pika.basicproperties(correlation_id = \ props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag = method.delivery_tag)# 負(fù)載均衡，同一時(shí)刻發(fā)送給該服務(wù)器的請求不超過一個(gè)channel.basic_qos(prefetch_count=1)channel.basic_consume(on_request, queue='rpc_queue')print(" [x] awaiting rpc requests")channel.start_consuming() |
客戶端
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import pikaimport uuidclass fibonaccirpcclient(object): def __init__(self): """ 客戶端啟動(dòng)時(shí)，創(chuàng)建回調(diào)隊(duì)列，會(huì)開啟會(huì)話用于發(fā)送rpc請求以及接受響應(yīng) """ # 建立連接，指定服務(wù)器的ip地址 self.connection = pika.blockingconnection(pika.connectionparameters( host='localhost')) # 建立一個(gè)會(huì)話，每個(gè)channel代表一個(gè)會(huì)話任務(wù) self.channel = self.connection.channel() # 聲明回調(diào)隊(duì)列，再次聲明的原因是，服務(wù)器和客戶端可能先后開啟，該聲明是冪等的，多次聲明，但只生效一次 result = self.channel.queue_declare(exclusive=true) # 將次隊(duì)列指定為當(dāng)前客戶端的回調(diào)隊(duì)列 self.callback_queue = result.method.queue # 客戶端訂閱回調(diào)隊(duì)列，當(dāng)回調(diào)隊(duì)列中有響應(yīng)時(shí)，調(diào)用`on_response`方法對響應(yīng)進(jìn)行處理; self.channel.basic_consume(self.on_response, no_ack=true, queue=self.callback_queue) # 對回調(diào)隊(duì)列中的響應(yīng)進(jìn)行處理的函數(shù) def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body # 發(fā)出rpc請求 def call(self, n): # 初始化 response self.response = none #生成correlation_id self.corr_id = str(uuid.uuid4()) # 發(fā)送rpc請求內(nèi)容到rpc請求隊(duì)列`rpc_queue`，同時(shí)發(fā)送的還有`reply_to`和`correlation_id` self.channel.basic_publish(exchange='', routing_key='rpc_queue', properties=pika.basicproperties( reply_to = self.callback_queue, correlation_id = self.corr_id, ), body=str(n)) while self.response is none: self.connection.process_data_events() return int(self.response)# 建立客戶端fibonacci_rpc = fibonaccirpcclient()# 發(fā)送rpc請求print(" [x] requesting fib(30)")response = fibonacci_rpc.call(30)print(" [.] got %r" % response) |
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持服務(wù)器之家。
原文鏈接：http://www.cnblogs.com/peng104/p/10555541.html
