python -- RabbitMq簡單使用

時間 2019-12-11

原文原文鏈接

轉至：http://blog.csdn.net/whycold/article/details/41119807

我轉到這裏主要是想在本身使用時方便查詢。。。

引言

你是否遇到過兩個（多個）系統間須要經過定時任務來同步某些數據？你是否在爲異構系統的不一樣進程間相互調用、通信的問題而苦惱、掙扎？若是是，那麼恭喜你，消息服務讓你能夠很輕鬆地解決這些問題。
消息服務擅長於解決多系統、異構系統間的數據交換（消息通知/通信）問題，你也能夠把它用於系統間服務的相互調用（RPC）。本文將要介紹的RabbitMQ就是當前最主流的消息中間件之一。
php

RabbitMQ簡介

AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，爲面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。
AMQP的主要特徵是面向消息、隊列、路由（包括點對點和發佈/訂閱）、可靠性、安全。
RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。
下面將重點介紹RabbitMQ中的一些基礎概念，瞭解了這些概念，是使用好RabbitMQ的基礎。java

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ對外提供的API中最基本的對象。Connection是RabbitMQ的socket連接，它封裝了socket協議相關部分邏輯。ConnectionFactory爲Connection的製造工廠。
Channel是咱們與RabbitMQ打交道的最重要的一個接口，咱們大部分的業務操做是在Channel這個接口中完成的，包括定義Queue、定義Exchange、綁定Queue與Exchange、發佈消息等。python

Queue

Queue（隊列）是RabbitMQ的內部對象，用於存儲消息，用下圖表示。
redis

RabbitMQ中的消息都只能存儲在Queue中，生產者（下圖中的P）生產消息並最終投遞到Queue中，消費者（下圖中的C）能夠從Queue中獲取消息並消費。
安全

多個消費者能夠訂閱同一個Queue，這時Queue中的消息會被平均分攤給多個消費者進行處理，而不是每一個消費者都收到全部的消息並處理。
服務器

Message acknowledgment

在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現其餘意外）的狀況，這種狀況下就可能會致使消息丟失。爲了不這種狀況發生，咱們能夠要求消費者在消費完消息後發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）後纔將該消息從Queue中移除；若是RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ鏈接斷開，則RabbitMQ會將該消息發送給其餘消費者（若是存在多個消費者）進行處理。這裏不存在timeout概念，一個消費者處理消息時間再長也不會致使該消息被髮送給其餘消費者，除非它的RabbitMQ鏈接斷開。
這裏會產生另一個問題，若是咱們的開發人員在處理完業務邏輯後，忘記發送回執給RabbitMQ，這將會致使嚴重的bug——Queue中堆積的消息會愈來愈多；消費者重啓後會重複消費這些消息並重復執行業務邏輯…負載均衡

另外pub message是沒有ack的。異步

Message durability

若是咱們但願即便在RabbitMQ服務重啓的狀況下，也不會丟失消息，咱們能夠將Queue與Message都設置爲可持久化的（durable），這樣能夠保證絕大部分狀況下咱們的RabbitMQ消息不會丟失。但依然解決不了小几率丟失事件的發生（好比RabbitMQ服務器已經接收到生產者的消息，但還沒來得及持久化該消息時RabbitMQ服務器就斷電了），若是咱們須要對這種小几率事件也要管理起來，那麼咱們要用到事務。因爲這裏僅爲RabbitMQ的簡單介紹，因此這裏將不講解RabbitMQ相關的事務。socket

Prefetch count

前面咱們講到若是有多個消費者同時訂閱同一個Queue中的消息，Queue中的消息會被平攤給多個消費者。這時若是每一個消息的處理時間不一樣，就有可能會致使某些消費者一直在忙，而另一些消費者很快就處理完手頭工做並一直空閒的狀況。咱們能夠經過設置prefetchCount來限制Queue每次發送給每一個消費者的消息數，好比咱們設置prefetchCount=1，則Queue每次給每一個消費者發送一條消息；消費者處理完這條消息後Queue會再給該消費者發送一條消息。
分佈式

Exchange

在上一節咱們看到生產者將消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的狀況是，生產者將消息發送到Exchange（交換器，下圖中的X），由Exchange將消息路由到一個或多個Queue中（或者丟棄）。

Exchange是按照什麼邏輯將消息路由到Queue的？這個將在Binding一節介紹。
RabbitMQ中的Exchange有四種類型，不一樣的類型有着不一樣的路由策略，這將在Exchange Types一節介紹。

routing key

生產者在將消息發送給Exchange的時候，通常會指定一個routing key，來指定這個消息的路由規則，而這個routing key須要與Exchange Type及binding key聯合使用才能最終生效。
在Exchange Type與binding key固定的狀況下（在正常使用時通常這些內容都是固定配置好的），咱們的生產者就能夠在發送消息給Exchange時，經過指定routing key來決定消息流向哪裏。
RabbitMQ爲routing key設定的長度限制爲255 bytes。

Binding

RabbitMQ中經過Binding將Exchange與Queue關聯起來，這樣RabbitMQ就知道如何正確地將消息路由到指定的Queue了。

Binding key

在綁定（Binding）Exchange與Queue的同時，通常會指定一個binding key；消費者將消息發送給Exchange時，通常會指定一個routing key；當binding key與routing key相匹配時，消息將會被路由到對應的Queue中。這個將在Exchange Types章節會列舉實際的例子加以說明。
在綁定多個Queue到同一個Exchange的時候，這些Binding容許使用相同的binding key。
binding key 並非在全部狀況下都生效，它依賴於Exchange Type，好比fanout類型的Exchange就會無視binding key，而是將消息路由到全部綁定到該Exchange的Queue。

Exchange Types

RabbitMQ經常使用的Exchange Type有fanout、direct、topic、headers這四種（AMQP規範裏還提到兩種Exchange Type，分別爲system與自定義，這裏不予以描述），下面分別進行介紹。

fanout

fanout類型的Exchange路由規則很是簡單，它會把全部發送到該Exchange的消息路由到全部與它綁定的Queue中。

上圖中，生產者（P）發送到Exchange（X）的全部消息都會路由到圖中的兩個Queue，並最終被兩個消費者（C1與C2）消費。

direct

direct類型的Exchange路由規則也很簡單，它會把消息路由到那些binding key與routing key徹底匹配的Queue中。

以上圖的配置爲例，咱們以routingKey=」error」發送消息到Exchange，則消息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；若是咱們以routingKey=」info」或routingKey=」warning」來發送消息，則消息只會路由到Queue2。若是咱們以其餘routingKey發送消息，則消息不會路由到這兩個Queue中。

topic

前面講到direct類型的Exchange路由規則是徹底匹配binding key與routing key，但這種嚴格的匹配方式在不少狀況下不能知足實際業務需求。topic類型的Exchange在匹配規則上進行了擴展，它與direct類型的Exchage類似，也是將消息路由到binding key與routing key相匹配的Queue中，但這裏的匹配規則有些不一樣，它約定：

routing key爲一個句點號「. 」分隔的字符串（咱們將被句點號「. 」分隔開的每一段獨立的字符串稱爲一個單詞），如「stock.usd.nyse」、「nyse.vmw」、「quick.orange.rabbit」
binding key與routing key同樣也是句點號「. 」分隔的字符串
binding key中能夠存在兩種特殊字符「*」與「#」，用於作模糊匹配，其中「*」用於匹配一個單詞，「#」用於匹配多個單詞（能夠是零個）

以上圖中的配置爲例，routingKey=」quick.orange.rabbit」的消息會同時路由到Q1與Q2，routingKey=」lazy.orange.fox」的消息會路由到Q1與Q2，routingKey=」lazy.brown.fox」的消息會路由到Q2，routingKey=」lazy.pink.rabbit」的消息會路由到Q2（只會投遞給Q2一次，雖然這個routingKey與Q2的兩個bindingKey都匹配）；routingKey=」quick.brown.fox」、routingKey=」orange」、routingKey=」quick.orange.male.rabbit」的消息將會被丟棄，由於它們沒有匹配任何bindingKey。

headers

headers類型的Exchange不依賴於routing key與binding key的匹配規則來路由消息，而是根據發送的消息內容中的headers屬性進行匹配。
在綁定Queue與Exchange時指定一組鍵值對；當消息發送到Exchange時，RabbitMQ會取到該消息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否徹底匹配Queue與Exchange綁定時指定的鍵值對；若是徹底匹配則消息會路由到該Queue，不然不會路由到該Queue。
該類型的Exchange沒有用到過（不過也應該頗有用武之地），因此不作介紹。

RPC

MQ自己是基於異步的消息處理，前面的示例中全部的生產者（P）將消息發送到RabbitMQ後不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條消息都不知道）。
但實際的應用場景中，咱們極可能須要一些同步處理，須要同步等待服務端將個人消息處理完成後再進行下一步處理。這至關於RPC（Remote Procedure Call，遠程過程調用）。在RabbitMQ中也支持RPC。

RabbitMQ中實現RPC的機制是：

客戶端發送請求（消息）時，在消息的屬性（MessageProperties，在AMQP協議中定義了14中properties，這些屬性會隨着消息一塊兒發送）中設置兩個值replyTo（一個Queue名稱，用於告訴服務器處理完成後將通知個人消息發送到這個Queue中）和correlationId（這次請求的標識號，服務器處理完成後須要將此屬性返還，客戶端將根據這個id瞭解哪條請求被成功執行了或執行失敗）
服務器端收到消息並處理
服務器端處理完消息後，將生成一條應答消息到replyTo指定的Queue，同時帶上correlationId屬性
客戶端以前已訂閱replyTo指定的Queue，從中收到服務器的應答消息後，根據其中的correlationId屬性分析哪條請求被執行了，根據執行結果進行後續業務處理

總結

　　本文介紹了RabbitMQ中我的認爲最重要的概念，充分利用RabbitMQ提供的這些功能就能夠處理咱們絕大部分的異步業務了。

一些例子

裏面都帶有我本身的一些理解註釋

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 # connection = pika.BlockingConnection(pika.ConnectionParameters(
 7 #     'localhost'))
 8 
 9 user_pass = pika.PlainCredentials('txowner','helloworld')
10 connection = pika.BlockingConnection(pika.ConnectionParameters(
11     host='localhost',
12     port=5672,
13     virtual_host='/',
14     credentials=user_pass
15 ))
16 
17 channel = connection.channel()
18 
19 channel.queue_declare(queue='message',durable=True)
20 
21 #這裏參數形式模塊定義好了，必須這樣寫，函數名是自定義的
22 def callback(ch,method,properties,body):
23     print(" [x] Received %r " %body.decode('utf8'))
24     ch.basic_ack(delivery_tag=method.delivery_tag)
25     #配套生產者的消息持久化，做用：讓rabbitmq在一條消息被接受執行完成後將該條消息刪除
26 
27 #同時最多取prefetch_count個消息，這裏爲1.做用：負載均衡
28 channel.basic_qos(prefetch_count=1)
29 
30 #no_ack=True 表示不用確認，queue='message'表示監聽message這個隊列
31 channel.basic_consume(callback,queue='message',no_ack=True)
32 
33 print(" [*] Waiting for messages.To exit press CTRL+C")
34 channel.start_consuming() #監聽通道並阻塞

rabbit_recieve

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     'localhost'))  #默認鏈接guest用戶，也有密碼
 8 
 9 #鏈接有密碼的用戶
10 user_pass = pika.PlainCredentials('txowner','helloworld')
11 connection = pika.BlockingConnection(pika.ConnectionParameters(
12     host='localhost',
13     port=5672,
14     virtual_host='/',
15     credentials=user_pass
16 ))
17 
18 channel = connection.channel()
19 
20 #聲明queue，默認非持久化
21 # channel.queue_declare(queue='hello')
22 
23 #聲明持久化的隊列,durable是把隊列那根管道持久化，不是隊列裏面的消息
24 channel.queue_declare(queue='message',durable=True)
25 
26 # RabbitMQ a message can never be sent directly to the queue,
27 # it always needs to go through an exchange
28 '''
29 #若同時開了多個接收方，這樣只有一個接收方能收到
30 channel.basic_publish(exchange='',
31                           routing_key='hello',
32                           body='hello world!')
33     print(" [x] send 'hello world!' ")
34 '''
35 #這種狀況下，會平均把三條消息輪詢發給三個接收方(若至少要三個接收方)
36 for i in range(3):
37     channel.basic_publish(exchange='',
38                           routing_key='message',
39                           body='hello world! %s' %i,
40                           properties=pika.BasicProperties(delivery_mode=2)
41                           #這句話是把消息設置持久化
42                           )
43     print(" [x] send 'hello world!' ")
44 connection.close()

rabbitmq_send

修改Exchange的類型，以決定究竟是哪些Queue符合條件，能夠接收消息：

 channel.basic_publish(exchange='',     #在這裏指定
                          routing_key='message',
                          body='hello world!'
                          )

一、fanout: 全部bind到此exchange的queue均可以接收消息：

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='logs', type='fanout')
12 
13 message = ' '.join(sys.argv[1:]) or "info: Hello World!"
14 channel.basic_publish(exchange='logs',
15                       routing_key='',  #fanout是綁定到此exchange的隊列都能接收到消息，因此不指定隊列名
16                       body=message)
17 print(" [x] Sent %r" % message)
18 connection.close()

rabbitmq_fanout_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host='127.0.0.1'))
 8 channel = connection.channel()
 9 
10 channel.exchange_declare(exchange='logs',type='fanout')
11 
12 result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit會隨機分配一個名字,exclusive=True會在使用此queue的消費者斷開後,自動將queue刪除
13 queue_name = result.method.queue #生成一個隨機的queue
14 
15 channel.queue_bind(exchange='logs',queue=queue_name)
16 
17 print(' [*] Waiting for logs. To exit press CTRL+C')
18 
19 
20 def callback(ch, method, properties, body):
21     print(" [x] %r" % body)
22 
23 
24 channel.basic_consume(callback,
25                       queue=queue_name,
26                       no_ack=True)
27 
28 channel.start_consuming()

rabbitmq_fanout_subscriber

二、direct: 經過routingKey和exchange決定的那個惟一的queue能夠接收消息：

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 '''有選擇的發佈消息'''
 5 
 6 import pika
 7 import sys
 8 
 9 connection = pika.BlockingConnection(pika.ConnectionParameters(
10     host='127.0.0.1'))
11 channel = connection.channel()
12 
13 channel.exchange_declare(exchange='direct_logs', type='direct')
14 
15 severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
16 message = ' '.join(sys.argv[2:]) or 'Hello World!'
17 channel.basic_publish(exchange='direct_logs',
18                       routing_key=severity,
19                       body=message)
20 print(" [x] Sent %r:%r" % (severity, message))
21 connection.close()

rabbitmnq_direct_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='direct_logs', type='direct')
12 
13 result = channel.queue_declare(exclusive=True) # 不指定queue名字,rabbit會隨機分配一個名字,exclusive=True會在使用此queue的消費者斷開後,自動將queue刪除
14 
15 queue_name = result.method.queue
16 
17 severities = sys.argv[1:]
18 if not severities:
19     sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
20     sys.exit(1)
21 
22 for severity in severities:
23     channel.queue_bind(exchange='direct_logs',
24                        queue=queue_name,
25                        routing_key=severity)
26 
27 print(' [*] Waiting for logs. To exit press CTRL+C')
28 
29 
30 def callback(ch, method, properties, body):
31     print(" [x] %r:%r" % (method.routing_key, body))
32 
33 
34 channel.basic_consume(callback,
35                       queue=queue_name,
36                       no_ack=True)
37 
38 channel.start_consuming()

rabbitmq_direct_subscriber

三、topic:全部符合routingKey(此時能夠是一個表達式)的routingKey所bind的queue能夠接收消息

　　表達式符號說明：#表明一個或多個字符，*表明任何字符
例：#.a會匹配a.a，aa.a，aaa.a等
*.a會匹配a.a，b.a，c.a等
注：使用RoutingKey爲#，Exchange Type爲topic的時候至關於使用fanout

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='topic_logs', type='topic')
12 
13 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
14 message = ' '.join(sys.argv[2:]) or 'Hello World!'
15 
16 channel.basic_publish(exchange='topic_logs',
17                       routing_key=routing_key,
18                       body=message)
19 
20 print(" [x] Sent %r:%r" % (routing_key, message))

rabbitmq_topic_publisher

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import sys
 6 
 7 connection = pika.BlockingConnection(pika.ConnectionParameters(
 8     host='127.0.0.1'))
 9 channel = connection.channel()
10 
11 channel.exchange_declare(exchange='topic_logs', type='topic')
12 
13 result = channel.queue_declare(exclusive=True)
14 queue_name = result.method.queue
15 
16 binding_keys = sys.argv[1:]
17 if not binding_keys:
18     sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
19     sys.exit(1)
20 
21 for binding_key in binding_keys:
22     channel.queue_bind(exchange='topic_logs',
23                        queue=queue_name,
24                        routing_key=binding_key)
25 
26 print(' [*] Waiting for logs. To exit press CTRL+C')
27 
28 
29 def callback(ch, method, properties, body):
30     print(" [x] %r:%r" % (method.routing_key, body))
31 
32 
33 channel.basic_consume(callback,
34                       queue=queue_name,
35                       no_ack=True)
36 
37 channel.start_consuming()

rabbitmq_topic_subscriber

四、headers: 經過headers 來決定把消息發給哪些queue

RPC

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 
 6 connection = pika.BlockingConnection(pika.ConnectionParameters(
 7     host='127.0.0.1'))
 8 
 9 channel = connection.channel()
10 
11 channel.queue_declare(queue='rpc_queue')
12 
13 
14 def fib(n):
15     if n == 0:
16         return 0
17     elif n == 1:
18         return 1
19     else:
20         return fib(n - 1) + fib(n - 2)
21 
22 
23 def on_request(ch, method, props, body):
24     n = int(body)
25 
26     print(" [.] fib(%s)" % n)
27     response = fib(n)
28 
29     ch.basic_publish(exchange='',
30                      routing_key=props.reply_to,
31                      properties=pika.BasicProperties(correlation_id= \
32                                                 props.correlation_id),
33                      body=str(response))
34     ch.basic_ack(delivery_tag=method.delivery_tag)
35 
36 
37 channel.basic_qos(prefetch_count=1)
38 channel.basic_consume(on_request, queue='rpc_queue')
39 
40 print(" [x] Awaiting RPC requests")
41 channel.start_consuming()

rabbitmq_rpc_server

 1 #!/usr/bin/env python
 2 # -*- coding:utf8 -*-
 3 
 4 import pika
 5 import uuid
 6 
 7 
 8 class FibonacciRpcClient(object):
 9     def __init__(self):
10         self.connection = pika.BlockingConnection(pika.ConnectionParameters(
11             host='127.0.0.1'))
12 
13         self.channel = self.connection.channel()
14 
15         result = self.channel.queue_declare(exclusive=True)
16         self.callback_queue = result.method.queue
17 
18         self.channel.basic_consume(self.on_response,
19                                    no_ack=True,
20                                    queue=self.callback_queue
21                                    )
22 
23     def on_response(self, ch, method, props, body):
24         if self.corr_id == props.correlation_id:
25             self.response = body
26 
27     def call(self, n):
28         self.response = None
29         self.corr_id = str(uuid.uuid4())
30         self.channel.basic_publish(exchange='',
31                                    routing_key='rpc_queue',
32                                    properties=pika.BasicProperties(
33                                        reply_to=self.callback_queue,
34                                        correlation_id=self.corr_id,
35                                    ),
36                                    body=str(n)
37                                 )
38 
39         while self.response is None:
40             self.connection.process_data_events()   #不斷去輪詢，看是否有數據返回
41         return int(self.response)
42 
43 
44 fibonacci_rpc = FibonacciRpcClient()
45 
46 print(" [x] Requesting fib(30)")
47 response = fibonacci_rpc.call(30)
48 print(" [.] Got %r" % response)

rabbitmq_rpc_client

使用rabbitmq注意點：

#!/usr/bin/env python
# -*- coding:utf8 -*-

'''
rabbitmq在非持久化狀況下：
    生產者和消費者不須要像redis同樣同時在線，只要rabbitmq服務器沒有重啓或關閉，
生成的隊列就一直在。此時，若生產者發送數據後，數據就會存放在隊列裏面，消費者一上線就
能接收到。

rabbitmq在持久化狀況下：
    一、只是生產者持久化管道隊列，生產者發送數據存在管道中時，重啓rabbitmq服務器將會
      使消息丟失，可是管道隊列仍是存在，此時能夠不用再聲明，能夠直接向該隊列存消息
    二、同時持久化隊列和消息，那麼不管服務器是否重啓，只要生產者發送消息到隊列裏面，
      消費者一上線就能接收到

一、rabbitmq默認消息不是持久化的，因此當rabbitmq-server關閉後，生產者發送到消息會丟失
二、若要rabbitmq-server重啓後消息不丟失，則須要把隊列設置爲持久化隊列
三、當一個隊列被設置爲非持久化(默認)時，不能直接修改成持久化隊列，有兩種方式：
    (1) 刪除該隊列從新設置爲持久化隊列
    (2) 從新設置其餘隊列(不能與已存在隊列重名)

四、聲明持久化的隊列
(這裏是把隊列設置爲持久化，因此重啓服務器後，隊列還在。並不會把消息也持久化)
# channel.queue_declare(queue='message',durable=True)
  
  聲明非持久化隊列
# channel.queue_declare(queue='message')

  消息持久化(在發送消息時，在消息後面一個參數設置這個)
    properties=pika.BasicProperties(delivery_mode=2)
   eg： 
    server:
    channel.basic_publish(exchange='',
                          routing_key='message',
                          body='hello world!',
                          properties=pika.BasicProperties(delivery_mode=2)
                          )
    
    client:
    同時在客戶端要在回調函數裏面設置這個：
    ch.basic_ack(delivery_tag=method.delivery_tag)
    做用：在客戶端消費消息後，通知rabbitmq將該條消息刪除
    
    def callback(ch,method,properties,body):
        print(" [x] Received %r " %body.decode('utf8'))
        ch.basic_ack(delivery_tag=method.delivery_tag)
'''

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。