RocketMQ

RocketMQ
Developer(s)	阿里巴巴集团
Initial release	2012年
	模板:Infobox software/simple
Repository	{{URL\|example.com\|optional display text}};
Written in	Java
Engine
Operating system	跨平台
Type	消息中间件
License	Apache License 2.0
Website	https://rocketmq.apache.org

RocketMQ是由阿里巴巴集团开发并开源的分布式消息中间件，于2012年首次发布，2016年捐赠给Apache软件基金会，2017年成为Apache顶级项目。它以其高吞吐量、低延迟和高可用性著称，广泛应用于电商、金融、物联网等领域的分布式系统架构中。

核心特性[编辑 | 编辑源代码]

RocketMQ具有以下主要特性：

高吞吐量：单机可支持10万级TPS（每秒事务处理量）
低延迟：在优化配置下可达到毫秒级延迟
高可用性：支持主从架构和故障转移
消息可靠性：确保消息至少被消费一次
分布式事务：支持半事务消息机制
消息回溯：可按时间或偏移量重新消费历史消息
消息过滤：支持基于SQL92语法的消息过滤
定时/延时消息：支持精确到秒级的定时消息投递

架构设计[编辑 | 编辑源代码]

RocketMQ采用经典的发布-订阅模式，主要包含以下组件：

核心组件[编辑 | 编辑源代码]

NameServer：轻量级的服务发现组件，负责Broker的管理和路由信息的维护
Broker：消息存储和转发服务器，负责消息的存储、投递和查询
Producer：消息生产者，负责发送消息
Consumer：消息消费者，负责消费消息

安装与配置[编辑 | 编辑源代码]

基本安装[编辑 | 编辑源代码]

以下是在Linux系统上安装RocketMQ的基本步骤：

# 下载RocketMQ
wget https://archive.apache.org/dist/rocketmq/5.0.0/rocketmq-all-5.0.0-bin-release.zip

# 解压安装包
unzip rocketmq-all-5.0.0-bin-release.zip
cd rocketmq-5.0.0/

# 启动NameServer
nohup sh bin/mqnamesrv &

# 启动Broker
nohup sh bin/mqbroker -n localhost:9876 &

使用示例[编辑 | 编辑源代码]

Java客户端示例[编辑 | 编辑源代码]

以下是一个简单的Java生产者和消费者示例：

// 生产者示例
public class ProducerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        
        Message msg = new Message("test_topic", "Hello RocketMQ".getBytes());
        SendResult result = producer.send(msg);
        System.out.println("发送结果: " + result);
        
        producer.shutdown();
    }
}

// 消费者示例
public class ConsumerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("test_topic", "*");
        
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println("收到消息: " + new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        
        consumer.start();
        System.out.println("消费者已启动");
    }
}

应用场景[编辑 | 编辑源代码]

RocketMQ广泛应用于以下场景：

异步解耦：系统间异步通信，降低耦合度
流量削峰：应对突发流量，保护后端系统
分布式事务：实现最终一致性的事务处理
消息通知：系统事件通知和状态更新
日志收集：分布式系统日志的集中收集和处理
实时计算：为流处理系统提供数据源

性能优化[编辑 | 编辑源代码]

为提高RocketMQ性能，可考虑以下优化措施：

Broker配置优化：调整刷盘策略、内存映射文件大小等参数
消息批量发送：减少网络开销
消费者负载均衡：合理设置消费者数量
消息过滤：减少不必要的消息传输
网络优化：使用高性能网络设备和协议

与其他消息中间件对比[编辑 | 编辑源代码]

特性	RocketMQ	Kafka	RabbitMQ
设计目标	金融级可靠性	高吞吐日志处理	企业级消息队列
消息模型	发布/订阅、队列	发布/订阅	队列、发布/订阅
协议支持	自定义协议	自定义协议	AMQP、STOMP等
事务支持	支持	不支持	支持
消息回溯	支持	支持	不支持
延迟消息	支持	不支持	支持