[TOC] ## 1. 安裝 ### 1.1 安裝 下載deb包就行，要求有erlang環境 ### 1.2 配置 1. 第一件事要創建用戶，因為缺省的guest/guest用戶只能在本地登錄，所以先用命令行創建一個tuna/tuna，并讓他成為管理員。 ~~~ rabbitmqctl add_user tuna tuna rabbitmqctl set_user_tags tuna administrator ~~~ 2. 啟用WEB管理。 `rabbitmq-plugins enable rabbitmq_management` 用新建的tuna用戶，密碼：tuna登錄 ~~~ http://192.168.56.130:15672 ~~~  3. 修改tuna訪問權限  ## 2. 使用 ### 2.1 簡單收發 1. producer ~~~ __author__ = 'dailin' import pika class MessageProducer: def getConnection(self): credentials = pika.PlainCredentials('tuna', 'tuna') connection = pika.BlockingConnection(pika.ConnectionParameters( host='192.168.56.130', credentials=credentials )) return connection if __name__ == '__main__': producer = MessageProducer() # 獲取連接 connection = producer.getConnection() # 打開管道 channel = connection.channel() # 在管道中聲明隊列 channel.queue_declare(queue='hello',durable=True) channel.basic_publish(exchange='', routing_key='hello', body='hello world', properties=pika.BasicProperties( delivery_mode=2 # 消息持久化 ) ) print(" [x] Sent 'Hello World!'") connection.close() ~~~ 2. consumer ~~~ __author__ = 'dailin' class Consumer: import pika # 使用tuna用戶登錄 credentials = pika.PlainCredentials('tuna', 'tuna') connection = pika.BlockingConnection(pika.ConnectionParameters( host='192.168.56.130', port=5672, credentials=credentials )) channel = connection.channel() # You may ask why we declare the queue again ? we have already declared it in our previous code. # We could avoid that if we were sure that the queue already exists. For example if send.py program # was run before. But we're not yet sure which program to run first. In such cases it's a good # practice to repeat declaring the queue in both programs. channel.queue_declare(queue='hello',durable=True) # durable=True 隊列持久化（隊列中的消息沒有持久化） def callback(ch, method, properties, body): print(ch, method, properties, body) channel.basic_consume(callback, queue='hello', no_ack=True) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming() ~~~ ## 3. 消息持久化 主要有兩步： 1. channel.queue_declare(queue='hello') 聲明隊列時（生產者和消費者），隊列持久化，rabbitmq重啟后，隊列依然存在 2. 生產者：隊列中的消息持久化 ~~~ channel.basic_publish(exchange='', routing_key='hello', body='hello world', properties=pika.BasicProperties( delivery_mode=2 # 消息持久化 ) ) ~~~ ## 4. 訂閱發布 > 之前的例子都基本都是1對1的消息發送和接收，即消息只能發送到指定的queue里，但有些時候你想讓你的消息被所有的Queue收到，類似廣播的效果，這時候就要用到exchange了， > An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type. > Exchange在定義的時候是有類型的，以決定到底是哪些Queue符合條件，可以接收消息 > fanout: 所有bind到此exchange的queue都可以接收消息（實時的，收到就收到了，類似于廣播） > direct: 通過routingKey和exchange決定的那個唯一的queue可以接收消息 > topic:所有符合routingKey(此時可以是一個表達式)的routingKey所bind的queue可以接收消息 > 　　 表達式符號說明：#代表一個或多個字符，*代表任何字符 > 例：#.a會匹配a.a，aa.a，aaa.a等 > *.a會匹配a.a，b.a，c.a等 > 注：使用RoutingKey為#，Exchange Type為topic的時候相當于使用fanout　 > headers: 通過headers 來決定把消息發給哪些queue 1. publisher ~~~ import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', # 聲明exchange，類型為fanout，名稱為logs type='fanout') message = ' '.join(sys.argv[1:]) or "info: Hello World!" channel.basic_publish(exchange='logs', # 使用logs（exchange） routing_key='', body=message) print(" [x] Sent %r" % message) connection.close() ~~~ 2. consumer ~~~ #_*_coding:utf-8_*_ import pika connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', # 聲明exchange，類型為fanout，名稱為logs type='fanout') # 不指定queue名字,rabbit會隨機分配一個名字,然后exchange向這個queue里發消息，消費者就可以收到了。clusive=True會在使用此queue的消費者斷開后,自動將queue刪除 result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind(exchange='logs', # 讓queue綁定exchange 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() ~~~