Rabbitmq-server 分析
阿新 • • 發佈：2018-05-31
rabbitmq queue一、
發送端：
發送端：
#coding:utf-8 import pika user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘) connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd)) #建立一個實例 channel = connection.channel() # 聲明一個管道，在管道裏發消息 channel.queue_declare(queue=‘hello‘) #在管道裏聲明queue，並指定queue_name #RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange. channel.basic_publish(exchange=‘‘, routing_key=‘hello‘, # queue名字 body=‘Hello World! ‘) # 消息內容 print(" [x] Sent ‘Hello World!‘") connection.close() # 隊列關閉
接收端：
#coding:utf-8 import pika import time user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘) connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.206‘, credentials=user_pwd)) # 建立實例 channel = connection.channel() # 聲明管道 # 在不確定procedure與consumer哪個先運行的情況下，在consumer端 再次聲明QUEUE。 channel.queue_declare(queue=‘hello‘) def callback(ch, method, properties, body): #定義回調函數，四個參數為標準格式 print(" [x] Received %r" % body) # 輸出接收到的消息 time.sleep(15) ch.basic_ack(delivery_tag=method.delivery_tag) # 告訴生產者消息處理完成 channel.basic_consume( # 消費消息 callback, # 如果收到消息，就調用callback函數來處理消息 queue=‘hello‘, # 你要從那個隊列裏收消息 #no_ack=True # True 表示consumer取完消息後不給procedure發送ack確認，註意本例中如果為True，它會與callback函數中的basic_ack語句產生沖突；默認False ) print(‘ [*] Waiting for messages. To exit press CTRL+C‘) channel.start_consuming() # 開始消費消息
驗證：
在多個shell窗口中運行consumer，然後運行procedure，發現每個consumer 輪流接收相同隊列中的消息。
二、
發送端：
#coding:utf-8 import pika user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘) connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd)) channel = connection.channel() channel.queue_declare(queue=‘hello2‘, durable=True) #創建一個新隊列task_queue，設置隊列持久化，註意不要跟已存在的隊列重名，否則有報錯 #n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange. channel.basic_publish(exchange=‘‘, routing_key=‘hello2‘, body=‘Hello World 2 !‘, properties=pika.BasicProperties(delivery_mode=2, ) # make message persistent ) print(" [x] Sent ‘Hello World!‘") connection.close()
接收端：
#coding:utf-8
import pika
import time
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.queue_declare(queue=‘hello2‘, durable=True)
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(10)
ch.basic_ack(delivery_tag=method.delivery_tag) # 告訴生產者，消息處理完成
channel.basic_qos(prefetch_count=1) # 類似權重，按能力分發，如果有一個消息，就不在給你發
channel.basic_consume( # 消費消息
callback, # 如果收到消息，就調用callback
queue=‘hello2‘,
)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()(1) rabbitmq循環調度，將消息循環發送給不同的消費者
(2) 消息確認機制，為了確保一個消息不會丟失，RabbitMQ支持消息的確認 , 一個 ack(acknowlegement) 是從消費者端發送一個確認去告訴RabbitMQ 消息已經接收了、處理了，RabbitMQ可以釋放並刪除掉了。如果一個消費者死掉了（channel關閉、connection關閉、或者TCP連接斷開了）而沒有發送ack，RabbitMQ 就會認為這個消息沒有被消費者處理，並會重新發送到生產者的隊列裏，如果同時有另外一個消費者在線，rabbitmq將會將消息很快轉發到另外一個消費者中。
(3) 消息持久化，將消息寫入硬盤中。RabbitMQ不允許你重新定義一個已經存在、但屬性不同的queue。關於消息為持久化， 需要生產者定義持久化的queue，以及通過設置 delivery_mode 屬性為 2來標記本消息持久化。
(4) 公平調度。在一個消費者未處理完一個消息之前不分發新的消息給它，而是將這個新消息分發給另一個不是很忙的消費者進行處理。為了解決這個問題可以在消費者代碼中使用 channel.basic.qos ( prefetch_count = 1 )，將消費者設置為公平調度
驗證：
在多個shell窗口中打開consumer，然後運行procedure，觀察收到消息為RR
三、
exchange：交換機。生產者不是將消息發送給隊列，而是將消息發送給交換機，由交換機決定將消息發送給哪個隊列。所以exchange必須準確知道消息是要送到哪個隊列，還是要被丟棄。因此要在exchange中給exchange定義規則，所有的規則都是在exchange的類型中定義的。
(1) fanout
生產者：
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一個實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
# 註意：這裏是廣播，不需要聲明queue
channel.exchange_declare(‘logs‘, ‘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()消費者：
#coding:utf-8
import pika
import time
user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
# 聲明管道
channel = connection.channel()
channel.exchange_declare(‘logs‘, ‘fanout‘)
# 不指定queue名字,rabbit會隨機分配一個名字,exclusive=True會在使用此queue的消費者斷開後,自動將queue刪除
result = channel.queue_declare(exclusive=True)
# 獲取隨機的queue名字
queue_name = result.method.queue
#print("random queuename:", queue_name)
channel.queue_bind(exchange=‘logs‘, # queue綁定到轉發器上
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) direct
生產者：
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一個實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘direct_logs‘, ‘direct‘)
# 重要程度級別，這裏默認定義為 info
severity = sys.argv[1] if len(sys.argv) > 1 else ‘warning‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World! direct log‘
channel.basic_publish(exchange=‘direct_logs‘, routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()消費者：
#coding:utf-8
import pika
import time, sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
# 建立實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘direct_logs‘, ‘direct‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 獲取運行腳本所有的參數
severities = sys.argv[1:]
if not severities:
#severities = [‘error‘, ‘warning‘]
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) topic
生產者:
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一個實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
# 聲明一個管道，在管道裏發消息
channel = connection.channel()
channel.exchange_declare(‘topic_logs‘, ‘topic‘)
routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘
channel.basic_publish(exchange=‘topic_logs‘, routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
消費者：
#coding:utf-8
import pika
import time, sys
user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立實例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘topic_logs‘, ‘topic‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_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-server 分析