搭建 RabbitMQ 集群

RabbitMQ 搭建可以参考：https://www.cnblogs.com/klvchen/p/14029336.html

官网集群介绍：https://www.rabbitmq.com/clustering.html
测试架构

名称	IP
node-1	192.168.0.100
node-2	192.168.0.101

# 在 node-1, node-2 上操作,加入 hosts
vi /etc/hosts
192.168.0.100 node-1
192.168.0.101 node-2

# 在 node-2 上操作，关闭 rabbitmq 集群
rabbitmqctl stop
rabbitmqctl status 

# 在 node-1 上操作
cd 
cat .erlang.cookie 

# 在 node-2 上操作，需要与 node-1 的 .erlang.cookie 值相同
cd 
vi .erlang.cookie 
 
rabbitmq-server -detached

# 停止节点 rabbit@node-2 
rabbitmqctl stop_app

# 重置节点 rabbit@node-2 
rabbitmqctl reset

# 创建节点 rabbit@node-2 with [rabbit@node-1]
rabbitmqctl join_cluster rabbit@node-1

# 启动节点 rabbit@node-2
rabbitmqctl start_app

# 查看集群状态
rabbitmqctl cluster_status

# 删除主节点（node-1），在 node2 节点上操作
rabbitmqctl -n rabbit@node-2 forget_cluster_node rabbit@node-1

集群名称

集群名称默认是集群中第一个节点的名称，通过这个命令可以重新设置。
集群名称在客户端连接时会通报给客户端。Federation和Shovel插件也会有用到集群名称的地方。

# 设置集群名称
rabbitmqctl set_cluster_name cluster_klvchen

镜像模式, 官网：https://www.rabbitmq.com/rabbitmqctl.8.html#set_policy

rabbitmqctl set_policy -p / ha "^" '{"ha-mode":"all","ha-sync-mode":"automatic"}'

查看现有的规则

rabbitmqctl list_policies
Listing policies for vhost "/" ...
vhost   name    pattern apply-to        definition      priority
/       ha      ^       all     {"ha-mode":"all","ha-sync-mode":"automatic"}    0

取消现有的 ha 规则

rabbitmqctl clear_policy ha

删除节点（node-1），在 node-2 节点上操作

rabbitmqctl -n rabbit@node-2 forget_cluster_node rabbit@node-1

使用 python 代码测试

创建 publish.py 发送到 RabbitMQ

import pika
# socket连接
credentials = pika.PlainCredentials('admin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.0.101', 28003, credentials=credentials))
my_queue = "task_mess"

# 申明一个管道
channel = connection.channel()

# 给管道里面申明一个queue
channel.queue_declare(queue=my_queue, durable=True)

# 通过管道发送消息
for l in range(0, 101):
    msg = "OderID: %d"%l
    channel.basic_publish(exchange='', routing_key=my_queue, body=msg)

print("消息以及发送到subscribe端")
connection.close()

创建 subscribe.py 发送到 RabbitMQ

import pika
# socket连接
credentials = pika.PlainCredentials('admin', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.0.101', 28003, credentials=credentials))
my_queue = "task_mess"

# 申明一个管道
channel = connection.channel()

# 给管道里面申明一个queue
channel.queue_declare(queue=my_queue, durable=True)


def callback(ch, method, properties, body):
    print("subscribe端已经接收到消息正在处理~~~", body)


# 管道接收消息
channel.basic_consume(callback, queue=my_queue, no_ack=True)
channel.start_consuming()