一、背景
今天准备写了关于Consumer获取消息的流程，即pullMessage()。结果在看到拉取消息的报文包装类PullRequest时，它其中有个队列消费快照ProcessQueue，进入源码一看，卧槽，500多行代码。看来有必要专门分析一下它了，也就有了今天的故事。
二、对ProcessQueue的疑问？
我们发现在消息消费的过程中很多地方都用到了ProcessQueue，比如：PullMessageService、DefaultMQPushConsumerImpl、ConsumerMessageOrderlyService。
我们思考一下：ProcessQueue是干什么的？它提供了哪些能力/方法？哪些地方用到了它？因为什么用到它？
三、ProcessQueue介绍
1、ProcessQueue简介
源码上关于ProcessQueue的注释：Queue consumption snapshot。
言简意赅，就很nice。
它是一个队列消费快照，消息拉取的时候，会将实际消息体、拉取相关的操作存放在其中。比如消费进度，消费等等功能的底层核心数据保存都是有ProcessQueue提供。
另外在拉取消息的时候使用的是PullRequest去请求，其结构如下：
可以看到，ProcessQueue和一个MessageQueue是对应的，MessageQueue用来表示拉取回来的消息元数据信息：topic、brokerName、queueId。
ProcessQueue几个重要的成员变量：
/**
* 读写锁：对消息的操作都要使用
*/
private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
/**
* 临时存放消息用的
*/
private final TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>();
/**
* 消息总数量
*/
private final AtomicLong msgCount = new AtomicLong();
/**
* 整个ProcessQueue处理单元的总消息长度
*/
private final AtomicLong msgSize = new AtomicLong();
private final Lock lockConsume = new ReentrantLock();
/**
* A subset of msgTreeMap, will only be used when orderly consume
* 一个临时的TreeMap，仅在顺序消费模式下使用
*/
private final TreeMap<Long, MessageExt> consumingMsgOrderlyTreeMap = new TreeMap<Long, MessageExt>();
private final AtomicLong tryUnlockTimes = new AtomicLong(0);
/**
* 整个ProcessQueue处理单元的offset最大边界
*/
private volatile long queueOffsetMax = 0L;
/**
* 是否正在消费
*/
private volatile boolean consuming = false;
/**
* broker端还有多少条消息没被处理（拉取消息的那一刻）
*/
private volatile long msgAccCnt = 0;
注意：每次对msgTreeMap、msgCount、msgSize进行操作时，都会加上读写锁lockTreeMap。
2、包含的方法
我们主要看下图框中的方法：
1）isLockExpired()
该方法是在处理消息时，判断拥有的锁是否过期，默认过期时间30s。
public boolean isLockExpired() {
return (System.currentTimeMillis() - this.lastLockTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
}
public final static long REBALANCE_LOCK_MAX_LIVE_TIME =
Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockMaxLiveTime", "30000"));
那么该方法是在哪些地方使用到了呢？
从上图我们可以看出，isLockExpired()方法在顺序消费时使用，用来判断向broker请求消息时，对ProcessQueue锁定时间是否超过阈值(默认30000ms，可以通过rocketmq.client.rebalance.lockMaxLiveTime设置)，如果没有超时，代表还是持有锁，超过这个时间则失效。
2）isPullExpired()
判断拉取消息是否过期，默认两分钟（可以通过rocketmq.client.pull.pullMaxIdleTime设置）
public boolean isLockExpired() {
return (System.currentTimeMillis() - this.lastLockTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
}
private final static long PULL_MAX_IDLE_TIME = Long.parseLong(System.getProperty("rocketmq.client.pull.pullMaxIdleTime", "120000"));
那么该方法是在哪些地方使用到了呢？
在RebalanceImpl#updateProcessQueueTableInRebalance()方法中，如下所示：
if (mq.getTopic().equals(topic)) {
if (!mqSet.contains(mq)) {
.......
} else if (pq.isPullExpired()) {
.......
}
}
在负载均衡更新ProcessQueueTable时调用，如果拉取失效，ProcessQueue将被丢弃。
3）cleanExpiredMsg()
清理过期的消息，规则如下：
- 最多16条消息一处理；
- 消息在客户端存在
超过15分钟就被认为已过期，然后从本地缓存中移除，以10s的延时消息方式发送会Broker。
**注意：**顺序消费不清理过期消息，进并发消费模式才清理过期消息。
public void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer) {
// 顺序消费不清理过期消息
if (pushConsumer.getDefaultMQPushConsumerImpl().isConsumeOrderly()) {
return;
}
// 每次最多清理16条消息
int loop = msgTreeMap.size() < 16 ? msgTreeMap.size() : 16;
for (int i = 0; i < loop; i++) {
MessageExt msg = null;
try {
// 处理消息之前，先拿到读锁
this.lockTreeMap.readLock().lockInterruptibly();
try {
// 临时存放消息的treeMap不为空，并且 判断当前时间 - TreeMap里第一条消息的开始消费时间 > 15分钟
if (!msgTreeMap.isEmpty() && System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msgTreeMap.firstEntry().getValue())) > pushConsumer.getConsumeTimeout() * 60 * 1000) {
// 把第一条消息拿出来
msg = msgTreeMap.firstEntry().getValue();
} else {
break;
}
} finally {
// 释放读锁
this.lockTreeMap.readLock().unlock();
}
} catch (InterruptedException e) {
log.error("getExpiredMsg exception", e);
}
try {
// 把过期消息以延时消息方式重新发给broker，10s之后才能消费。
pushConsumer.sendMessageBack(msg, 3);
log.info("send expire msg back. topic={}, msgId={}, storeHost={}, queueId={}, queueOffset={}", msg.getTopic(), msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(), msg.getQueueOffset());
try {
// 获取写锁
this.lockTreeMap.writeLock().lockInterruptibly();
try {
if (!msgTreeMap.isEmpty() && msg.getQueueOffset() == msgTreeMap.firstKey()) {
try {
// 将过期消息从本地缓存中的消息列表中移除掉， Collections.singletonList表示只有一个元素的List集合
removeMessage(Collections.singletonList(msg));
} catch (Exception e) {
log.error("send expired msg exception", e);
}
}
} finally {
// 释放写锁
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("getExpiredMsg exception", e);
}
} catch (Exception e) {
log.error("send expired msg exception", e);
}
}
}
方法被调用的地方：
// 启动定时任务（每15分钟跑一次），清理过期消息
public void start() {
this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
cleanExpireMsg();
}
}, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
}
// 针对每个ProcessQueue每次最多清理16条过期消息
private void cleanExpireMsg() {
Iterator<Map.Entry<MessageQueue, ProcessQueue>> it =
this.defaultMQPushConsumerImpl.getRebalanceImpl().getProcessQueueTable().entrySet().iterator();
while (it.hasNext()) {
Map.Entry<MessageQueue, ProcessQueue> next = it.next();
ProcessQueue pq = next.getValue();
pq.cleanExpiredMsg(this.defaultMQPushConsumer);
}
}
并发消费ConsumeMessageConcurrentlyService#start()方法中启动了一个定时任务，每15分钟调用一次cleanExpiredMsg()清理过期消息：超过15分钟未被消费，针对每个ProcessQueue每次最多清理16条消息。
4）putMessage()
用来初始化属性：msgTreeMap（存放拉取到的消息）、queueOffsetMax（消费到的最大offset）、msgSize（ProcessQueue中的总消息长度）、msgCount（ProcessQueue中的总消息数量）、msgAccCnt（表示broker端还有多少消息未被消费）。
public boolean putMessage(final List<MessageExt> msgs) {
// 这个只有在顺序消费的时候才会遇到，并发消费不会用到
boolean dispatchToConsume = false;
try {
this.lockTreeMap.writeLock().lockInterruptibly();
try {
//有效消息数量
int validMsgCnt = 0;
for (MessageExt msg : msgs) {
// 把传过来的消息都都放在msgTreeMap中，以消息在queue中的offset作为key，msg做为value
MessageExt old = msgTreeMap.put(msg.getQueueOffset(), msg);
// 正常情况，说明原本msgTreeMap中不包含此条消息
if (null == old) {
validMsgCnt++;
// 将最后一个消息的offset赋值给queueOffsetMax
this.queueOffsetMax = msg.getQueueOffset();
// 把当前消息的长度加到msgSize中
msgSize.addAndGet(msg.getBody().length);
}
}
// 增加有效消息数量
msgCount.addAndGet(validMsgCnt);
// msgTreeMap不为空(含有消息)，并且不是正在消费状态
// // 这个值在放消息的时候会设置为true，在顺序消费模式，取不到消息则设置为false
if (!msgTreeMap.isEmpty() && !this.consuming) {
// 将ProcessQueue置为正在被消费状态
// 有消息，且为未消费状态，则顺序消费模式可以消费
dispatchToConsume = true;
this.consuming = true;
}
if (!msgs.isEmpty()) {
// 拿到最后一条消息
MessageExt messageExt = msgs.get(msgs.size() - 1);
// 获取broker端（拉取消息时）queue里最大的offset，maxOffset会存在每条消息里
String property = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);
// 计算broker端还有多少条消息没有被消费
if (property != null) {
// broker端的最大偏移量 - 当前ProcessQueue中处理的最大消息偏移量
long accTotal = Long.parseLong(property) - messageExt.getQueueOffset();
if (accTotal > 0) {
this.msgAccCnt = accTotal;
}
}
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("putMessage exception", e);
}
return dispatchToConsume;
}
方法被调用的地方：
MQClientInstance#start()方法中调用pullMessageService#start()，
PullMessageService本身是一个ServiceThread，它的run()方法中会调用DefaultMQPushConsumerImpl#pullMessage()方法，
然后在其中new出的PullCallBack的onSuccess()方法中：
- 当传入的pullResult的pullStatus是FOUND时，并且PullRequest的msgFoundList不为空时 会调用该putMessage()方法。
在拉取消息后处理消息时，在ProcessQueue中通过putMessage()方法，往红黑树msgTreeMap存储一分消息数据，并统计消息总数量、消息总长度、broker端还有多少消息未被拉取。返回结果dispatchToConsume如果为true，表示消费集合有数据，顺序消息会据此构造消费请求继续处理。
5）getMaxSpan()
返回processQueue中处理的一批消息中最大offset和最小offset之间的差距。
public long getMaxSpan() {
try {
this.lockTreeMap.readLock().lockInterruptibly();
try {
if (!this.msgTreeMap.isEmpty()) {
// msgTreeMap中最后一个消息的offset - 第一个消息的offset
return this.msgTreeMap.lastKey() - this.msgTreeMap.firstKey();
}
} finally {
this.lockTreeMap.readLock().unlock();
}
} catch (InterruptedException e) {
log.error("getMaxSpan exception", e);
}
return 0;
}
方法被使用的地方：
DefaultMQPushConsumerImpl#pullMessage()方法中，会根据这个判断当前还有多少消息未处理，如果大于某个值(默认2000)，则进行流控处理：将PullRequest先存起来，默认过50ms再执行拉消息逻辑。
6）removeMessage()
该方法将从ProcessQueue中移除部分消息，并行消费模式中使用。
public long removeMessage(final List<MessageExt> msgs) {
// 当前消费进度的下一个offset
long result = -1;
final long now = System.currentTimeMillis();
try {
// 获取写锁
this.lockTreeMap.writeLock().lockInterruptibly();
// 最后消费的时间
this.lastConsumeTimestamp = now;
try {
if (!msgTreeMap.isEmpty()) {
result = this.queueOffsetMax + 1;
// 删除的消息个数
int removedCnt = 0;
// 遍历消息，将其从Treemap中移除
for (MessageExt msg : msgs) {
MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
// 不为null，说明删除成功
if (prev != null) {
// 已经移除的消息数量
removedCnt--;
// ProcessQueue中消息的长度 - 当前消息长度
msgSize.addAndGet(0 - msg.getBody().length);
}
}
// ProcessQueue中的消息数量 - 删除的消息数量，即加上removedCnt（是一个负数）
msgCount.addAndGet(removedCnt);
// 如果还有消息存在，则使用msgTreeMap中的第一消息的offset（即最小offset）作为消费进度
if (!msgTreeMap.isEmpty()) {
result = msgTreeMap.firstKey();
}
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (Throwable t) {
log.error("removeMessage exception", t);
}
return result;
}
方法被调用的地方：
ConsumeMessageConcurrentlyService#processConsumeResult()中
removeMessage()方法在并发消费后进行调用，消息处理完之后将ProcessQueue中msgTreeMap红黑树的这批消息移除。
以下的方法全部是在顺序消费模式下使用
7）takeMessages()
该方法在顺序消费模式下使用，取到消息后，就会调用我们定义的MessageListener进行消费。
调用takeMessages从msgTreeMap中获取消息后：其内部会将消息都放在一个临时的TreeMap中，然后进行顺序消费。
public List<MessageExt> takeMessages(final int batchSize) {
List<MessageExt> result = new ArrayList<MessageExt>(batchSize);
final long now = System.currentTimeMillis();
try {
this.lockTreeMap.writeLock().lockInterruptibly();
this.lastConsumeTimestamp = now;
try {
if (!this.msgTreeMap.isEmpty()) {
// 从 msgTreeMap中获取batchSize条数据，每次都返回offset最小的那条消息并从msgTreeMap中移除
for (int i = 0; i < batchSize; i++) {
Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
if (entry != null) {
// 把消息放到返回列表
result.add(entry.getValue());
// 把消息的offset和消息体msg，放到顺序消费TreeMap中
consumingMsgOrderlyTreeMap.put(entry.getKey(), entry.getValue());
} else {
break;
}
}
}
// 没有取到消息，说明不需要消费，即将consuming置为FALSE。
if (result.isEmpty()) {
consuming = false;
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("take Messages exception", e);
}
return result;
}
方法被调用的地方：
ConsumeMessageOrderlyService.ConsumeRequest#run()方法中：
List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
递归往上层找，找到最终调用的地方：
在ConsumeMessageOrderlyService#submitConsumeRequest()方法中会new一个ConsumeRequest（是一个Runnable实现类），然后将其放到线程中执行；即将消息丢入消费服务线程中进行消费
而submitConsumeRequest()方法被调用的场景有两种：
1. 集群消费模式下调用submitConsumeRequestLater方法（位于ConsumeMessageService的两个实现类中）重新提交client端消费请求时，
2. DefaultMQPushConsumerImpl#pullMessage()中拉取消息成功之后，将消息丢入消费服务线程中进行消费。
8）commit()
在顺序消费模式下，调用takeMessages从msgTreeMap中获取消息：其内部会将消息都放在一个临时的TreeMap中，然后进行消费。消费完消息之后，我们需要调用commit()方法将这个临时的TreeMap清除。
public long commit() {
try {
this.lockTreeMap.writeLock().lockInterruptibly();
try {
// 获取临时TreeMap中最后一个消息的offset
Long offset = this.consumingMsgOrderlyTreeMap.lastKey();
// 消费完成之后，减去该批次的消息数量
msgCount.addAndGet(0 - this.consumingMsgOrderlyTreeMap.size());
// 维护ProcessQueue中的总消息长度
for (MessageExt msg : this.consumingMsgOrderlyTreeMap.values()) {
// 减去每条已消费消息的长度
msgSize.addAndGet(0 - msg.getBody().length);
}
// 清除临时TreeMap中的所有消息
this.consumingMsgOrderlyTreeMap.clear();
// 返回下一个消费进度
if (offset != null) {
return offset + 1;
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("commit exception", e);
}
return -1;
}
方法被调用的地方：
ConsumeMessageOrderlyService#processConsumeResult()方法中
if (context.isAutoCommit()) {
switch (status) {
case SUCCESS:
//清空msgTreeMapTemp
commitOffset = consumeRequest.getProcessQueue().commit();
......
}
} else {
switch (status) {
case COMMIT:
//清空msgTreeMapTemp
commitOffset = consumeRequest.getProcessQueue().commit();
break;
}
}
9）rollback()
该方法把临时TreeMap中的消息全部放到msgTreeMap中，等待下一次消费。
public void rollback() {
try {
this.lockTreeMap.writeLock().lockInterruptibly();
try {
// 把临时TreeMap中的消息全部放到msgTreeMap中，等待下一次消费
this.msgTreeMap.putAll(this.consumingMsgOrderlyTreeMap);
this.consumingMsgOrderlyTreeMap.clear();
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("rollback exception", e);
}
}
方法被调用的地方：
ConsumerMessageOrderlyService#processConsumeResult()中
case ROLLBACK:
consumeRequest.getProcessQueue().rollback();
在顺序消费客户端处理消息后，如果消息处理结果的状态为ROLLBACK，此时调用ProcessQueue#rollback方法；将msgTreeMapTmp中的消息重新写回msgTreeMap中；ROLLBACK状态在顺序消费中已不建议使用。
10）makeMessageToConsumeAgain()
顺序消费模式下，使消息可以被重新消费。
public void makeMessageToConsumeAgain(List<MessageExt> msgs) {
try {
this.lockTreeMap.writeLock().lockInterruptibly();
try {
for (MessageExt msg : msgs) {
// 从临时TreeMap中取出消息，
this.consumingMsgOrderlyTreeMap.remove(msg.getQueueOffset());
//再将消息放到msgTreeMap中
this.msgTreeMap.put(msg.getQueueOffset(), msg);
}
} finally {
this.lockTreeMap.writeLock().unlock();
}
} catch (InterruptedException e) {
log.error("makeMessageToCosumeAgain exception", e);
}
}
方法被调用的地方：
ConsumerMessageOrderlyService#processConsumeResult()
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToConsumeAgain(msgs);
.....
}
makeMessageToCosumeAgain()方法在顺序消费模式下，当客户端返回消息状态为SUSPEND_CURRENT_QUEUE_A_MOMENT时调用；其将消息从msgTreeMapTemp移除，并将该批消息重新放入msgTreeMap，使这批消息可以被再次消费。
三、总结
ProcessQueue作为MessageQueue在Consumer端的镜像，从负载均衡、消息拉取、消费状态处理、offset提交等功能层面，控制着整个消费的脉搏，尤其是在顺序消费中。
还没有评论，来说两句吧...