消息队列--RabbitMQ 高可用集群部署

系列导读本篇将深入讲解 RabbitMQ 高可用集群的部署与最佳实践。文章目录一、RabbitMQ 架构1.1 核心概念1.2 核心组件1.3 Exchange 类型二、集群部署2.1 集群架构2.2 Docker Compose 部署2.3 集群配置三、镜像队列3.1 镜像策略3.2 镜像模式四、生产者最佳实践4.1 配置4.2 发送消息五、消费者最佳实践5.1 消费配置5.2 消费消息总结一、RabbitMQ 架构1.1 核心概念┌─────────────────────────────────────────────────────────────┐ │ RabbitMQ 架构 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ Producer ──► Exchange ──► Queue ──► Consumer │ │ │ │ │ ▼ │ │ Binding │ │ (路由规则) │ │ │ └─────────────────────────────────────────────────────────────┘1.2 核心组件组件说明Exchange交换机接收消息并路由Queue队列存储消息Binding绑定Exchange 与 Queue 的关联Virtual Host虚拟主机隔离环境1.3 Exchange 类型类型说明Direct精确匹配路由键Topic通配符匹配Fanout广播到所有队列Headers根据消息头匹配二、集群部署2.1 集群架构┌─────────────────────────────────────────────────────────────┐ │ RabbitMQ 集群 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Node 1 │◄──►│ Node 2 │◄──►│ Node 3 │ │ │ │ (Disk) │ │ (RAM) │ │ (RAM) │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ 磁盘节点持久化集群状态 │ │ 内存节点高性能但不持久化 │ │ │ └─────────────────────────────────────────────────────────────┘2.2 Docker Compose 部署version:3services:rabbitmq1:image:rabbitmq:3-managementhostname:rabbitmq1ports:-5672:5672-15672:15672environment:RABBITMQ_ERLANG_COOKIE:secret_cookieRABBITMQ_NODENAME:rabbitrabbitmq1volumes:-./data1:/var/lib/rabbitmqrabbitmq2:image:rabbitmq:3-managementhostname:rabbitmq2ports:-5673:5672-15673:15672environment:RABBITMQ_ERLANG_COOKIE:secret_cookieRABBITMQ_NODENAME:rabbitrabbitmq2volumes:-./data2:/var/lib/rabbitmq2.3 集群配置# 在 Node 2 上加入集群dockerexecrabbitmq2 rabbitmqctl stop_appdockerexecrabbitmq2 rabbitmqctl join_cluster rabbitrabbitmq1dockerexecrabbitmq2 rabbitmqctl start_app# 查看集群状态dockerexecrabbitmq1 rabbitmqctl cluster_status三、镜像队列3.1 镜像策略# 设置镜像策略所有队列镜像到所有节点rabbitmqctl set_policy ha-all^{ha-mode:all,ha-sync-mode:automatic}# 设置镜像策略指定队列rabbitmqctl set_policy ha-order^order\.{ha-mode:all}3.2 镜像模式模式说明all镜像到所有节点exactly镜像到指定数量节点nodes镜像到指定节点四、生产者最佳实践4.1 配置ConfigurationpublicclassRabbitMQConfig{BeanpublicConnectionFactoryconnectionFactory(){CachingConnectionFactoryfactorynewCachingConnectionFactory();factory.setAddresses(192.168.1.100:5672,192.168.1.101:5672);factory.setUsername(admin);factory.setPassword(admin);factory.setVirtualHost(/order);returnfactory;}BeanpublicRabbitTemplaterabbitTemplate(){RabbitTemplatetemplatenewRabbitTemplate(connectionFactory());template.setMandatory(true);// 消息路由失败回调template.setConfirmCallback((correlationData,ack,cause)-{if(ack){log.info(消息发送成功);}else{log.error(消息发送失败: {},cause);}});returntemplate;}// 订单交换机BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange(order.exchange,true,false);}// 订单队列BeanpublicQueueorderQueue(){returnQueueBuilder.durable(order.queue).withArgument(x-message-ttl,86400000)// TTL.withArgument(x-dead-letter-exchange,dlx.exchange)// 死信.build();}// 绑定BeanpublicBindingorderBinding(){returnBindingBuilder.bind(orderQueue()).to(orderExchange()).with(order.created);}}4.2 发送消息ServicepublicclassOrderProducer{AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidsendOrder(Orderorder){// 发送消息rabbitTemplate.convertAndSend(order.exchange,order.created,JSON.toJSONString(order),message-{message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);returnmessage;});}}五、消费者最佳实践5.1 消费配置ConfigurationpublicclassRabbitConsumerConfig{BeanpublicSimpleRabbitListenerContainerFactoryrabbitListenerContainerFactory(){SimpleRabbitListenerContainerFactoryfactorynewSimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory());factory.setConcurrentConsumers(3);// 并发消费者数factory.setMaxConcurrentConsumers(10);factory.setPrefetchCount(10);// 预取数量factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手动确认returnfactory;}}5.2 消费消息ComponentpublicclassOrderConsumer{RabbitListener(queuesorder.queue)publicvoidconsume(Messagemessage,Channelchannel)throwsIOException{try{OrderorderJSON.parseObject(newString(message.getBody()),Order.class);// 业务处理processOrder(order);// 手动确认channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch(Exceptione){log.error(消费失败,e);// 拒绝并重新入队channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);}}}总结✅RabbitMQ 架构Exchange、Queue、Binding✅集群部署磁盘节点、内存节点✅镜像队列高可用配置✅生产者实践确认机制、持久化✅消费者实践手动确认、并发消费下篇预告RocketMQ 架构设计与优化作者刘~浪地球系列消息队列二更新时间2026-04-12