RabbitMQ集群搭建（二）

简介本文介绍如何搭建 RabbitMQ 集群使用 RabbitMQ-3.8.34 版本相关博客RabbitMQ集群搭建环境规划服务器清单如下我在本地创建了如下三台虚拟机IP 地址节点名称说明192.168.249.137node1主节点集群入口192.168.249.138node2从节点192.168.249.139node3从节点服务器环境示意集群架构共 3 个 RabbitMQ 节点采用镜像队列 / Quorum 队列实现高可用采用奇数节点数 3 节点避免脑裂问题Quorum 队列基于 Raft 算法至少需要 2 个节点在线才能正常工作环境准备Docker 环境所有服务器需提前安装 Docker 环境确认 Docker 服务正常运行systemctl status docker端口开放说明RabbitMQ 集群需要开放以下端口4369epmd节点发现服务5672AMQP 协议客户端连接端口15672管理界面 Web 端口25672节点间通信端口搭建1添加节点信息在三台服务器上执行以下命令添加节点 hosts 映射保证节点间可通过名称互相访问cat/etc/hostsEOF 192.168.249.137 node1 192.168.249.138 node2 192.168.249.139 node3 EOF配置完成示意2创建数据目录为保证容器删除后数据不丢失建议挂载数据卷# 在三台服务器分别执行mkdir-p/data/rabbitmq3创建 RabbitMQ 容器重要说明所有节点必须使用相同的 Erlang Cookie才能加入同一集群。节点1192.168.249.137连接 192.168.249.137 执行下面的命令dockerrun-d\--hostnamenode1\--namerabbitmq\--nethost\--restartalways\-v/data/rabbitmq:/var/lib/rabbitmq\-eRABBITMQ_ERLANG_COOKIErabbitmq_cookie_123\-eRABBITMQ_DEFAULT_USERadmin\-eRABBITMQ_DEFAULT_PASSadmin\rabbitmq:3.8.34-management启动完成示意节点2192.168.249.138连接 192.168.249.138 执行下面的命令dockerrun-d\--hostnamenode2\--namerabbitmq\--nethost\--restartalways\-v/data/rabbitmq:/var/lib/rabbitmq\-eRABBITMQ_ERLANG_COOKIErabbitmq_cookie_123\-eRABBITMQ_DEFAULT_USERadmin\-eRABBITMQ_DEFAULT_PASSadmin\rabbitmq:3.8.34-management启动完成示意节点 3192.168.249.139连接 192.168.249.139 执行下面的命令dockerrun-d\--hostnamenode3\--namerabbitmq\--nethost\--restartalways\-v/data/rabbitmq:/var/lib/rabbitmq\-eRABBITMQ_ERLANG_COOKIErabbitmq_cookie_123\-eRABBITMQ_DEFAULT_USERadmin\-eRABBITMQ_DEFAULT_PASSadmin\rabbitmq:3.8.34-management启动完成示意4组建集群以 node1 为主节点将 node2 和 node3 加入到 node1 组成的集群中连接 192.168.249.138 执行下面的命令将节点 2 加入集群# 停止节点应用dockerexecrabbitmq rabbitmqctl stop_app# 重置节点状态dockerexecrabbitmq rabbitmqctl reset# 加入到node1集群dockerexecrabbitmq rabbitmqctl join_cluster rabbitnode1# 重新启动应用dockerexecrabbitmq rabbitmqctl start_app加入完成示意连接 192.168.249.139 执行下面的命令将节点3加入集群dockerexecrabbitmq rabbitmqctl stop_appdockerexecrabbitmq rabbitmqctl resetdockerexecrabbitmq rabbitmqctl join_cluster rabbitnode1dockerexecrabbitmq rabbitmqctl start_app加入完成示意验证1 查看集群状态连接任意一个节点执行以下命令查看集群状态dockerexecrabbitmq rabbitmqctl cluster_status成功标志三个节点都显示在running_nodes列表中集群状态示意2管理控制台验证通过浏览器访问任意节点的管理后台使用账号密码admin/admin登录node1: http://192.168.249.137:15672node2: http://192.168.249.138:15672node3: http://192.168.249.139:15672登录后在 “Nodes” 页面可以看到所有三个节点状态。控制台示意3 高可用验证Quorum 队列RabbitMQ 3.8 推荐使用Quorum 队列替代传统镜像队列特性普通队列镜像队列Quorum 队列高可用❌ 单点✅ 主从复制✅ Raft算法数据一致性-最终一致强一致脑裂问题-可能发生不会发生最小节点数123建议奇数4故障转移测试# 1. 停掉一个节点dockerstop rabbitmq# 2. 在其他节点查看集群状态确认队列仍可用dockerexecrabbitmq rabbitmqctl cluster_status# 3. 重启节点观察是否自动重新加入集群dockerstart rabbitmq项目整合1 配置文件修改原单机配置spring:rabbitmq:host:${rabbitmq.ip}port:${rabbitmq.port}username:${rabbitmq.username}password:${rabbitmq.password}template:reply-timeout:30000listener:simple:missing-queues-fatal:false修改为集群配置配置所有节点地址spring:rabbitmq:addresses:192.168.249.137:5672,192.168.249.138:5672,192.168.249.139:5672username:${rabbitmq.username}password:${rabbitmq.password}template:reply-timeout:30000listener:simple:missing-queues-fatal:false配置建议将全部 3 个节点都写入配置作为集群入口。只配置少数节点时若这些入口节点同时宕机即使集群本身仍可用应用也无法连接。2Java 代码修改a. ConnectionFactory Bean 初始化如果你的项目中使用了手动初始化 RabbitMQ 的连接工厂需要做以下修改原代码Value(${rabbitmq.ip})privateStringhost;Value(${rabbitmq.port})privateintport;Value(${rabbitmq.username})privateStringusername;Value(${rabbitmq.password})privateStringpassword;Value(${rabbitmq.cache_size})privateintcacheSize;BeanpublicConnectionFactoryrabbitMQConnectionFactory(){CachingConnectionFactoryconnectionFactorynewCachingConnectionFactory();connectionFactory.getRabbitConnectionFactory().setRequestedChannelMax(10240);connectionFactory.setHost(host);connectionFactory.setPort(port);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setChannelCacheSize(cacheSize);connectionFactory.setChannelCheckoutTimeout(2000);connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);connectionFactory.getRabbitConnectionFactory().setNetworkRecoveryInterval(10000);connectionFactory.getRabbitConnectionFactory().setTopologyRecoveryEnabled(true);returnconnectionFactory;}修改后改用集群地址配置Value(${spring.rabbitmq.addresses})privateStringaddresses;Value(${spring.rabbitmq.username})privateStringusername;Value(${spring.rabbitmq.password})privateStringpassword;Value(${rabbitmq.cache_size})privateintcacheSize;BeanpublicConnectionFactoryrabbitMQConnectionFactory(){CachingConnectionFactoryconnectionFactorynewCachingConnectionFactory();connectionFactory.getRabbitConnectionFactory().setRequestedChannelMax(10240);connectionFactory.setAddresses(addresses);// 使用 addresses 替代 hostconnectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setChannelCacheSize(cacheSize);connectionFactory.setChannelCheckoutTimeout(2000);connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);connectionFactory.getRabbitConnectionFactory().setNetworkRecoveryInterval(10000);connectionFactory.getRabbitConnectionFactory().setTopologyRecoveryEnabled(true);returnconnectionFactory;}b. 创建队列非注解方式- 使用 Quorum 队列原代码BeanpublicQueuetestQueue(){returnnewQueue(testQueue,true);}修改为Quorum队列保证高可用BeanpublicQueuetestQueue(){returnQueueBuilder.durable(testQueue).quorum()// 声明为Quorum队列.build();}c. 创建队列注解方式- 使用 Quorum 队列原代码RabbitListener(bindingsQueueBinding(valueQueue(valuetestQueue,autoDeletefalse),exchangeExchange(valuetestExchange,typeExchangeTypes.TOPIC)),containerFactoryrabbitMQConnectionFactory)修改后增加x-queue-type参数声明为 Quorum 队列RabbitListener(bindingsQueueBinding(valueQueue(valuetestQueue,autoDeletefalse,argumentsArgument(namex-queue-type,valuequorum)),exchangeExchange(valuetestExchange,typeExchangeTypes.TOPIC)),containerFactoryrabbitMQConnectionFactory)注意事项Erlang Cookie一致所有节点的 Cookie 必须完全相同否则无法加入集群主机名解析确保所有节点能通过 hostname 互相访问/etc/hosts配置必须一致Quorum 队列节点数Quorum 队列基于 Raft 算法至少需要(N/2)1个节点在线才能写入。3 节点集群最多允许 1 个节点宕机数据持久化生产环境务必挂载数据卷防止容器删除导致数据丢失防火墙配置开放 4369、5672、15672、25672 四个端口密码安全生产环境建议修改默认的 admin/admin 密码监控告警建议监控队列堆积、节点状态、连接数等指标