今天的博客主题
MQ消息中间件 --》RocketMQ --》RocketMQ进阶（二）
本文主要讲解RocketMQ消费者实现及底层实现原理刨根问底
同样和生产者一样，对于使用普通Java项目来实现的例子就不多说了，官方文档都有
样例地址：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md
主要看下springboot实现消费者。
springboot集成rocketmq消费者实现
pom.xml文件引入MQ依赖jar包
<!-- rocketmq依赖 -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.3</version>
</dependency>application.yml配置文件添加配置
rocketmq:
#服务地址
name-server: 10.10.16.20:9876
#消费者配置
consumer:
#消费组名
group: CG_DEMO
#主题
topic: TOPIC_DEMO_ONE
#标签
tag: TAG_ONE具体代码实现
@Service
@RocketMQMessageListener(consumerGroup = "${rocketmq.consumer.group}", topic = "${rocketmq.consumer.topic}")
public class MsgListener implements RocketMQListener<MessageExt> {
private org.slf4j.Logger logger = LoggerFactory.getLogger(getClass());
@Override
public void onMessage(MessageExt msg) {
logger.debug("RECEIVE_MSG_BEGIN: " + msg.toString());
logger.debug(String.format("消费消息，消息ID：%s，消息KEY：%s，消息体：%s ", msg.getMsgId(), msg.getKeys(), new String(msg.getBody())));
}
}测试
直接启动服务即可，会自动监听MQ服务，监听到消息就消费。
启动完就看到控制台输出：
2019-07-10 10:22:16.633 INFO 9504 --- [MessageThread_1] com.mq.listener.MsgListener : RECEIVE_MSG_BEGIN: MessageExt [queueId=2, storeSize=282, queueOffset=0, sysFlag=0, bornTimestamp=1562577172354, bornHost=/10.10.51.219:57247, storeTimestamp=1562576637320, storeHost=/10.10.16.20:10911, msgId=0A0A101400002A9F000000000006982F, commitLogOffset=432175, bodyCRC=1703141525, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TOPIC_DEMO_ONE', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, KEYS=123456, CONSUME_START_TIME=1562725336633, id=6b3efbac-03e4-0ced-4a9d-a722be0f775c, UNIQ_KEY=0A0A33DB31DC18B4AAC227BE22DD0000, WAIT=false, TAGS=TAG_ONE, timestamp=1562577172027}, body=[72, 101, 108, 108, 111, 32, 87, 111, 114, 108, 100, 32, 50, 48, 49, 57, 45, 48, 55, 45, 48, 56, 84, 49, 55, 58, 49, 50, 58, 53, 50, 46, 48, 50, 50], transactionId='null'}]
2019-07-10 10:22:16.633 INFO 9504 --- [MessageThread_1] com.mq.listener.MsgListener : 消费消息，消息ID：0A0A33DB31DC18B4AAC227BE22DD0000，消息KEY：123456，消息体：Hello World 2019-07-08T17:12:52.022 到这里就在springboot中集成了rocketmq消费者并成功消费了生产者发送的消息。
下面开始对消费者消费消息进行刨根问底。
@RocketMQMessageListener 注解
在这个注解类里面，只有一些字符串类型的方法，主要来管控一些参数。
看一下这个注解是怎么初始化它的。看下被引用的地方，因为是自动装配的，我们直接找 Configuration 的哪个
ListenerContainerConfiguration 类，会在启动时候被spring扫描到
@Configuration
public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {
// 实现了ApplicationContextAware 应用程序上下文感知 SmartInitializingSingleton 初始化单例
...省略部分代码
// 因为是实现必须要重写接口方法，直接从这个两个方法作为入口
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void afterSingletonsInstantiated() {
// 获取带有 RocketMQMessageListener 的bean
Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);
if (Objects.nonNull(beans)) {
beans.forEach(this::registerContainer);
}
}
// 初始化单利，将监听器注册到容器
private void registerContainer(String beanName, Object bean) {
Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
// 是否可分配的实例
if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
}
// 获得消息监听器，进行验证
RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
validate(annotation);
// 容器bean名称
String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
counter.incrementAndGet());
GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
// 将bean注册到通用上下文应用中
genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
() -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
// 从这个上下文中在获的这个bean?
DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
DefaultRocketMQListenerContainer.class);
if (!container.isRunning()) {
try {
// 调用这个容器里的开始方法，就是启动consumer
container.start();
} catch (Exception e) {
log.error("Started container failed. {}", container, e);
throw new RuntimeException(e);
}
}
log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
}
}通过 DefaultRocketMQListenerContainer 里的 initRocketMQPushConsumer 来初始化 DefaultMQPushConsumer
public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
省略部分代码...
private void initRocketMQPushConsumer() throws MQClientException {
Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
Assert.notNull(nameServer, "Property 'nameServer' is required");
Assert.notNull(topic, "Property 'topic' is required");
// 权限钩子
RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(applicationContext.getEnvironment(),
this.rocketMQMessageListener.accessKey(), this.rocketMQMessageListener.secretKey());
boolean enableMsgTrace = rocketMQMessageListener.enableMsgTrace();
// 实例化DefaultMQPushConsumer
if (Objects.nonNull(rpcHook)) {
consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(),
enableMsgTrace, this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
consumer.setVipChannelEnabled(false);
consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, consumerGroup));
} else {
log.debug("Access-key or secret-key not configure in " + this + ".");
consumer = new DefaultMQPushConsumer(consumerGroup, enableMsgTrace,
this.applicationContext.getEnvironment().
resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
}
// 为DefaultMQPushConsumer设置上下文参数
String customizedNameServer = this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.nameServer());
if (customizedNameServer != null) {
consumer.setNamesrvAddr(customizedNameServer);
} else {
consumer.setNamesrvAddr(nameServer);
}
if (accessChannel != null) {
consumer.setAccessChannel(accessChannel);
}
consumer.setConsumeThreadMax(consumeThreadMax);
if (consumeThreadMax < consumer.getConsumeThreadMin()) {
consumer.setConsumeThreadMin(consumeThreadMax);
}
consumer.setConsumeTimeout(consumeTimeout);
consumer.setInstanceName(this.name);
// 消费方式
switch (messageModel) {
case BROADCASTING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
break;
case CLUSTERING:
consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
break;
default:
throw new IllegalArgumentException("Property 'messageModel' was wrong.");
}
// 过滤方式
switch (selectorType) {
case TAG:
consumer.subscribe(topic, selectorExpression);
break;
case SQL92:
consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
break;
default:
throw new IllegalArgumentException("Property 'selectorType' was wrong.");
}
// 通过不同消费姿势，来注册不同的监听器
switch (consumeMode) {
case ORDERLY:
consumer.setMessageListener(new DefaultMessageListenerOrderly());
break;
case CONCURRENTLY:
consumer.setMessageListener(new DefaultMessageListenerConcurrently());
break;
default:
throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
}
if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer);
}
}
}DefaultMQPushConsumer
/**
* 1.推荐使用的消息消费方式
* 2.从技术上来说，这个push实现底层实际上是对pull方式的一个封装。从服务器定时拉取数据，来回调监听程序。
* 3.它是线程安全的
*/
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
// 部分代码省略...
// 提供多个构造函数
// 所支持消费的一些方法
// 启动停止消费的操控方法
// 都没有具体实现，具体的一些实现在 DefaultMQPushConsumerImpl 类里面
}实例化完成之后，DefaultRocketMQListenerContainer.start(); 启动消费者
@Override
public void start() {
// 不是运行状态在去start
if (this.isRunning()) {
throw new IllegalStateException("container already running. " + this.toString());
}
try {
// 启动消费者
consumer.start();
} catch (MQClientException e) {
throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
}
this.setRunning(true);
log.info("running container: {}", this.toString());
}DefaultMQPushConsumer.start();
@Override
public void start() throws MQClientException {
// 启动MQ拉取模式消费者
this.defaultMQPushConsumerImpl.start();
if (null != traceDispatcher) {
try {
traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
} catch (MQClientException e) {
log.warn("trace dispatcher start failed ", e);
}
}
}defaultMQPushConsumerImpl.start();
public synchronized void start() throws MQClientException {
switch (this.serviceState) {
// 服务未运行再去实例做些事情
case CREATE_JUST:
log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
this.serviceState = ServiceState.START_FAILED;
// 检查必要配置
this.checkConfig();
// 复制订阅关系
this.copySubscription();
if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
this.defaultMQPushConsumer.changeInstanceNameToPID();
}
// 获取并创建MQ客户端实例
this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
// 设置平衡算法实例在基类
this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
// 拉取消息API包装实例化
this.pullAPIWrapper = new PullAPIWrapper(
mQClientFactory,
this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
// 注册过滤消息钩子
this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
//偏移量存储
if (this.defaultMQPushConsumer.getOffsetStore() != null) {
this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
} else {
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
case CLUSTERING:
this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
break;
default:
break;
}
this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
}
this.offsetStore.load();
// 内部消息监听器
if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
this.consumeOrderly = true;
this.consumeMessageService =
new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
this.consumeOrderly = false;
this.consumeMessageService =
new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
}
// 启动消费消息服务
this.consumeMessageService.start();
// 消费者是否成功注册到MQ客户端工厂
boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
// 失败更改服务启动状态，停止消费消息服务
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
this.consumeMessageService.shutdown();
throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}
// 注册成功就启动客户端工厂
mQClientFactory.start();
log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
// 服务状态为运行中
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING: // 服务运行中
case START_FAILED: // 服务启动失败
case SHUTDOWN_ALREADY: // 服务停止，抛出异常可能重启一次
throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}
// 订阅关系更改时更新主题订阅信息
this.updateTopicSubscribeInfoWhenSubscriptionChanged();
// 在代理中检查客户端
this.mQClientFactory.checkClientInBroker();
// 锁定心跳 发送到所有的代理
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
// 立即实行平衡
this.mQClientFactory.rebalanceImmediately();
}consumeMessageService.start();
consumeMessageService是一个接口有两个实现ConsumeMessageConcurrentlyService(并发消费消息服务) 和 ConsumeMessageOrderlyService(顺序消费消息服务)
ConsumeMessageConcurrentlyService.start(); 实现 consumeMessageService.start();
public void start() {
// 用了jdk的一个实例，计划执行任务服务
this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// 定期清理过期消息
cleanExpireMsg();
}
}, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
}ConsumeMessageOrderlyService.start(); 实现 consumeMessageService.start();
public void start() {
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
// 集群消费时，才会指定这个计划任务
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// 定期锁定MQ
ConsumeMessageOrderlyService.this.lockMQPeriodically();
}
}, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
}
}mQClientFactory.start(); == MQClientInstance.start();
public void start() throws MQClientException {
synchronized (this) {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
// 如果未指定，则从名称服务器查找地址
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// 启动请求响应通道（远程客户端，netty实现）
this.mQClientAPIImpl.start();
// 开始各种计划任务
this.startScheduledTask();
// 启动拉取服务
this.pullMessageService.start();
// 启动平衡服务
this.rebalanceService.start();
// 启动推送服务
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
log.info("the client factory [{}] start OK", this.clientId);
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
break;
case SHUTDOWN_ALREADY:
break;
case START_FAILED:
throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
default:
break;
}
}
}这里和生产者一样，都启动了客户端工厂的一些任务。
到这里消费者就启动注入成功了》。。
接下来看下消费者是怎么拉取消息进行消费的
在执行mQClientFactory.start(); 会发现有这么一行代码 this.pullMessageService.start();
看下pullMessageService这个类（拉取消息服务类）
public class PullMessageService extends ServiceThread {
// 部分代码省略。。。
//
private void pullMessage(final PullRequest pullRequest) {
final MQConsumerInner consumer =
this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
if (consumer != null) {
DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
impl.pullMessage(pullRequest);
} else {
log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
}
}
@Override
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
PullRequest pullRequest = this.pullRequestQueue.take();
this.pullMessage(pullRequest);
} catch (InterruptedException ignored) {
} catch (Exception e) {
log.error("Pull Message Service Run Method exception", e);
}
}
log.info(this.getServiceName() + " service end");
}
}DefaultMQPushConsumerImpl.pullMessage(); 具体拉取实现
public void pullMessage(final PullRequest pullRequest) {
// 从拉取请求获取消费队列快照
final ProcessQueue processQueue = pullRequest.getProcessQueue();
// 请求是否被抛弃
if (processQueue.isDropped()) {
log.info("the pull request[{}] is dropped.", pullRequest.toString());
return;
}
// 最后一次拉取时间戳
pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());
// 确保服务正在运行，否则存储到稍后执行拉取请求队列中
try {
this.makeSureStateOK();
} catch (MQClientException e) {
log.warn("pullMessage exception, consumer state not ok", e);
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
return;
}
// 消费是否被暂停，否则存储到稍后执行拉取请求队列中
if (this.isPause()) {
log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
return;
}
// 缓存消息总数
long cachedMessageCount = processQueue.getMsgCount().get();
// 缓存消息大小
long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);
// 缓存消息总数是否超过阈值，否则存储到稍后执行拉取请求队列中
if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
if ((queueFlowControlTimes++ % 1000) == 0) {
log.warn(
"the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
}
return;
}
// 缓存消息大小是否超过阈值，否则存储到稍后执行拉取请求队列中
if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
if ((queueFlowControlTimes++ % 1000) == 0) {
log.warn(
"the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
}
return;
}
// 是否顺序消费
if (!this.consumeOrderly) { // 并发消费
// 是否超过并发最大偏移量，否则存储到稍后执行拉取请求队列中
if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
log.warn(
"the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
pullRequest, queueMaxSpanFlowControlTimes);
}
return;
}
} else { // 顺序消费
if (processQueue.isLocked()) {
// 顺序拉取消息，没有被锁定
if (!pullRequest.isLockedFirst()) {
// 计算从什么位置拉取
final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());
// 计算出的拉取位置 小于 请求下一个偏移量 则代理忙
boolean brokerBusy = offset < pullRequest.getNextOffset();
log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
pullRequest, offset, brokerBusy);
if (brokerBusy) {
log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset."+
"pullRequest: {} NewOffset: {}",pullRequest, offset);
}
// 请求加锁
pullRequest.setLockedFirst(true);
// 重置偏移量
pullRequest.setNextOffset(offset);
}
} else {
// 没有锁定在代理中，稍后在执行拉取消费
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
log.info("pull message later because not locked in broker, {}", pullRequest);
return;
}
}
// 根据topic获取内部订阅关系
final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
// 查找消费者的订阅关系失败，稍后在执行拉取消费
if (null == subscriptionData) {
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
log.warn("find the consumer's subscription failed, {}", pullRequest);
return;
}
// 开始时间戳
final long beginTimestamp = System.currentTimeMillis();
// 拉取回调 ======》核心
PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
// 请求响应结果
if (pullResult != null) {
// 处理拉取结果（处理下消息进行二进制，重置下偏移量...）
pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
subscriptionData);
// 拉取结果
switch (pullResult.getPullStatus()) {
case FOUND: // 消息存在
// 上一个请求偏移量
long prevRequestOffset = pullRequest.getNextOffset();
// 获取下一个开始偏移量
pullRequest.setNextOffset(pullResult.getNextBeginOffset());
// 拉取结果时间
long pullRT = System.currentTimeMillis() - beginTimestamp;
// 消费统计管理，记录消费时间
DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
pullRequest.getMessageQueue().getTopic(), pullRT);
long firstMsgOffset = Long.MAX_VALUE;
// 消息列表为空，就稍后执行拉取
if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
} else {
// 第一个消息的 队列偏移量
firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
// 消费统计管理，记录消费TPS
DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
// 发送到消费
boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
// 提交消费请求 ======》核心
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
pullResult.getMsgFoundList(),
processQueue,
pullRequest.getMessageQueue(),
dispatchToConsume);
// 拉取间隔，稍稍还是立即
if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
} else {
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
}
}
// 拉取偏移量是否合法
if (pullResult.getNextBeginOffset() < prevRequestOffset
|| firstMsgOffset < prevRequestOffset) {
log.warn(
"[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
pullResult.getNextBeginOffset(),
firstMsgOffset,
prevRequestOffset);
}
break;
case NO_NEW_MSG: // 没有新消息
pullRequest.setNextOffset(pullResult.getNextBeginOffset());
DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
break;
case NO_MATCHED_MSG: // 没有匹配消息
pullRequest.setNextOffset(pullResult.getNextBeginOffset());
DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
break;
case OFFSET_ILLEGAL: // 偏移量非法（太大或太小）
log.warn("the pull request offset illegal, {} {}",
pullRequest.toString(), pullResult.toString());
pullRequest.setNextOffset(pullResult.getNextBeginOffset());
pullRequest.getProcessQueue().setDropped(true);
DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {
@Override
public void run() {
try {
DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
pullRequest.getNextOffset(), false);
DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
log.warn("fix the pull request offset, {}", pullRequest);
} catch (Throwable e) {
log.error("executeTaskLater Exception", e);
}
}
}, 10000);
break;
default:
break;
}
}
}
@Override
public void onException(Throwable e) {
if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
log.warn("execute the pull request exception", e);
}
DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
}
};
// 启动提交偏移量
boolean commitOffsetEnable = false;
// 提交偏移量值
long commitOffsetValue = 0L;
// 集群消费时才启用
if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {
commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);
if (commitOffsetValue > 0) {
commitOffsetEnable = true;
}
}
// 订阅表达式
String subExpression = null;
// 类过滤器
boolean classFilter = false;
// 根据topic从平衡算法实现里获取内部订阅关系数据
SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
if (sd != null) {
// 拉取时是否更新订阅关系 并且 类过滤器模式
if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
subExpression = sd.getSubString();
}
classFilter = sd.isClassFilterMode();
}
// 构建拉取系统标识
int sysFlag = PullSysFlag.buildSysFlag(
commitOffsetEnable, // commitOffset
true, // suspend
subExpression != null, // subscription
classFilter // class filter
);
try {
// 拉取核心 实现
this.pullAPIWrapper.pullKernelImpl(
pullRequest.getMessageQueue(),
subExpression,
subscriptionData.getExpressionType(),
subscriptionData.getSubVersion(),
pullRequest.getNextOffset(),
this.defaultMQPushConsumer.getPullBatchSize(),
sysFlag,
commitOffsetValue,
BROKER_SUSPEND_MAX_TIME_MILLIS,
CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
CommunicationMode.ASYNC,
pullCallback
);
} catch (Exception e) {
log.error("pullKernelImpl exception", e);
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
}
}pullAPIWrapper.pullKernelImpl(); 看下这个拉取的核心实现
public PullResult pullKernelImpl(
final MessageQueue mq,
final String subExpression,
final String expressionType,
final long subVersion,
final long offset,
final int maxNums,
final int sysFlag,
final long commitOffset,
final long brokerSuspendMaxTimeMillis,
final long timeoutMillis,
final CommunicationMode communicationMode,
final PullCallback pullCallback
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
// 从订阅关系中查找代理地址
FindBrokerResult findBrokerResult =
this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
this.recalculatePullFromWhichNode(mq), false);
// 查询代理结果为空
if (null == findBrokerResult) {
// 从名称服务器更新主题路由信息
this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
// 在从订阅关系中查找代理地址
findBrokerResult =
this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
this.recalculatePullFromWhichNode(mq), false);
}
// 查询代理结果不为空
if (findBrokerResult != null) {
{
// check version 检查版本及消息过滤表达式类型
if (!ExpressionType.isTagType(expressionType)
&& findBrokerResult.getBrokerVersion() < MQVersion.Version.V4_1_0_SNAPSHOT.ordinal()) {
throw new MQClientException("The broker[" + mq.getBrokerName() + ", "
+ findBrokerResult.getBrokerVersion() + "] does not upgrade to support for filter message by " + expressionType, null);
}
}
int sysFlagInner = sysFlag;
// 查找代理结果如果是从
if (findBrokerResult.isSlave()) {
sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
}
// 拉取消息请求头
PullMessageRequestHeader requestHeader = new PullMessageRequestHeader();
requestHeader.setConsumerGroup(this.consumerGroup);
requestHeader.setTopic(mq.getTopic());
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setQueueOffset(offset);
requestHeader.setMaxMsgNums(maxNums);
requestHeader.setSysFlag(sysFlagInner);
requestHeader.setCommitOffset(commitOffset);
requestHeader.setSuspendTimeoutMillis(brokerSuspendMaxTimeMillis);
requestHeader.setSubscription(subExpression);
requestHeader.setSubVersion(subVersion);
requestHeader.setExpressionType(expressionType);
String brokerAddr = findBrokerResult.getBrokerAddr();
if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
// 计算筛选从哪个服务器中提取
brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
}
// 拉取消息实现
PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
brokerAddr,
requestHeader,
timeoutMillis,
communicationMode,
pullCallback);
return pullResult;
}
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}getMQClientAPIImpl().pullMessage(); 拉取实现
public PullResult pullMessage(
final String addr,
final PullMessageRequestHeader requestHeader,
final long timeoutMillis,
final CommunicationMode communicationMode,
final PullCallback pullCallback
) throws RemotingException, MQBrokerException, InterruptedException {
// 请求服务端标识
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);
// 通信模式
switch (communicationMode) {
case ONEWAY:
assert false;
return null;
case ASYNC: // 异步拉取
this.pullMessageAsync(addr, request, timeoutMillis, pullCallback);
return null;
case SYNC: // 同步拉取
return this.pullMessageSync(addr, request, timeoutMillis);
default:
assert false;
break;
}
return null;
}pullMessageAsync(); 异步拉取
private void pullMessageAsync(
final String addr,
final RemotingCommand request,
final long timeoutMillis,
final PullCallback pullCallback
) throws RemotingException, InterruptedException {
this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback() {
@Override
public void operationComplete(ResponseFuture responseFuture) {
RemotingCommand response = responseFuture.getResponseCommand();
if (response != null) {
try {
// 设置拉取状态
PullResult pullResult = MQClientAPIImpl.this.processPullResponse(response);
// 断言不为空
assert pullResult != null;
// 执行回调
pullCallback.onSuccess(pullResult);
} catch (Exception e) {
pullCallback.onException(e);
}
} else {
// 异常回调
if (!responseFuture.isSendRequestOK()) {
pullCallback.onException(new MQClientException("send request failed to " + addr + ". Request: " + request, responseFuture.getCause()));
} else if (responseFuture.isTimeout()) {
pullCallback.onException(new MQClientException("wait response from " + addr + " timeout :" + responseFuture.getTimeoutMillis() + "ms" + ". Request: " + request,
responseFuture.getCause()));
} else {
pullCallback.onException(new MQClientException("unknown reason. addr: " + addr + ", timeoutMillis: " + timeoutMillis + ". Request: " + request, responseFuture.getCause()));
}
}
}
});
}pullMessageSync(); 同步拉取
private PullResult pullMessageSync(
final String addr,
final RemotingCommand request,
final long timeoutMillis
) throws RemotingException, InterruptedException, MQBrokerException {
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
assert response != null;
return this.processPullResponse(response);
}看异步的回调：pullCallback.onSuccess(pullResult);
这个回调在DefaultMQPushConsumerImpl.pullMessage(); 在这个回调方法里还有一段比较核心的代码
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
pullResult.getMsgFoundList(),
processQueue,
pullRequest.getMessageQueue(),
dispatchToConsume);consumeMessageService.submitConsumeRequest();
是提供的一个接口有两个实现类：ConsumeMessageConcurrentlyService 和 ConsumeMessageOrderlyService
/**
* 并发消费
* @param msgs 消息集合
* @param processQueue 消费进程队列快照
* @param messageQueue 消息队列
* @param dispatchToConsume 发送到消费者
*/
@Override
public void submitConsumeRequest(
final List<MessageExt> msgs,
final ProcessQueue processQueue,
final MessageQueue messageQueue,
final boolean dispatchToConsume) {
final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
// 消息数量 小于等于 批量消费量
if (msgs.size() <= consumeBatchSize) {
// 消费请求。
ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
try {
this.consumeExecutor.submit(consumeRequest);
} catch (RejectedExecutionException e) {
this.submitConsumeRequestLater(consumeRequest);
}
} else { // 大于则遍历消费
for (int total = 0; total < msgs.size(); ) {
List<MessageExt> msgThis = new ArrayList<MessageExt>(consumeBatchSize);
for (int i = 0; i < consumeBatchSize; i++, total++) {
if (total < msgs.size()) {
msgThis.add(msgs.get(total));
} else {
break;
}
}
// 消费请求。
ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
try {
this.consumeExecutor.submit(consumeRequest);
} catch (RejectedExecutionException e) {
for (; total < msgs.size(); total++) {
msgThis.add(msgs.get(total));
}
this.submitConsumeRequestLater(consumeRequest);
}
}
}
}new ConsumeRequest(msgThis, processQueue, messageQueue);
class ConsumeRequest implements Runnable {
private final List<MessageExt> msgs;
private final ProcessQueue processQueue;
private final MessageQueue messageQueue;
public ConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue) {
this.msgs = msgs;
this.processQueue = processQueue;
this.messageQueue = messageQueue;
}
public List<MessageExt> getMsgs() {
return msgs;
}
public ProcessQueue getProcessQueue() {
return processQueue;
}
@Override
public void run() {
if (this.processQueue.isDropped()) {
log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
return;
}
// 并发消费消息监听类
MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
// 消费者并发消费上下文
ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
// 消费状态
ConsumeConcurrentlyStatus status = null;
// 构建消费者并发消费上下文
ConsumeMessageContext consumeMessageContext = null;
// 消费消息钩子集合不为空
if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext = new ConsumeMessageContext();
consumeMessageContext.setConsumerGroup(defaultMQPushConsumer.getConsumerGroup());
consumeMessageContext.setProps(new HashMap<String, String>());
consumeMessageContext.setMq(messageQueue);
consumeMessageContext.setMsgList(msgs);
consumeMessageContext.setSuccess(false);
// 执行消费前消息消费钩子。主要是记录消息轨迹
ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
}
// 开始消费时间戳
long beginTimestamp = System.currentTimeMillis();
boolean hasException = false;
ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
try {
// 重置重试主题
ConsumeMessageConcurrentlyService.this.resetRetryTopic(msgs);
if (msgs != null && !msgs.isEmpty()) {
for (MessageExt msg : msgs) {
// 设置消费开始时间戳
MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis()));
}
}
// 消费消息，交给消息监听器 ======》核心
status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
} catch (Throwable e) {
log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
RemotingHelper.exceptionSimpleDesc(e),
ConsumeMessageConcurrentlyService.this.consumerGroup,
msgs,
messageQueue);
hasException = true;
}
// 消费耗时时间戳
long consumeRT = System.currentTimeMillis() - beginTimestamp;
// 设置消费返回状态
if (null == status) {
if (hasException) {
returnType = ConsumeReturnType.EXCEPTION;
} else {
returnType = ConsumeReturnType.RETURNNULL;
}
} else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
returnType = ConsumeReturnType.TIME_OUT;
} else if (ConsumeConcurrentlyStatus.RECONSUME_LATER == status) {
returnType = ConsumeReturnType.FAILED;
} else if (ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status) {
returnType = ConsumeReturnType.SUCCESS;
}
// 消费消息钩子集合不为空 记录消费者消费上下文状态
if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
}
if (null == status) {
log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}",
ConsumeMessageConcurrentlyService.this.consumerGroup,
msgs,
messageQueue);
status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
// 消费消息钩子集合不为空
if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
consumeMessageContext.setStatus(status.toString());
consumeMessageContext.setSuccess(ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status);
// 执行消费前消息消费钩子。主要是记录消息轨迹
ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
}
// 消费统计管理，记录消息消费耗时
ConsumeMessageConcurrentlyService.this.getConsumerStatsManager()
.incConsumeRT(ConsumeMessageConcurrentlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);
// 消费队列进程是否存在？？
if (!processQueue.isDropped()) {
// 处理消费结果 ======》核心
ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
} else {
log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
}
}
public MessageQueue getMessageQueue() {
return messageQueue;
}
}在继续看 this.processConsumeResult(status, context, this); 比较核心的代码
public void processConsumeResult(
final ConsumeConcurrentlyStatus status,
final ConsumeConcurrentlyContext context,
final ConsumeRequest consumeRequest
) {
// 确认消息是否消费成功
int ackIndex = context.getAckIndex();
// 消息为空直接返回
if (consumeRequest.getMsgs().isEmpty())
return;
// 记录消费者TPS
switch (status) {
case CONSUME_SUCCESS:
if (ackIndex >= consumeRequest.getMsgs().size()) {
ackIndex = consumeRequest.getMsgs().size() - 1;
}
int ok = ackIndex + 1;
int failed = consumeRequest.getMsgs().size() - ok;
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), ok);
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), failed);
break;
case RECONSUME_LATER:
ackIndex = -1;
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(),
consumeRequest.getMsgs().size());
break;
default:
break;
}
// 通过消费姿势来处理消费消息消费结果
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING: // 广播消费
for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
MessageExt msg = consumeRequest.getMsgs().get(i);
log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
}
break;
case CLUSTERING: // 集群消费
// 记录失败消息
List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());
for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
MessageExt msg = consumeRequest.getMsgs().get(i);
// 消费失败，重试消费消息
boolean result = this.sendMessageBack(msg, context);
if (!result) {
msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
msgBackFailed.add(msg);
}
}
// 失败消息不为空
if (!msgBackFailed.isEmpty()) {
// 从消费请求消息集合移除失败消息
consumeRequest.getMsgs().removeAll(msgBackFailed);
// 稍后提交消费请求
this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
}
break;
default:
break;
}
long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
// 偏移量大于 0 并且队列没有被删除 更新偏移量
if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
}
}到这里如果都能正常消费成功，消费者消费消息算是有一个结束了。
那如果消费失败，就进入到后面的程序了。重试消费。重试消费后面开个章节在去刨根问底。。
这个消费者主要是客户端调用做的一些事情。
那服务端接收到消费者的拉取请求去做了什么事呢？这些到高级再去研究它。。。
感谢阅读支持ღ( ´･ᴗ･` )比心
微信扫一扫，赞他~
版权声明：本文为weixin_43439073原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。