当前位置：首页 > news >正文

消息队列--RocketMQ 架构设计与优化

news 2026/7/1 3:07:01

系列导读：本篇将深入讲解 RocketMQ 架构设计与生产环境最佳实践。

文章目录

- 一、RocketMQ 架构
- - 1.1 核心组件
  - 1.2 组件说明
  - 1.3 与 Kafka 对比
- 二、集群部署
- - 2.1 集群架构
  - 2.2 Docker 部署
  - 2.3 Broker 配置
- 三、生产者最佳实践
- - 3.1 配置
  - 3.2 发送消息
- 四、消费者最佳实践
- - 4.1 消费配置
  - 4.2 消费消息
- 五、事务消息
- - 5.1 事务消息流程
  - 5.2 实现代码
- 总结

一、RocketMQ 架构

1.1 核心组件

┌─────────────────────────────────────────────────────────────┐ │ RocketMQ 架构 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ Producer ──► NameServer ──► Broker ──► Consumer │ │ │ │ │ │ │ ▼ │ │ │ CommitLog │ │ │ ConsumeQueue │ │ │ │ │ └─── 路由注册 ───► Topic 路由信息 │ │ │ └─────────────────────────────────────────────────────────────┘

1.2 组件说明

组件	说明
NameServer	路由注册中心，轻量级
Broker	消息服务器，存储消息
Producer	消息生产者
Consumer	消息消费者

1.3 与 Kafka 对比

特性	RocketMQ	Kafka
吞吐量	十万级	百万级
延迟	ms级	ms级
事务消息	支持	不支持
延迟消息	支持	不支持
消息过滤	支持	不支持

二、集群部署

2.1 集群架构

┌─────────────────────────────────────────────────────────────┐ │ RocketMQ 集群 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ NameServer1 │ │ NameServer2 │ │ NameServer3 │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ │ │ │ └────────────────┼────────────────┘ │ │ │ │ │ ┌────────────────┼────────────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Broker-Master│ │ Broker-Master│ │ Broker-Master│ │ │ │ (主) │ │ (主) │ │ (主) │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │Broker-Slave │ │Broker-Slave │ │Broker-Slave │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘

2.2 Docker 部署

version:'3'services:namesrv:image:apache/rocketmq:5.1.0container_name:rmqnamesrvports:-"9876:9876"command:sh mqnamesrvbroker:image:apache/rocketmq:5.1.0container_name:rmqbrokerports:-"10911:10911"-"10909:10909"environment:NAMESRV_ADDR:"namesrv:9876"command:sh mqbroker-c /opt/rocketmq/conf/broker.confvolumes:-./broker.conf:/opt/rocketmq/conf/broker.conf-./logs:/opt/rocketmq/logs-./store:/opt/rocketmq/store

2.3 Broker 配置

# broker.conf brokerClusterName = DefaultCluster brokerName = broker-a brokerId = 0 namesrvAddr = 192.168.1.100:9876;192.168.1.101:9876 defaultTopicQueueNums = 4 autoCreateTopicEnable = true autoCreateSubscriptionGroup = true listenPort = 10911 deleteWhen = 04 fileReservedTime = 48 mapedFileSizeCommitLog = 1073741824 mapedFileSizeConsumeQueue = 300000 diskMaxUsedSpaceRatio = 88 storePathRootDir = /opt/rocketmq/store maxMessageSize = 65536

三、生产者最佳实践

3.1 配置

@ConfigurationpublicclassRocketMQConfig{@BeanpublicDefaultMQProducerproducer(){DefaultMQProducerproducer=newDefaultMQProducer("order-producer-group");producer.setNamesrvAddr("192.168.1.100:9876;192.168.1.101:9876");producer.setRetryTimesWhenSendFailed(3);producer.setSendMsgTimeout(3000);producer.setMaxMessageSize(4*1024*1024);// 4MBreturnproducer;}}

3.2 发送消息

@ServicepublicclassOrderProducer{@AutowiredprivateDefaultMQProducerproducer;// 同步发送publicSendResultsendSync(Orderorder)throwsException{Messagemessage=newMessage("order-topic","order-created",JSON.toJSONString(order).getBytes());returnproducer.send(message);}// 异步发送publicvoidsendAsync(Orderorder){Messagemessage=newMessage("order-topic","order-created",JSON.toJSONString(order).getBytes());producer.send(message,newSendCallback(){@OverridepublicvoidonSuccess(SendResultresult){log.info("发送成功: {}",result);}@OverridepublicvoidonException(Throwablee){log.error("发送失败",e);}});}// 延迟消息publicvoidsendDelay(Orderorder,intdelayLevel){Messagemessage=newMessage("order-topic",JSON.toJSONString(order).getBytes());message.setDelayTimeLevel(delayLevel);// 1-18producer.send(message);}}

四、消费者最佳实践

4.1 消费配置

@ConfigurationpublicclassRocketMQConsumerConfig{@BeanpublicDefaultMQPushConsumerconsumer(){DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("order-consumer-group");consumer.setNamesrvAddr("192.168.1.100:9876;192.168.1.101:9876");consumer.setConsumeThreadMin(10);consumer.setConsumeThreadMax(20);consumer.setConsumeMessageBatchMaxSize(10);returnconsumer;}}

4.2 消费消息

@ComponentpublicclassOrderConsumer{@PostConstructpublicvoidstart()throwsMQClientException{DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer("order-consumer-group");consumer.setNamesrvAddr("192.168.1.100:9876");consumer.subscribe("order-topic","order-created");consumer.registerMessageListener(newMessageListenerConcurrently(){@OverridepublicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeConcurrentlyContextcontext){for(MessageExtmsg:msgs){try{Orderorder=JSON.parseObject(newString(msg.getBody()),Order.class);processOrder(order);}catch(Exceptione){log.error("消费失败",e);returnConsumeConcurrentlyStatus.RECONSUME_LATER;}}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();}}

五、事务消息

5.1 事务消息流程

1. 发送半消息（Half Message） 2. 执行本地事务 3. 提交或回滚消息 4. 事务回查机制

5.2 实现代码

@ServicepublicclassOrderTransactionProducer{@AutowiredprivateTransactionMQProducerproducer;publicvoidsendTransactionMessage(Orderorder){Messagemessage=newMessage("order-topic",JSON.toJSONString(order).getBytes());producer.sendMessageInTransaction(message,order);}}// 事务监听器@ComponentpublicclassOrderTransactionListenerimplementsTransactionListener{@AutowiredprivateOrderServiceorderService;@OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){try{Orderorder=(Order)arg;orderService.createOrder(order);returnLocalTransactionState.COMMIT_MESSAGE;}catch(Exceptione){returnLocalTransactionState.ROLLBACK_MESSAGE;}}@OverridepublicLocalTransactionStatecheckLocalTransaction(MessageExtmsg){// 事务回查Orderorder=JSON.parseObject(newString(msg.getBody()),Order.class);if(orderService.exists(order.getId())){returnLocalTransactionState.COMMIT_MESSAGE;}returnLocalTransactionState.ROLLBACK_MESSAGE;}}