消息中间件----rabbitmq 远程连接

rabbitmq相关命令

# 前台启动服务（必须先启动服务）
rabbitmq-server start
# ctrl + c 关闭

# 后台启动
rabbitmq-server -detached

# 查看服务节点状态 (服务必须处于运行状态，前台 or 后台运行)
rabbitmqctl status
# 开启应用
rabbitmqctl start_app
# 添加用户
rabbitmqctl add_user tom tom123
# 设置用户为管理员
rabbitmqctl set_user_tags tom administrator 

# 配置当前的用户lauf  在'/'虚拟机 具有配置、读、写的权限 
rabbitmqctl set_permission -p '/' lauf ".*" ".*" ".*" 

# 查看所有用户
rabbitmqctl list_users

# 关闭应用，关闭后无法使用rabbitmqctl进行相关操作
rabbitmqctl stop_app

# 关闭rabbitmq-server (节点关闭)
rabbitmqctl shutdown

# 启动服务节点
rabbitmq-server -detached  # 后台启动
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

远程连接rabbitmq-server

安装好rabbitmq后，本地客户端连接正常，但是远程客户端无法连接。

解决：只需关闭防火墙；或者防火墙打开对应的端口。

# 关闭防火墙
ufw status
ufw disable
1
2
3

或者 添加端口规则

# 开启防火墙
ufw enable
# 添加端口访问规则
ufw allow 5672/tcp
ufw allow 15672/tcp

# 删除规则
ufw delete allow 15672/tcp

# 查看规则号 也可以通过规则号删除
ufw status numbered
# ufw delete 1
1
2
3
4
5
6
7
8
9
10
11
12

这样设置一下防火墙，就可以远程连接rabbitmq服务器了。

普通集群

准备服务节点

1. ubuntu1804: 192.168.0.113，rabbitmq节点1
2. ubuntu1804: 192.168.0.112，rabbitmq节点2

• 确保以上两个节点相互可以ping通（虚拟机桥接模式、防火墙关闭）
• 更改ubuntu的主机名

# 均使用如下方式，更改主机名字
vim /etc/hostname

# 编辑 
node1
1
2
3
4
5

更改主机名后，重启

• 更改 ip地址映射，通过域名可以解析出ip地址

# 如果权限不够，可以切换到超级管理员权限
sudo -s
# 编辑
vim /etc/hosts
1
2
3
4

在两个服务器上均编辑以上信息，保存后相互间就可以通过域名访问。

生产环境中，一般需要为静态ip地址。
 

服务器安装rabbitmq

• 安装erlang环境
  所有节点必须安装相同版本的erlang
   
• 安装rabbitmq服务
  所有节点安装相同版本的rabbitmq

在一个节点配置完成，克隆该节点即可，避免频繁的配置环境。
 

集群节点同步cookie

在linux的家目录下，可以找到.erlang.cookie :

两台linux保持该cookie相同，这里将node1的cookie复制到node2:

scp /home/laufing/.erlang.cookie laufing@node2:/home/laufing
1

注意：scp依赖ssh协议，需要防火墙开启22端口，一般服务器还会设置一些安全组，有安全组的情况下，安全组也要开启端口。

这里的集群节点防火墙均开启一下端口：
5672/tcp
15672/tcp
4369/tcp

# ubuntu下
ufw allow 5672/tcp
ufw allow 15672/tcp
ufw allow 4369/tcp
ufw allow 22
ufw enable 

# centos系统的防火墙操作参考
firewall-cmd --zone=public --add-port=5672/tcp --permanent
...
firewall-cmd --reload
# systemctl start firewalld
1
2
3
4
5
6
7
8
9
10
11
12

 

加入集群

• 先启动一个节点node1;

# 切换超级权限
sudo -s
# 后台启动rabbitmq-server  若已启动，则重启
rabbitmq-server -detached
# 查看状态
rabbitmqctl status
1
2
3
4
5
6

• 依次将其他节点加入集群

sudo -s
# 重启rabbitmq server   rabbitmqctl stop 关闭
rabbitmq-server -detached
# 查看状态
rabbitmqctl status

# 关闭应用
rabbitmqctl stop_app
# 重置
rabbitmqctl reset

# 加入集群
rabbitmqctl join_cluster rabbit@node1
# 启动应用
rabbitmqctl start_app
# 查看集群状态
rabbitmqctl cluster_status
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

 

创建用户

节点加入集群后，自己的用户等信息就不能再使用，需要重新创建用户。

# 在任意一个节点上操作
rabbitmqctl add_user lauf lauf
rabbitmqctl set_user_tags lauf administrator
rabbitmqctl set_permissions -p '/' lauf ".*" ".*" ".*"
rabbitmqctl list_users
1
2
3
4
5

创建的用户，所有节点共享。
可以在任意一个节点，进行连接。
如下，在node1节点连接node2节点：

 

python客户端测试

生产者，连接集群的任意一个节点。

# __author__ = "laufing"
from pika import PlainCredentials, BlockingConnection, ConnectionParameters, BasicProperties


# 连接
credentials = PlainCredentials("lauf", "lauf")
conn = BlockingConnection(ConnectionParameters(host="192.168.0.113", port=5672, credentials=credentials,
                                               virtual_host='/', heartbeat=0))
channel = conn.channel()


# 声明队列
channel.queue_declare(queue="q1")


# 确认每次接收一个消息
channel.basic_publish(exchange="", routing_key="q1", body="message1")
channel.basic_publish(exchange="", routing_key="q1", body="message2")
channel.basic_publish(exchange="", routing_key="q1", body="message3")
channel.basic_publish(exchange="", routing_key="q1", body="message4")

channel.close()
conn.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

消费者1，连接node1

# __author__ = "laufing"
from pika import PlainCredentials, BlockingConnection, ConnectionParameters, BasicProperties


# 连接
credentials = PlainCredentials("lauf", "lauf")
conn = BlockingConnection(ConnectionParameters(host="192.168.0.113", port=5672, credentials=credentials,
                                               virtual_host='/', heartbeat=0))
channel = conn.channel()


# 声明队列
channel.queue_declare(queue="q1")
# 声明交换机
# channel.exchange_declare(exchange="e1", exchange_type="direct") # 正则转发

# 队列绑定交换机
# channel.queue_bind(queue="q1", exchange="e2", routing_key="*.laufing") # 消费者端 正则匹配


# 定义消费程序
def func(ch, delivery, prop, body):
    print(ch is channel)
    print("消费的信息:", body)


# 确认每次接收一个消息
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="q1", on_message_callback=func, auto_ack=True)
channel.start_consuming()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

同样，消费者2连接node2，其他不变。

原理图：

node1宕机后，q1队列的消息丢失，不是高可用状态。

 

脱离集群

node2 脱离集群，在node2执行如下：

# 
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
rabbitmqctl cluster_status
1
2
3
4
5

在node1（最先启动的那个节点）执行：

rabbitmqctl forget_cluster_node rabbit@node2
1

若node1节点没启动，则加入一个--offline参数。