RocketMQ

RocketMQ 是阿里巴巴开源的分布式消息中间件，支持事务消息。核心概念模型与 Kafka 同构但有差异：Topic 是消息分类，MessageQueue 是并行分片，ConsumerGroup 是消费水平扩展单元，Tag 是 Topic 内的二级过滤：

概念	作用	与 Kafka 对照
Topic	消息主题，逻辑分类	同 Topic
MessageQueue	Topic 下的物理分片，并行度来源	Partition
ConsumerGroup	消费同一 Topic 的一组消费者，组内分摊队列	Consumer Group
Tag	Topic 内二级标签，消费时可过滤	无直接对应（用 key 过滤）
Broker	消息存储节点，主从复制	Broker
NameServer	路由注册中心，无状态、可多部署	ZooKeeper

消费模型

同一 ConsumerGroup 内多个消费者分摊队列（一个队列同一时刻只被组内一个消费者消费），不同 ConsumerGroup 各自独立消费同一份消息——这与 Kafka 完全一致：组内是竞争关系（伸缩吞吐），组间是广播关系（多业务订阅）。用 Tag 可以做精细过滤：订阅 topicA || tag1 || tag2 只收指定标签的消息。

底层存储采用 CommitLog + ConsumeQueue 双写结构：所有消息顺序写入单个 CommitLog（顺序写磁盘，性能高），再按 Topic/Queue 生成索引 ConsumeQueue 供消费定位——顺序写日志 + 异步建索引，是 RocketMQ 高吞吐的核心设计。

网络问题导致消费失败可重试（默认每条消息重试 16 次）
多线程异步消费失败 → 先执行本地事务再返回成功状态

docker run -d -p 9876:9876 --name rmqserver apache/rocketmq ./mqnamesrv

Broker

配置 broker.conf：

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
brokerIP1 = 192.168.2.252

docker run -d -p 10911:10911 -p 10909:10909 \
 --name rmqbroker --link rmqserver:namesrv \
 -e "NAMESRV_ADDR=namesrv:9876" \
 -v /docker/rocketmq/conf/broker.conf:/etc/rocketmq/broker.conf \
 apache/rocketmq ./mqbroker -c /etc/rocketmq/broker.conf

使用

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>

发送消息：

@Autowired
private RocketMQTemplate rocketMQTemplate;

@GetMapping("/test01")
public Result test01() {
    String str = RandomUtil.randomString(8);
    rocketMQTemplate.convertAndSend("test-topic", str);
    return Result.ok();
}

接收消息：

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-group")
public class RocketMQReceive implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("接收到消息：" + s);
    }
}

消息类型

类型	说明	适用场景
同步发送	等待响应	重要通知、报名短信
异步发送	回调接口接收响应	链路耗时较长的业务
单向发送	不等待应答	日志收集
顺序消息	严格按顺序发布和消费	订单处理
事务消息	分布式事务最终一致	支付系统

RocketMQ

消息丢失的场景

生产者发送时如何保证不丢失？

MQ 写入硬盘如何保证不丢失？

硬盘坏了如何保证不丢失？

消费者如何保证不丢失？

MQ 节点挂了如何保证不丢失？

安装

NameServer

Broker

使用

消息类型

常见问题处理

消费失败重试耗尽

消息重复消费