MQ是什麼

選型

RabitMq 搭建

安裝

docker pull rabiitmq:3.7-alpine

執行方式

幾個注意點:

• 設定hostname, 因為rabbimq資料存放依賴node name; 而預設的節點名稱是hostname;

$ docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3.7-alpine
# 為持久化可在啟動時增加以下兩引數
-v "rabbitmq_log:/var/log/rabbitmq" \
-v "rabbitmq_data:/var/lib/rabbitmq" \

預設監聽埠5672;

• 記憶體限制
  rabbitmq引數vm_memory_high_watermark, 由環境變數RABBITMQ_VM_MEMORY_HIGH_WATERMARK決定;

  –memory 2048m (and the implied upstream-default RABBITMQ_VM_MEMORY_HIGH_WATERMARK of 40%) will set the effective limit to 819MB (which is 40% of 2048MB).

• Erlang Cookie

  RabbitMQ nodes 和 CLI 命令列工具 (e.g. rabbitmqctl) 利用 cookie來決定是否允許互相通訊;

RABBITMQ_ERLANG_COOKIE 環境變數設定cookie
RABBITMQ_NODENAME 可反覆喚起rabbitmqctl

$ docker run -it --rm --link some-rabbit:my-rabbit -e RABBITMQ_ERLANG_COOKIE='secret cookie here' -e RABBITMQ_NODENAME=[email protected]-rabbit rabbitmq:3 bash

管理外掛

帶管理外掛的rabbitmq映象

安裝

docker pull rabbitmq:3.7-management-alpine

執行

$ docker run -d --hostname my-rabbit --name some-rabbit -p 8080:15672 rabbitmq:3.7-management-alpine

預設埠15672, 預設賬號密碼: guest / guest; 可通過RABBITMQ_DEFAULT_USER 和 RABBITMQ_DEFAULT_PASS修改；

配置

參考

• RABBITMQ_DEFAULT_VHOST　可修改vhost; vhost預設是’/’; 在多機系統時，可以有不同的vhost
• Hipe在alpine OS中有bug;
  

HiPE(High Performance Erlang) 霸爺有一個一語中的的描述”erlang的hipe相當於jit, 根據語言評測有hipe支援在純erlang的運算上會快2-3倍，這個效能的提升對於計算密集型的應用還是比較可觀的。”

外掛

通過docker file來檢視執行

FROM rabbitmq:3.7-alpine-management
RUN rabbitmq-plugins enable --offline rabbitmq_mqtt rabbitmq_federation_management rabbitmq_stomp

使用

啟動服務

docker run -d --hostname my-rabbit --name some-rabbit -p 15672:15672 rabbitmq:3.7-management-alpine
docker logs some-rabbit

 node           : [email protected]
 home dir       : /var/lib/rabbitmq
 config file(s) : /etc/rabbitmq/rabbitmq.conf
 cookie hash    : 9dc/v2egQn2ldwnTkb+DSg==
 log(s)         : <stdout>
 database dir   : /var/lib/rabbitmq/mnesia/[email protected]
2018-08-07 10:20:49.914 [info] <0.205.0> Memory high watermark set to 801 MiB (840700723 bytes) of 2004 MiB (2101751808 bytes) total
...
2018-08-07 10:20:49.925 [info] <0.197.0> Node database directory at /var/lib/rabbitmq/mnesia/[email protected] is empty. Assuming we need to join an existing cluster or initialise from scratch...
...
2018-08-07 10:20:50.201 [info] <0.197.0> Adding vhost '/'
2018-08-07 10:20:50.227 [info] <0.415.0> Making sure data directory '/var/lib/rabbitmq/mnesia/[email protected]/msg_stores/vhosts/628WB79CIFDYO9LJI6DKMI09L' for vhost '/' exists
2018-08-07 10:20:50.233 [info] <0.415.0> Starting message stores for vhost '/'
2018-08-07 10:20:50.233 [info] <0.419.0> Message store "628WB79CIFDYO9LJI6DKMI09L/msg_store_transient": using rabbit_msg_store_ets_index to provide index
2018-08-07 10:20:50.235 [info] <0.415.0> Started message store of type transient for vhost '/'
2018-08-07 10:20:50.236 [info] <0.422.0> Message store "628WB79CIFDYO9LJI6DKMI09L/msg_store_persistent": using rabbit_msg_store_ets_index to provide index
2018-08-07 10:20:50.237 [warning] <0.422.0> Message store "628WB79CIFDYO9LJI6DKMI09L/msg_store_persistent": rebuilding indices from scratch
2018-08-07 10:20:50.238 [info] <0.415.0> Started message store of type persistent for vhost '/'
2018-08-07 10:20:50.240 [info] <0.197.0> Creating user 'guest'
2018-08-07 10:20:50.247 [info] <0.197.0> Setting user tags for user 'guest' to [administrator]
2018-08-07 10:20:50.250 [info] <0.197.0> Setting permissions for 'guest' in '/' to '.*', '.*', '.*'
2018-08-07 10:20:50.256 [info] <0.460.0> started TCP Listener on 0.0.0.0:5672
2018-08-07 10:20:50.263 [info] <0.197.0> Setting up a table for connection tracking on this node: '[email protected]'
2018-08-07 10:20:50.269 [info] <0.197.0> Setting up a table for per-vhost connection counting on this node: '[email protected]'

python post/subscribe模型

需要Pika庫;

pip3 install pika

hello wold

publish端

# 建立連線
#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 建立queue
channel.queue_declare(queue='hello')

# 訊息不能直接傳送到queue，必須經過exchange; 
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

client端

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')  #可以多次呼叫，總是同一個queue (即idempotent)

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()

exchange繫結多個queue

send.py

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

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         exchange_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()

receive.py

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

channel.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()

AMQ 基本API

basic
basic.ack
basic.cancel
basic.consume
basic.deliver
basic.get
basic.nack
basic.publish
basic.qos
basic.recover
basic.recover-async
basic.reject
basic.return
channel 
channel.close
channel.flow
channel.open
confirm
confirm.select
exchange  路由/交換
exchange.bind
exchange.declare
exchange.delete
exchange.unbind
queue 佇列
queue.bind
queue.declare
queue.delete
queue.purge
queue.unbind
tx  事物
tx.commit
tx.rollback
tx.select

許可權

參考

RabbitMQ的使用者角色分類：
none、management、policymaker、monitoring、administrator
兩種資料;
* 定義類 Definitions (Topology)
定義類資料在內部資料庫儲存，並在所有叢集節點內自動同步. 當一個節點改變時，其他節點也會同步改變; 也即備份時，從任一節點備份皆可.
- 訊息資料 Messages data
訊息資料存在訊息store區域; 對使用者透明; 使用者也不需要關心;