RocketMQ——总结（一）基础使用
官网：https://rocketmq.apache.org/
源码地址：https://github.com/apache/rocketmq
一、RocketMQ的优点
1、天然支持集群模式、负载均衡、水平扩展能力
2、上亿级别的消息堆积能力
3、采用零拷贝的原理、顺序写盘、随机读（借鉴kafka）
4、丰富的API使用，支持顺序消息，事务消息，rabbitmq不支持
5、代码优秀，底层通信通过Netty NIO框架
6、NameServer代替Zookeeper（2.x时还是zk）
7、强调集群无单点，可扩展，任意一点高可用，水平扩展
8、消息失败重试机制（rabbit没有）、消息可查询
9、开源社区活跃、成熟度高（经过双十一的考验）
10，支持类sql表达式过滤，可能借鉴activemq
- rabbit,kafka,rocket对比
| 对比点 | RabbitMQ | Kafka | RocketMQ |
| 消息堆积能力 | 百万 | 上亿 | 上亿 |
| 读写原理 | - | 零拷贝 | 零拷贝 |
| 顺序消息 | 不支持 | 支持 | 支持 |
| 事务消息 | 不支持 | 不支持 | 支持 |
| 注册中心 | - | Zookeeper | NameServer |
| 消息失败重试机制 | 不支持 | 不支持 | 支持 |
| 消息可查询 | 不支持 | 不支持(需二次开发) | 支持 |
| 订阅和拉取模式 | 支持 | 支持 | 支持 |
| 长轮询 | 不支持 | 不支持 | 支持 |
| 开发语言 | Erlang | Scala | Java |
二、概念
Producer: 消息生产者,负责产生消息，一般由业务系统负责产生消息。
Consumer:消息消费者,负责消费消息，一般是后台系统负责异步消费。
Push Consumer: Consumer的一 种，需要向Consumer对象注册监听。
Pull Consumer: Consumer的一种，需要主动请求Broker拉取消息。
Producer Group:生产者集合, 一般用于发送一类消息，可用于事务消息
Consumer Group:消费者集合，一般用于接受一类消息进行消费
Broker : MQ消息服务(中转角色，用于消息存储与生产消费转发)
NameServer: 消息主题的注册中心
Tag: 标签，相当于子主题，可用来过滤或筛选
其他核心概念参见：https://rocketmq.apache.org/docs/core-concept/
说明：推拉两种消费模式，类似rabbitmq，但推送实现机制不同，这里的推模式实际是基于长轮询
三、技术架构
Broker常见集群架构部署模式：
1.单点模式
2.主从模式
3.双主模式
4.双主双从模式、多主多从模式
其他说明：
brokerId为0的表示master
四、生产者
- 使用说明：
1.创建生产者对象DefaultMQProducer
2.设置NamesrvAddr
3.启动生产者服务
4.创建消息并发送
5.关闭生产者
- 生产者类
- 生产者核心参数
| 属性 | 说明 |
| producerGroup | 组名，在一个应用里必须唯一 |
| createTopicKey | 自动创建的Topic名称，在发送消息时，自动创建服务器不存在的topic，需要指定Key。broker必须开启isAutoCreateTopicEnable |
| defaultTopicQueueNums | 创建topic时默认创建的队列数（默认4） |
| sendMsgTimeout | 发送超时时间 |
| compressMsgBodyOverHowmuch | 消息超过多少（默认4096）字节后进行压缩 |
| retryTimesWhenSendFailed | 同步发送失败时的重试次数 |
| retryTimesWhenSendAsyncFailed | 异步发送失败时的重试次数 |
| retryAnotherBrokerWhenNotStoreOK | 声明发送失败时，下次是否投递给其他Broker，默认false |
| maxMessageSize | 最大消息大小。默认4M; 客户端限制的消息大小，超过报错，同时服务端也会限制 |
- 其他说明：
topic是某一类主题， tag是标签用来过滤，key一般是用来作为用户自定义的Key,唯一标识
topic对消息队列是一对多的关系，默认队列有4个，没有主题的可自动创建
rocketmq的异常区分较细，异常区分细说明rocketmq的源码较为细致
- 示例代码
发送消息的示例代码：org.apache.rocketmq.example.quickstart.Producer
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
/*
* Instantiate with a producer group name.
*/
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
/*
* Specify name server addresses.
*
*
* Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
*
* {@code
* producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
* }
*
*/
/*
* Launch the instance.
*/
producer.start();
for (int i = 0; i < 1000; i++) {
try {
/*
* Create a message instance, specifying topic, tag and message body.
*/
Message msg = new Message("TopicTest" /* Topic */,
"TagA" /* Tag */,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);
/*
* Call send message to deliver message to one of brokers.
*/
SendResult sendResult = producer.send(msg);
/*
* There are different ways to send message, if you don't care about the send result,you can use this way
* {@code
* producer.sendOneway(msg);
* }
*/
/*
* if you want to get the send result in a synchronize way, you can use this send method
* {@code
* SendResult sendResult = producer.send(msg);
* System.out.printf("%s%n", sendResult);
* }
*/
/*
* if you want to get the send result in a asynchronize way, you can use this send method
* {@code
*
* producer.send(msg, new SendCallback() {
* @Override
* public void onSuccess(SendResult sendResult) {
* // do something
* }
*
* @Override
* public void onException(Throwable e) {
* // do something
* }
*});
*
*}
*/
System.out.printf("%s%n", sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
/*
* Shut down once the producer instance is not longer in use.
*/
producer.shutdown();
}
}
五、消费者
- 使用说明：
1.创建消费者对象DefaultMQPushConsumer
2.设置NamesrvAddr及其消费位置ConsumeFromWhere
3.进行订阅主题subscribe
4.注册监听并消费registerMessageListener
- 消费者核心参数
| 属性 | 说明 |
|
comsumeFromWhere |
作用于消费组的第一次消费，有消费后会同步到Broker，下次消费就在此前的基础上消费 CONSUME_FROM_FIRST_OFFSET： 初次从消息队列头部开始消费，即历史消息（还储存在 broker 的）全部消费一遍 CONSUME_FROM_LAST_OFFSET： 默认策略，初次从该队列最尾开始消费，即跳过历史消息 CONSUME_FROM_TIMESTAMP：从某个时间点开始消费，默认是半个小时以前 |
|
allocateMessageQueueStrategy |
负载均衡策略算法，即消费者分配到 queue 的算法，默认值是 AllocateMessageQueueAveragely 即取模平均分配 |
| offsetStore | 消息消费进度存储器 offsetStore 有两个策略：LocalFileOffsetStore 和 RemoteBrokerOffsetStore ，广播模式默认使用 LocalFileOffsetStore， 集群模式默认使用 RemoteBrokerOffsetStore |
| messageModel | 消费者消费模式 CLUSTERING：集群模式（默认配置） BROADCASTING：广播模式 |
| consumeThreadMax | 最大消费线程池数量 |
| consumeThreadMin | 最小消费线程池数量 |
| consumeConcurrentlyMaxSpan | 单个队列消费并行的最大跨度？？？ |
| pullThresholdForQueue | 一个队列消费的最大个数 |
| pullInterval | 消费者拉取的时间间隔 |
| pullBatchSize | 消费者去 broker 拉取消息时，一次拉取多少条。可选配置 |
| consumeMessageBatchMaxSize | 单个线程消费时一次性消费多少条消息，批量消费接口才有用，可选配置 |
- 其他说明：
rocketmq的订阅模式不是服务端真的推，而是客户端长轮询的机制
消费成功返回：ConsumeConcurrentlyStatus.CONSUME_SUCCESS
消费失败了可以自动重试，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER
生产环境下重试多次失败后，需要自己做补偿和记录日志等
重试策略：messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9r 10m 20m 30m 1h 2h
- 示例代码
消费者的示例代码：org.apache.rocketmq.example.quickstart.Consumer
public class Consumer { public static void main(String[] args) throws InterruptedException, MQClientException { /* * Instantiate with specified consumer group name. */ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4"); /* * Specify name server addresses. * * * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR ** {@code * consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876"); * } **/ /* * Specify where to start in case the specific consumer group is a brand-new one. */ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); /* * Subscribe one more topic to consume. */ consumer.subscribe("TopicTest", "*"); /* * Register callback to execute on arrival of messages fetched from brokers. */ consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(Listmsgs, ConsumeConcurrentlyContext context) { System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); /* * Launch the consumer instance. */ consumer.start(); System.out.printf("Consumer Started.%n"); } }