一、RabbitMQ在Ubuntu 16.04下的安裝與配置

新增源

echo 'deb http://www.rabbitmq.com/debian/ testing main' | sudo tee /etc/apt/sources.list.d/rabbitmq.list

新增公鑰

wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add -

更新源

sudo apt-get update

安裝rabbitmq-server

sudo apt-get install rabbitmq-server

通過上篇介紹的命令invoke-rc.d rabbitmq-server stop/start/etc我們看見程序是啟動的

[email protected]:/etc/apt/sources.list.d$ invoke-rc.d rabbitmq-server status 
● rabbitmq-server.service - RabbitMQ broker
   Loaded: loaded (/lib/systemd/system/rabbitmq-server.service; enabled; vendor preset: enabled)
   Active: active (running) since 四 2017-05-25 20:42:25 CST; 5min ago
 Main PID: 4865 (beam.smp)
   Status: "Initialized"
   CGroup: /system.slice/rabbitmq-server.service
           ├─4865 /usr/lib/erlang/erts-7.3/bin/beam.smp -W w -A 64 -P 1048576 -t 5000000 -stbt db -zdbbl 32000 -K true -- -root /usr/lib/erlang
           ├─4942 /usr/lib/erlang/erts-7.3/bin/epmd -daemon
           ├─5074 inet_gethost 4
           └─5075 inet_gethost 4

5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:               RabbitMQ 3.6.10. Copyright (C) 2007-2017 Pivotal Software, Inc.
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:   ##  ##      Licensed under the MPL.  See http://www.rabbitmq.com/
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:   ##  ##
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:   ##########  Logs: /var/log/rabbitmq/
 
[email protected]
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:   ######  ##        /var/log/rabbitmq/[email protected]
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:   ##########
5月 25 20:42:21 xq-VPCEG17YC rabbitmq-server[4865]:               Starting broker...
5月 25 20:42:25 xq-VPCEG17YC rabbitmq-server[4865]: systemd unit for activation check: "rabbitmq-server.service"
5月 25 20:42:25 xq-VPCEG17YC systemd[1]: Started RabbitMQ broker.
5月 25 20:42:25 xq-VPCEG17YC rabbitmq-server[4865]:  completed with 0 plugins.
 

開啟管理頁面 

sudo rabbitmq-plugins enable rabbitmq_management

檢視安裝的外掛 

sudo rabbitmqctl list_users

檢視使用者 

sudo rabbitmqctl list_users

新增管理員使用者 

sudo rabbitmqctl add_user admin admin 
sudo rabbitmqctl set_user_tags admin administrator

用剛設定的賬戶登入管理頁面 

二、python RabbitMQ佇列使用

1. 關於python的queue介紹

    關於python的佇列，內建的有兩種，一種是執行緒queue，另一種是程序queue，但是這兩種queue都是隻能在同一個程序下的執行緒間或者父程序與子程序之間進行佇列通訊，並不能進行程式與程式之間的資訊交換，這時候我們就需要一箇中間件，來實現程式之間的通訊。

2. RabbitMQ

    MQ並不是python內建的模組，而是一個需要你額外安裝的程式，安裝完畢後可通過python中內建的pika模組來呼叫MQ傳送或接收佇列請求。接下來我們就看幾種python呼叫MQ的模式（作者自定義中文形象的模式名稱）與方法。

3. RabbitMQ設定遠端連結賬號密碼

1) 啟動rabbitmq web服務

2) 遠端訪問rabbitmq: 自己增加一個使用者，步驟如下：

(1)  建立一個admin使用者

sudo rabbitmqctl add_user admin 123123

(2)  設定該使用者為administrator角色

sudo rabbitmqctl set_user_tags admin administrator

(3)  設定許可權

sudo rabbitmqctl  set_permissions  -p  '/'  admin '.' '.' '.'

(4)  重啟rabbitmq服務

sudo service rabbitmq-server restart

之後就能用admin使用者遠端連線rabbitmq server了。

4. 輪詢消費模式

此模式下，傳送佇列的一方把訊息存入mq的指定佇列後，若有消費者端聯入相應佇列，即會獲取到訊息，並且佇列中的訊息會被消費掉。

若有多個消費端同時連線著佇列，則會已輪詢的方式將佇列中的訊息消費掉。

接下來是程式碼例項：

producer生產者

# !/usr/bin/env python
import pika
credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',credentials))
channel = connection.channel()

# 宣告queue
channel.queue_declare(queue='balance')

# 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='balance',
                      body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

傳送過佇列後，可在MQ伺服器中檢視佇列狀態

[[email protected] ~]# rabbitmqctl list_queues
Listing queues ...
hello    1

consumer消費者

# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika

credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',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='balance')


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)


channel.basic_consume(callback,
                      queue='balance',
                      no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

接收佇列後，檢視一下佇列狀態

[[email protected] ~]#  rabbitmqctl list_queues
Listing queues ...
hello    0

5. 佇列持久化

當rabbitMQ意外宕機時，可能會有持久化儲存佇列的需求（佇列中的訊息不消失）。
producer

# Cheng
# !/usr/bin/env python
import pika

credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',credentials))
channel = connection.channel()

# 宣告queue
channel.queue_declare(queue='durable',durable=True)

# 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='durable',
                      body='Hello cheng!',
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                      )
                      )
print(" [x] Sent 'Hello cheng!'")
connection.close()

執行後檢視佇列，記下佇列名字與佇列中所含訊息的數量

[[email protected] ~]# rabbitmqctl list_queues
Listing queues ...
durable    1
#重啟rabbitmq
[[email protected] ~]# systemctl restart rabbitmq-server
#重啟完畢後再次檢視
[[email protected] ~]# rabbitmqctl list_queues
Listing queues ...
durable   #佇列以及訊息並未消失

執行消費者程式碼

cunsumer

# Cheng
# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika

credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',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='durable',durable=True)


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback,
                      queue='durable',
                      #no_ack=True
                      )

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

可正確接收到資訊。

再次檢視佇列的情況。

[[email protected] ~]# rabbitmqctl list_queues
Listing queues ...
durable    0

7. 廣播模式

當producer傳送訊息到佇列後，所有的consumer都會收到訊息，需要注意的是，此模式下producer與concerned之間的關係類似與廣播電臺與收音機，如果廣播後收音機沒有接受到，那麼訊息就會丟失。
建議先執行concerned
concerned

# _*_coding:utf-8_*_
__author__ = 'Alex Li'
import pika

credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',credentials))
channel = connection.channel()

channel.exchange_declare(exchange='Clogs',
                         type='fanout')

result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit會隨機分配一個名字,exclusive=True會在使用此queue的消費者斷開後,自動將queue刪除
queue_name = result.method.queue

channel.queue_bind(exchange='Clogs',
                   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()

producer

import pika
import sys

credentials = pika.PlainCredentials('admin','123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.56.19',5672,'/',credentials))
channel = connection.channel()

channel.exchange_declare(exchange='Clogs',
                         type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='Clogs',
                      routing_key='',
                      body=message)
print(" [x] Sent %r" % message)
connection.close()