目录
- 1 原生操作
- 1.1 原生生产消息
- 1.2 原生消费消息
- 1.2.1 原生Push和Pull对比
- 1.2.2 Push和SCS和rocketmq
- 1.2.3 Pull 模式（手动拉取）
- 1.2.4 Push 模式
- 1.2.5 其他问题
- 1.2.5.1 拉取间隔pullInterval
- 1.2.5.2 回调线程数ClientCallbackExecutorThreads
1 原生操作
使用原生消息时引入依赖
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.4</version>
</dependency>
1.1 原生生产消息
使用原生 DefaultMQProducer可以提供对 RocketMQ 的完全控制，可以精确配置各种参数（如 sendMsgTimeout、retryTimesWhenSendFailed 等），还可以直接访问所有 RocketMQ 特性
操作示例：
构建生产者配置
@Slf4j
public class MessageProducer{protected transient DefaultMQProducer defaultProducer;public MessageProducer() {DefaultMQProducer defaultProducer = new DefaultMQProducer();//基础配置//设置 NameServer 地址，用于发现 broker 节点defaultProducer.setNamesrvAddr("127.0.0.1:9876");//消息处理配置//配置 DefaultMQProducer 的消费者偏移量持久化间隔defaultProducer.setPersistConsumerOffsetInterval(5000);//设置消息体压缩阈值，超过该大小的消息会被压缩defaultProducer.setCompressMsgBodyOverHowmuch(4096);//设置单条消息的最大大小限制 2MdefaultProducer.setMaxMessageSize(1024*1024*2);//重试机制配置//当消息发送到 broker 失败时是否重试其他 brokerdefaultProducer.setRetryAnotherBrokerWhenNotStoreOK(false);//设置发送失败时的重试次数defaultProducer.setRetryTimesWhenSendFailed(2);//网络和性能配置//设置消息发送超时时间defaultProducer.setSendMsgTimeout(3000);//设置客户端回调执行线程数defaultProducer.setClientCallbackExecutorThreads(Runtime.getRuntime().availableProcessors());//设置客户端 IP 地址defaultProducer.setClientIP(RemotingUtil.getLocalAddress());//心跳和轮询配置//设置与 broker 的心跳间隔defaultProducer.setHeartbeatBrokerInterval(30000);//设置生产者实例名称defaultProducer.setInstanceName("test");//设置轮询 NameServer 的间隔时间defaultProducer.setPollNameServerInterval(30000);this.defaultProducer = defaultProducer;}//启动public void doStart() throws Exception {defaultProducer.start();log.info("mcq started.");}//停止public void doStop() {defaultProducer.shutdown();log.info("mcq producer[%s] stopped.");}
}
构建发送消息方法
// 发送多条消息public SendResult send(Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {return defaultProducer.send(msgs);}//发送单条 同步public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {return defaultProducer.send(msg);}//附带超时时间 同步发送单条public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {return defaultProducer.send(msg, timeout);}//附带异步发送 通过 SendCallback 回调通知 两个回调方法：//onSuccess(SendResult sendResult): 发送成功时调用//onException(Throwable e): 发送失败时调用public void send(Message msg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {defaultProducer.send(msg, sendCallback);}//异步发送多条public void send(Collection<Message> msgs, SendCallback sendCallback) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {defaultProducer.send(msgs,sendCallback);}
1.2 原生消费消息
1.2.1 原生Push和Pull对比
原生 RocketMQ 消费方式主要有两类：Push 模式 和 Pull 模式。
Push 模式特点 和 Pull方式（手动拉取）比较
| 特性 | Push 模式 | Pull 模式 |
|---|---|---|
| 复杂度 | 低，注册监听即可 | 高，需要手动拉取和管理 offset |
| 并发控制 | 消费线程池控制 | 用户线程池控制 |
| 拉取节奏 | pullInterval + 队列并发，间隔不是精确全局 | 可精确控制，按需拉取 |
| 高吞吐量 | 支持 | 可支持，但需要自己实现批量处理 |
| 延迟 | 较低 | 取决于拉取间隔和逻辑 |
| 适用场景 | 日常消费，高吞吐量 | 限流或复杂控制，批量处理，测试等 |
1.2.2 Push和SCS和rocketmq
原生 Push 模式：
- 核心类：
DefaultMQPushConsumer - 消息由
Broker推送到客户端，客户端注册Listener处理 - 特点
- 使用难度较低，注册
Listener即可 - 并发控制通过
consumeThreadMin / consumeThreadMax控制消费线程数 - 拉取控制通过
pullBatchSize、consumeMessageBatchMaxSize、pullInterval - 消费模式是并发消费，可按队列分片并行处理
- 吞吐量高，延迟低
- 但是
pullInterval不是全局严格间隔，消费逻辑完全由应用自己处理，容易出现消息积压
- 使用难度较低，注册
Spring Cloud Stream（SCS）
Spring Cloud Stream（SCS）框架封装了RocketMQ消费逻辑，通过注解@StreamListener或者functional binding (Consumer<T>)接收消息
点击此处了解 SpringCloud之Stream消息驱动RocketMQ讲解- 特点
- 使用注解或函数式接口即可
- 并发控制，通过
binder配置concurrency参数控制 SCS内部自己管理PullTask，不暴露pullInterval- 使用
异步线程池 + 负载均衡的消费模式，自动 ack - 自动管理 offset，支持消息重试和失败管理；集成 Spring 生态
- 但是对低延迟或超高吞吐量的精细控制不如原生 Push 灵活
rocketmq-spring-boot-starter
- 核心是基于
RocketMQ原生SDK封装，和Spring Boot集成 - 注解方式
@RocketMQMessageListener，提供简化配置 点击此处了解 SpringBoot整合RocketMQ - 特点
- 使用难度低，Spring Boot 风格配置
- 并发控制通过
consumeThreadMin / consumeThreadMax或concurrency配置 - 内部使用
Push模式，可通过配置控制批量消费 - 消费模式与原生
Push相似，但集成了Spring Boot配置管理和生命周期 - 简化原生
Push开发，自动ack，支持Spring Boot注解和配置 - 但是灵活性稍低，不如原生 SDK 可精确控制 pullInterval、自定义批量策略
三者比较
| 特性 | 原生 Push | SCS | rocketmq-spring-boot-starter |
|---|---|---|---|
| 易用性 | 中 | 高 | 高 |
| 并发控制 | 高（线程池） | 中（binder concurrency） | 高（线程池或 concurrency） |
| 拉取控制 | 精细，可批量、间隔 | 框架自动管理，不暴露 | 接近原生 Push，但简化配置 |
| 消息确认 | 手动 ack | 自动 ack | 自动 ack（可配置） |
| 消息重试/死信 | SDK 控制 | 框架自动管理 | 框架自动管理 |
| 高吞吐量 | 支持 | 取决于线程池和 binder | 支持，但需配置批量 |
| 使用场景 | 精细化控制、批量消费 | Spring 集成、快速开发 | Spring Boot 项目快速接入 |
1.2.3 Pull 模式（手动拉取）
核心类：DefaultMQPullConsumer
消费逻辑由用户主动控制：
//创建 DefaultMQPullConsumer 实例，指定消费者组名为 "consumerGroup"
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumerGroup");
//设置 NameServer 地址，用于发现 RocketMQ 的 broker 节点
consumer.setNamesrvAddr("127.0.0.1:9876");
//启动消费者实例，初始化与 broker 的连接
consumer.start();//获取指定 "topic" 的所有 MessageQueue 集合，一个 topic 可能包含多个队列
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("topic");
for (MessageQueue mq : mqs) {//获取当前队列的消费位点(offset)，第二个参数 true 表示从 broker 获取最新位点long offset = consumer.fetchConsumeOffset(mq, true);while (true) {//从指定队列拉取消息： mq: 目标消息队列 "*": 消息过滤标签表达式 offset: 从该位点开始拉取 32: 最大拉取消息数量PullResult pullResult = consumer.pullBlockIfNotFound(mq, "*", offset, 32);for (MessageExt msg : pullResult.getMsgFoundList()) {System.out.println(new String(msg.getBody()));}//更新消费位点为下次拉取的起始位置offset = pullResult.getNextBeginOffset();//将新的消费位点更新到 broker，确保消费进度不会丢失consumer.updateConsumeOffset(mq, offset);if (pullResult.getPullStatus() == PullStatus.NO_NEW_MSG) break;}
}
1.2.4 Push 模式
核心类：DefaultMQPushConsumer
消费逻辑通过 注册消息监听器 实现：
构造消费核心
@Slf4j
public class MessageConsumer {protected transient DefaultMQPushConsumer defaultConsumer;private final ConsumerConfig config;public MessageConsumer(ConsumerConfig config,String topic,String group) throws MQClientException{DefaultMQPushConsumer defaultConsumer = new DefaultMQPushConsumer();//消费者基础配置//设置 NameServer 地址，用于发现 broker 节点defaultConsumer.setNamesrvAddr("127.0.0.1:9876");//设置消费者组名称，用于标识一组相关的消费者实例defaultConsumer.setConsumerGroup(group);//线程池配置//设置客户端回调执行线程数defaultConsumer.setClientCallbackExecutorThreads(Runtime.getRuntime().availableProcessors()*2);//设置消费线程池最小线程数defaultConsumer.setConsumeThreadMin(1);//设置消费线程池最大线程数defaultConsumer.setConsumeThreadMax(2);//消费行为配置// 设置一次从 broker 拉取的最大消息数defaultConsumer.setPullBatchSize(100);//设置拉取间隔时间（毫秒），控制拉取频率defaultConsumer.setPullInterval(1000);// 拉取间隔// 设置一次交给业务处理的最大消息数defaultConsumer.setConsumeMessageBatchMaxSize(50); //偏移量和心跳配置//设置消费者偏移量持久化间隔defaultConsumer.setPersistConsumerOffsetInterval(5000);//设置与 broker 的心跳间隔defaultConsumer.setHeartbeatBrokerInterval(30000);//订阅配置//订阅指定主题的所有消息（"*"表示不过滤）defaultConsumer.subscribe(topic, "*");//监听器注册//注册并发消息监听器，当消息到达时调用 consumerEvent 方法处理defaultConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context)->consumerEvent(msgs, context));this.defaultConsumer = defaultConsumer;}//启动public void doStart(String desc) throws Exception {defaultConsumer.start();log.info("RocketMQ consumer :{},started", desc);}//注销public void doStop(String desc) {defaultConsumer.shutdown();log.info("RocketMQ consumer :{},stop", desc);}//业务逻辑public ConsumeConcurrentlyStatus consumerEvent(List<MessageExt> msgList, ConsumeConcurrentlyContext context){try {MDC.put("trace_id", IdUtil.simpleUUID());if(CollUtil.isEmpty(msgList)) return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;List<UserInfo> userInfoList = msgList.stream().filter(f -> StrUtil.isNotBlank(f.getTopic()) && f.getTopic().equals("test")).map(m -> JSONObject.parseArray(new String(m.getBody(), StandardCharsets.UTF_8), UserInfo.class)).flatMap(m -> m.stream()).toList();if(CollUtil.isNotEmpty(userInfoList )){// 业务逻辑处理部分}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}catch (Exception e){log.error("RocketMQ消费异常:{}",e.getMessage(),e);return ConsumeConcurrentlyStatus.RECONSUME_LATER;}finally {log.info("RocketMQ消费结束....");MDC.clear();}}
}
使用示例
public void init() {try {MessageConsumer eventConsumer = new MessageConsumer(consumer, consumer.getEventTopic(),consumer.getEventGroup());eventConsumer.doStart("MQ测试");} catch (Exception e) {log.error("MQ启动失败:{}",e.getMessage(),e);if(eventConsumer!=null) eventConsumer.doStop("埋点测试");}}
1.2.5 其他问题
1.2.5.1 拉取间隔pullInterval
pullInterval 的行为可以理解成 每个 PullTask 完成后的最小间隔，具体是这样的：
RocketMQ Pull模型
对每个 队列（Queue），RocketMQ会创建一个PullTask
PullTask的作用是不断从 Broker 拉取消息，然后交给消费线程处理pullInterval的作用
defaultConsumer.setPullInterval(xxx)设置的 单位毫秒
表示 每个PullTask拉取一次消息后，至少等待pullInterval再进行下一次拉取
如果队列有未消费的消息，PullTask完成后立即尝试拉下一批，但每个队列的间隔保证至少 xxx 毫秒，只有在拉取为空（没有新消息）时，才会等待pullInterval毫秒后再尝试下一次拉取
注意：这里是每个 队列的PullTask间隔，不是整个主题的，也不是整个Consumer Group的全局间隔- 实际行为
如果一个PullTask拉取到消息并立即消费完毕，PullTask会等待pullInterval后再触发下一次拉取
如果队列里消息很多、消费慢PullTask拉取完后消息还没消费完，不会阻塞下一次拉取
RocketMQ会根据队列的状态尽可能连续拉取，以保证消息不积压 PullTask在一次拉取后会决定立即再拉还是等 pullInterval 再拉，需要看下列几个条件，满足可以接收更多并且处理能力允许的条件就 立即再拉；否则就休眠 pullInterval再拉：- 本次拉取是否返回了消息（
pulledCount > 0）， - 本地
ProcessQueue（缓存）是否已接近或超过高水位， - 消费（业务）线程池是否有空闲线程能立即处理更多消息，
- 以及是否在做流控/限速（Broker/Client 流控信号）
- 本次拉取是否返回了消息（
1.2.5.2 回调线程数ClientCallbackExecutorThreads
ClientCallbackExecutorThreads：是指客户端处理网络回调（包括消息拉取结果、心跳响应、offset 更新、发送确认等）时的线程池大小
换句话说，它是 RocketMQ 客户端网络 I/O 层（Netty 客户端） 的回调处理线程数，主要负责执行客户端与 Broker 之间通信事件的回调逻辑。
RocketMQ 客户端内部有三个关键线程池概念：
| 线程池名称 | 控制参数 | 作用 |
|---|---|---|
| Netty Callback Executor（网络回调线程池） | setClientCallbackExecutorThreads | 负责处理来自 Broker 的网络响应，例如：拉取结果回调、发送响应、心跳响应等 |
| Pull Message Service（拉取线程池） | 内部固定线程（每个 consumer 一个） | 循环执行 pullMessage()，从 Broker 拉消息（异步发请求）——不是由上面的 callback pool 控制 |
| Consume Message ThreadPool（业务消费线程池） | setConsumeThreadMin/Max 或 spring 配置 consumeThreadMin/Max | 负责执行用户的 MessageListener（也就是业务逻辑），和 BulkProcessor、ES 写入等操作 |
状态(state)和属性(props)之间有何不同？)
