ToB企服应用市场:ToB评测及商务社交产业平台

标题: RabbitMQ（集群干系部署） [打印本页]

---

作者: 大连密封材料    时间: 2024-8-30 16:19
标题: RabbitMQ（集群干系部署）
RabbitMQ

集群部署

环境预备：阿里云centos8 服务器，3台服务器，分别举行安装；
下载Erlang

Erlang和RabbitMQ版本对照：https://www.rabbitmq.com/which-erlang.html

• 创建yum库配置文件
  

1. vim /etc/yum.repos.d/rabbitmq.repo

复制代码

• 内容
  

1. # 内容 - 来自官方文档：https://www.rabbitmq.com/docs/install-rpm
2. # In /etc/yum.repos.d/rabbitmq.repo
3. ##
4. ## Zero dependency Erlang RPM
5. ##
6. [modern-erlang]
7. name=modern-erlang-el8
8. # uses a Cloudsmith mirror @ yum.novemberain.com in addition to its Cloudsmith upstream.
9. # Unlike Cloudsmith, the mirror does not have any traffic quotas
10. baseurl=https://yum1.novemberain.com/erlang/el/8/$basearch
11.         https://yum2.novemberain.com/erlang/el/8/$basearch
12.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/$basearch
13. repo_gpgcheck=1
14. enabled=1
15. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key
16. gpgcheck=1
17. sslverify=1
18. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
19. metadata_expire=300
20. pkg_gpgcheck=1
21. autorefresh=1
22. type=rpm-md
24. [modern-erlang-noarch]
25. name=modern-erlang-el8-noarch
26. # uses a Cloudsmith mirror @ yum.novemberain.com.
27. # Unlike Cloudsmith, it does not have any traffic quotas
28. baseurl=https://yum1.novemberain.com/erlang/el/8/noarch
29.         https://yum2.novemberain.com/erlang/el/8/noarch
30.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/noarch
31. repo_gpgcheck=1
32. enabled=1
33. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key
34.        https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
35. gpgcheck=1
36. sslverify=1
37. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
38. metadata_expire=300
39. pkg_gpgcheck=1
40. autorefresh=1
41. type=rpm-md
43. [modern-erlang-source]
44. name=modern-erlang-el8-source
45. # uses a Cloudsmith mirror @ yum.novemberain.com.
46. # Unlike Cloudsmith, it does not have any traffic quotas
47. baseurl=https://yum1.novemberain.com/erlang/el/8/SRPMS
48.         https://yum2.novemberain.com/erlang/el/8/SRPMS
49.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/SRPMS
50. repo_gpgcheck=1
51. enabled=1
52. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key
53.        https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
54. gpgcheck=1
55. sslverify=1
56. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
57. metadata_expire=300
58. pkg_gpgcheck=1
59. autorefresh=1
62. ##
63. ## RabbitMQ Server
64. ##
66. [rabbitmq-el8]
67. name=rabbitmq-el8
68. baseurl=https://yum2.novemberain.com/rabbitmq/el/8/$basearch
69.         https://yum1.novemberain.com/rabbitmq/el/8/$basearch
70.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/$basearch
71. repo_gpgcheck=1
72. enabled=1
73. # Cloudsmith's repository key and RabbitMQ package signing key
74. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key
75.        https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
76. gpgcheck=1
77. sslverify=1
78. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
79. metadata_expire=300
80. pkg_gpgcheck=1
81. autorefresh=1
82. type=rpm-md
84. [rabbitmq-el8-noarch]
85. name=rabbitmq-el8-noarch
86. baseurl=https://yum2.novemberain.com/rabbitmq/el/8/noarch
87.         https://yum1.novemberain.com/rabbitmq/el/8/noarch
88.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/noarch
89. repo_gpgcheck=1
90. enabled=1
91. # Cloudsmith's repository key and RabbitMQ package signing key
92. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key
93.        https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
94. gpgcheck=1
95. sslverify=1
96. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
97. metadata_expire=300
98. pkg_gpgcheck=1
99. autorefresh=1
100. type=rpm-md
102. [rabbitmq-el8-source]
103. name=rabbitmq-el8-source
104. baseurl=https://yum2.novemberain.com/rabbitmq/el/8/SRPMS
105.         https://yum1.novemberain.com/rabbitmq/el/8/SRPMS
106.         https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/SRPMS
107. repo_gpgcheck=1
108. enabled=1
109. gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key
110. gpgcheck=0
111. sslverify=1
112. sslcacert=/etc/pki/tls/certs/ca-bundle.crt
113. metadata_expire=300
114. pkg_gpgcheck=1
115. autorefresh=1
116. type=rpm-md

复制代码

• 更新yum库
  

1. # --nobest表示所需安装包即使不是最佳选择也接受
2. yum update -y --nobest
4. # 还需关闭防火墙
5. systemctl stop firewalld
6. # 开机关闭防火墙：
7. systemctl disable firewalld
8. # 查看防火墙状态：
9. systemctl status firewalld

复制代码

• 举行安装
  

1. yum install -y erlang

复制代码

安装RabbitMQ

Erlang和RabbitMQ版本对照：https://www.rabbitmq.com/which-erlang.html

1. # 导入GPG密钥，如失败多重试几次
2. rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc'
3. rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key'
4. rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key'
6. # 下载 RPM 包
7. wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.13.0/rabbitmq-server-3.13.0-1.el8.noarch.rpm
9. # 安装
10. rpm -ivh rabbitmq-server-3.13.0-1.el8.noarch.rpm

复制代码

举行配置

1. # 启用管理界面插件
2. rabbitmq-plugins enable rabbitmq_management
4. # 启动 RabbitMQ 服务：
5. systemctl start rabbitmq-server
7. # 将 RabbitMQ 服务设置为开机自动启动
8. systemctl enable rabbitmq-server
10. # 新增登录账号密码 - 用户名 密码
11. rabbitmqctl add_user yan 123456
13. # 设置登录账号权限，该用户设置角色是administrator管理员。再设置权限
14. rabbitmqctl set_user_tags yan administrator
15. rabbitmqctl set_permissions -p / yan ".*" ".*" ".*"
17. # 查询当前用户和角色
18. rabbitmqctl list_users
20. # 配置所有稳定功能 flag 启用
21. rabbitmqctl enable_feature_flag all
23. # 重启RabbitMQ服务生效
24. systemctl restart rabbitmq-server
26. # 启动rabbitmq服务器
27. systemctl start rabbitmq-server
29. # 查看rabbitmq服务器状态
30. systemctl status rabbitmq-server
32. # 停止rabbitmq服务器
33. systemctl stop rabbitmq-server
35. # 重启rabbitmq服务器
36. systemctl restart rabbitmq-server

复制代码

首尾配置

1. # 删除yum库配置文件，避免下次yum安装在执行一遍
2. rm -rf /etc/yum.repos.d/rabbitmq.repo
4. # 分别在不同服务器修改主机名称
5. vim /etc/hostname
6. # 分别起名 node01、node02、node03
7. # 例子：机器1，该文件/etc/hostname 下 填写node1
8. #      机器2，该文件/etc/hostname 下 填写node2
9. #      机器3，该文件/etc/hostname 下 填写node3

复制代码

集群配置

机器1配置：

• 分别在3台机器设置IP 所在到对应主机名称的映射
  修改文件/etc/hosts，追加如下内容：
  
  1. ip1 node01
  2. ip2 node02
  3. ip3 node03

  复制代码
• 检察该MQ的Cookie值并记录
  
  1. cat /var/lib/rabbitmq/.erlang.cookie
  2. # 并记录该值，后续机器2,机器3上的cookie值保持一致
  3. # 在node01节点运行，把该文件复制到其他机器上
  4. scp /var/lib/rabbitmq/.erlang.cookie root@node02:/var/lib/rabbitmq/.erlang.cookie
  5. scp /var/lib/rabbitmq/.erlang.cookie root@node03:/var/lib/rabbitmq/.erlang.cookie
  6. # 需要将rabbitmq分别重启
  7. systemctl stop rabbitmq-server
  8. systemctl start rabbitmq-server  

  复制代码
• 重置节点应用
  
  1. rabbitmqctl stop_app
  2. rabbitmqctl reset
  3. rabbitmqctl start_app

  复制代码

机器2配置：

• 分别在3台机器设置IP 所在到对应主机名称的映射
  修改文件/etc/hosts，追加如下内容：
  
  1. ip1 node01
  2. ip2 node02
  3. ip3 node03

  复制代码
• 检察该MQ的Cookie值并记录
  
  1. cat /var/lib/rabbitmq/.erlang.cookie
  2. # 把机器1上的cookie值复制到这里，与机器1保持一致

  复制代码
• 重置节点应用，并加入集群
  
  1. rabbitmqctl stop_app
  2. rabbitmqctl reset
  3. # 加入集群，去node01机器上找rabbit的程序，进行加入
  4. rabbitmqctl join_cluster rabbit@node01
  5. rabbitmqctl start_app

  复制代码

机器3配置：

• 分别在3台机器设置IP 所在到对应主机名称的映射
  修改文件/etc/hosts，追加如下内容：
  
  1. ip1 node01
  2. ip2 node02
  3. ip3 node03

  复制代码
• 检察该MQ的Cookie值并记录
  
  1. cat /var/lib/rabbitmq/.erlang.cookie
  2. # 把机器1上的cookie值复制到这里，与机器1保持一致

  复制代码
• 重置节点应用，并加入集群
  
  1. rabbitmqctl stop_app
  2. rabbitmqctl reset
  3. # 加入集群，去node01机器上找rabbit的程序，进行加入
  4. rabbitmqctl join_cluster rabbit@node01
  5. rabbitmqctl start_app

  复制代码
• 检察集群状态
  
  1. rabbitmqctl cluster_status
  2. # 也可以在图形化界面上，查看集群节点，在Nodes这里查看集群节点

  复制代码

负载均衡

一般MQ的服务端口5672、UI管理界面端口15672
浏览器访问，某个端口，通过HAProxy工具，举行分流每个节点的15672端口上，举行管理UI上的负载均衡；
服务端访问，某个端口，通过HAProxy工具，举行分流每个节点的5672端口上，举行服务端的负载均衡；
安装HAProxy

1. yum install -y haproxy
2. # 查看安装的版本
3. haproxy -v
4. # 启动该程序
5. systemctl start haproxy
6. # 设置该服务开机自动启动
7. systemctl enable haproxy
8. # 查看状态
9. systemctl status haproxy
10. # 停止启动
11. systemctl stop haproxy
12. # 重启
13. systemctl restart haproxy
14. # 查看当前的线程
15. ps -ef|grep haproxy

复制代码

修改配置文件

• 修改配置文件
  
  1. vim /etc/haproxy/haproxy.cfg

  复制代码
  文件内容：
  
  1. # 在最后一行添加该内容，该部分配置UI管理器上的负载均衡
  2. # 起MQ的UI的名字（rabbitmq_ui_frontend），前端部分配置
  3. frontend rabbitmq_ui_frontend
  4. # 绑定一个地址，端口号，外部浏览器访问该IP地址和端口后，然后通过HAProxy进行转发
  5. bind 192.168.xxx.xxx:22222
  6. # http协议
  7. mode http
  8. # 默认的后端服务名，使用下面的名字
  9. default_backend rabbitmq_ui_backend
  11. # 起MQ的服务端的名字（rabbitmq_ui_backend）
  12. backend rabbitmq_ui_backend
  13. # http协议
  14. mode http
  15. # 启用轮询的负载均衡
  16. balance roundrobin
  17. option httpchk GET /
  18. # 分别设置，不同的MQ服务节点的IP地址
  19. server rabbitmq_ui1 192.168.xxx.1:15672 check
  20. server rabbitmq_ui2 192.168.xxx.2:15672 check
  21. server rabbitmq_ui3 192.168.xxx.3:15672 check
  23. # 配置服务端（生产者、消费者）的负载均衡
  24. # 服务端（生产者、消费者）的名字
  25. frontend rabbitmq_frontend
  26. # 绑定一个地址，端口号，服务端（生产者、消费者）访问该IP地址和端口后，然后通过HAProxy进行转发
  27. bind 192.168.xxx.xxx:11111
  28. # tcp协议
  29. mode tcp
  30. default_backend rabbitmq_backend
  31. # 与上面的名字保持一致
  32. backend rabbitmq_backend
  33. # tcp协议
  34. mode tcp
  35. # 启用轮询的负载均衡
  36. balance roundrobin
  37. # 分别设置，不同的MQ服务节点的IP地址
  38. server rabbitmq1 192.168.xxx.1:5672 check
  39. server rabbitmq2 192.168.xxx.2:5672 check
  40. server rabbitmq3 192.168.xxx.3:5672 check

  复制代码
• 设置SELinux策略，答应HAProxy拥有权限连接任意端口：
  
  1. # 保存上面的配置文件，执行下面
  2. setsebool -P haproxy_connect_any=1
  4. # 绑定非本机的IP需要在sysctl.conf文件中配置：
  5. vi   /etc/sysctl.conf
  6. # 添加
  7. net.ipv4.ip_nonlocal_bind=1
  8. # 保存退出，执行
  9. sysctl -p

  复制代码
  SELinux是Linux系统中的安全模块，它可以限定进程的权限以进步系统的安全性。在某些环境下，SELinux可能会阻止HAProxy绑定指定的端口，这就必要通过设置域（domain）的安全策略来解决此问题。通过执行setsebool -P haproxy_connect_any=1命令，您已经为HAProxy设置了一个布尔值，答应HAProxy连接到任意端口。如许，HAProxy就可以成功绑定指定的socket，并正常工作。
  
• 重启HAProxy
  
  1. # 测试配置文件的正确性
  2. haproxy -c -f /etc/haproxy/haproxy.cfg
  3. # 进行重启
  4. systemctl restart haproxy

  复制代码

配置成功

1. # 访问MQ管理UI页面：
2. http://192.168.xxx.xxx:22222
3. # 生产者、消费者服务连接MQ
4.   rabbitmq:
5.     host: 192.168.xxx.xxx
6.     port: 11111
7.     username: guest
8.     password: 123456
9. # 可通过生产者发送消息，监听消费者是否接受到消息

复制代码

仲裁队列

仲裁队列是3.8版本以后才有的新功能，是主从模式，支持主从数据同步，在添加队列，队列类型为Quorum就是仲裁队列。
创建交换机
与之前创建平凡交换机同等
创建仲裁队列

绑定交换机
与之前的同等
验证，停止某个节点

1. # 停止rabbit应用
2. rabbitmqctl stop_app
3. # 我们在某个节点进行停止MQ，然后生产者发送消息，查看消费者是否可以接受到，判断集群的数据一致性
4. # 默认，主节点挂掉，默认重新选举新的主节点

复制代码

流式队列

流式队列是3.9版本以后才有的新功能，每个消息都分配一个偏移量，该消息被消耗掉也不会被删除，可以重复消耗；类似Kafka，但没有到达Kafka的效果。
前提预备 - 安装干系插件

1. # 启用Stream插件
2. rabbitmq-plugins enable rabbitmq_stream
4. # 重启rabbit应用
5. rabbitmqctl stop_app
6. rabbitmqctl start_app
8. # 查看插件状态
9. rabbitmq-plugins list

复制代码

负载均衡配置

1. frontend rabbitmq_stream_frontend
2. bind 192.168.xxx.100:33333
3. mode tcp
4. default_backend rabbitmq_stream_backend
6. backend rabbitmq_stream_backend
7. mode tcp
8. balance roundrobin
9. # 它的端口是5552
10. server rabbitmq1 192.168.xxx.1:5552 check
11. server rabbitmq2 192.168.xxx.1:5552 check
12. server rabbitmq3 192.168.xxx.2:5552 check

复制代码

JAVA干系代码

1. // 引包
2. <dependencies>
3.     <dependency>
4.         <groupId>com.rabbitmq</groupId>
5.         <artifactId>stream-client</artifactId>
6.         <version>0.15.0</version>
7.     </dependency>
9.     <dependency>
10.         <groupId>org.slf4j</groupId>
11.         <artifactId>slf4j-api</artifactId>
12.         <version>1.7.30</version>
13.     </dependency>
14.     <dependency>
15.         <groupId>ch.qos.logback</groupId>
16.         <artifactId>logback-classic</artifactId>
17.         <version>1.2.3</version>
18.     </dependency>
19. </dependencies>

复制代码

创建Stream，不必要创建交换机

1. // 代码创建
2. Environment environment = Environment.builder()
3.         .host("192.168.xxx.1")
4.         .port(33333)
5.         .username("用户名")
6.         .password("密码")
7.         .build();
9. environment.streamCreator().stream("stream.atguigu.test2").create();
10. environment.close();

复制代码

通过管理UI页面创建的话，队列类型选择：Stream
生产者

1. Environment environment = Environment.builder()
2.         .host("192.168.xxx.1")
3.         .port(33333)
4.         .username("用户名")
5.         .password("密码")
6.         .build();
8. Producer producer = environment.producerBuilder()
9.         // 队列名称
10.         .stream("stream.test")
11.         .build();
12. // 消息内容
13. byte[] messagePayload = "hello rabbit stream".getBytes(StandardCharsets.UTF_8);
15. CountDownLatch countDownLatch = new CountDownLatch(1);
16. producer.send(
17.         producer.messageBuilder().addData(messagePayload).build(),
18.         confirmationStatus -> {
19.             if (confirmationStatus.isConfirmed()) {
20.                 System.out.println("[生产者端]the message made it to the broker");
21.             } else {
22.                 System.out.println("[生产者端]the message did not make it to the broker");
23.             }
24.             countDownLatch.countDown();
25.         });
27. countDownLatch.await();
28. producer.close();
29. environment.close();

复制代码

消耗者

1. Environment environment = Environment.builder()
2.         .host("192.168.xxx.1")
3.         .port(33333)
4.         .username("用户名")
5.         .password("密码")
6.         .build();
7. environment.consumerBuilder()
8.             // 队列名称
9.         .stream("stream.test")
10.             // 这个是消费端起名字
11.         .name("stream.test.consumer")
12.         .autoTrackingStrategy()
13.         .builder()
14.         .messageHandler((offset, message) -> {
15.             // 接受消息
16.             byte[] bodyAsBinary = message.getBodyAsBinary();
17.             String messageContent = new String(bodyAsBinary);
18.             System.out.println("[消费者端]messageContent = " + messageContent + " Offset=" + offset.offset());
19.         })
20.         .build();

复制代码

指定偏移量消耗

1. // 指定Offset消费
2. Environment environment = Environment.builder()
3.         .host("192.168.xxx.1")
4.         .port(33333)
5.         .username("用户名")
6.         .password("密码")
7.         .build();
9. CountDownLatch countDownLatch = new CountDownLatch(1);
10. Consumer consumer = environment.consumerBuilder()
11.         .stream("stream.test")
12.         .offset(OffsetSpecification.first())
13.         .messageHandler((offset, message) -> {
14.             byte[] bodyAsBinary = message.getBodyAsBinary();
15.             String messageContent = new String(bodyAsBinary);
16.             System.out.println("[消费者端]messageContent = " + messageContent);
17.             countDownLatch.countDown();
18.         })
19.         .build();
21. countDownLatch.await();
22. consumer.close();

复制代码

Federation插件

它是MQ在差别的Broker节点之间举行消息通报而无须建立集群，可以实现跨集群的数据同步。
Federation交换机

• 前提预备：启用该插件、分为上游和下游，创建2个独立的MQ服务；
  
  1. # 使用docker进行安装MQ1服务
  2. docker run -d \
  3. --name rabbitmq-shenzhen \
  4. -p 51000:5672 \
  5. -p 52000:15672 \
  6. -v rabbitmq-plugin:/plugins \
  7. -e RABBITMQ_DEFAULT_USER=guest \
  8. -e RABBITMQ_DEFAULT_PASS=123456 \
  9. rabbitmq:3.13-management
  11. # 使用docker进行安装MQ2服务
  12. docker run -d \
  13. --name rabbitmq-shanghai \
  14. -p 61000:5672 \
  15. -p 62000:15672 \
  16. -v rabbitmq-plugin:/plugins \
  17. -e RABBITMQ_DEFAULT_USER=guest \
  18. -e RABBITMQ_DEFAULT_PASS=123456 \
  19. rabbitmq:3.13-management

  复制代码
• 启用插件
  
  1. # 分别在2台MQ服务，进行安装该插件
  2. rabbitmq-plugins enable rabbitmq_federation
  3. rabbitmq-plugins enable rabbitmq_federation_management

  复制代码
  安装完插件，在右侧菜单栏可看到
  
• 添加上游连接端点
  在下游的MQ服务，举行操纵，点击右侧Federation Upstreams，举行添加，必要起一个上游的连接点名字，（federation-upstreams），URL填写：amqp://用户名:暗码@访问ip:端口 （是上游的可访问的所在）
  
  
  
  
• 创建控制策略
  举行配置，在下游的MQ服务，举行操纵
  
  
  
  
• 验证
  
  1. # 需要满足：
  2. # 普通交换机和联邦交换机名称要一致
  3. # 交换机名称要能够和策略正则表达式匹配上
  4. # 发送消息时，两边使用的路由键也要一致
  5. # 队列名称不要求一致

  复制代码
  
  
  生产者（上游）发送消息、消息会被上游（MQ）的队列接收到，然后下游（MQ）的队列也会担当到消息；
  满足差别集群举行数据同步；
  

Federation队列

Federation队列和Federation交换机的最焦点区别就是：

• Federation Police作用在交换机上，就是Federation交换机
  
• Federation Police作用在队列上，就是Federation队列
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4