图解RocketMQ之初识MQ

消息队列（Message Queue）是分布式系统中常用的一种解耦、异步、扩展性扩展架构模式。RocketMQ是阿里巴巴开源的分布式消息中间件，具备高吞吐量、单机万级吞吐量等特点，是阿里内部广泛使用的消息中间件。

1. 什么是消息队列？

消息队列（MQ）是一种应用程序到应用程序的通信方法。利用异步消息传递，消息队列可以在应用程序之间传递数据，而无需直接耦合。消息队列可以帮助应用解耦，允许执行密集型的工作进行异步处理。

典型场景

订单与库存：下单成功后异步通知库存系统扣减
日志收集：多个服务将日志写入MQ，统一由Logstash处理
异步通知：支付成功后异步通知用户

2. RocketMQ 核心概念

2.1 基本概念

概念	说明
Topic	消息主题，消息的物理目的地
Producer	消息生产者，负责发送消息
Consumer	消息消费者，负责接收消息
Broker	消息代理，存储消息并接收生产者和消费者的连接
NameServer	轻量级注册中心，管理Broker信息

2.2 架构设计

graph TD
    A[客户端] --> B[Producer]
    A --> C[Consumer]
    B --> D[Broker]
    C --> D
    D --> E[NameServer]
    E --> D

3. RocketMQ 消息模型

3.1 生产消息

DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
producer.start();

Message msg = new Message("TopicTest", "TagA", "Hello World".getBytes());
producer.send(msg);

3.2 消费消息

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
consumer.subscribe("TopicTest", "*");
consumer.registerMessageListener((messages, context) -> {
    System.out.println(messages);
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

4. RocketMQ 消息过滤

4.1 订阅策略

订阅表达式：TagA || TagB
SQL92：color = 'red' AND price > 10

4.2 实现方式

consumer.subscribe("TopicTest", "TagA || TagB");

5. 消息顺序性

5.1 顺序消息

分区顺序：同一分区内保证顺序
全局顺序：所有消息严格有序

// 生产者
Message msg = new Message("TopicTest", "TagA", msgBody);
// 设置Key
msg.setKeys("key1", "key2");

// 消费者
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

6. RocketMQ 特性总结

6.1 主要优势

高吞吐量：单机支持万级吞吐量
可靠性：支持消息持久化、消息回溯
扩展性：支持集群部署
多语言支持：支持 Java、Python、Go、Node.js 等

6.2 与 Kafka 对比

特性	RocketMQ	Kafka
消息顺序	支持	支持
消息过滤	支持	不支持
消息回溯	支持	支持
开发语言	纯Java	纯Java

7. 最佳实践

合理设置 Topic 和 Tag
使用 Key 进行消息去重
合理设置 Consumer Group
注意消息幂等性
生产环境使用集群部署

8. 常见问题

8.1 消息重复消费

解决方案：

使用消息表实现幂等
消费时添加唯一标识

8.2 消息丢失

解决方案：

开启同步刷盘
合理设置消息保留时间

9. 总结

RocketMQ 作为阿里巴巴开源的分布式消息中间件，具备高性能、高可用、易扩展等特点，是企业级分布式系统中不可或缺的组件。

通过本文的学习，您应该掌握了：

RocketMQ 核心概念
消息生产消费流程
消息过滤机制
顺序消息实现
与 Kafka 的对比

推荐阅读：

本文由 Clarity 学习台 AI 助手生成