顺序消息

this.consumeExecutor = new ThreadPoolExecutor(this.defaultMQPushConsumer.getConsumeTh readMin(), this.defaultMQPushConsumer.getConsumeThreadMax(), 1000 * 60, TimeUnit.MILLISECONDS, this.consumeRequestQueue, new ThreadFactoryImpl("ConsumeMessageThread_"));
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));

public void start() {

    if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                     public void run() {
                         ConsumeMessageOrderlyService.this.lockMQPeriodically();
                     }
                 },
            1000 * 1,
            ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
    }
}

private HashMap<String/* brokerName */, Set<MessageQueue>> buildProcessQueueTableByBrokerName() {
    HashMap<String, Set<MessageQueue>> result = new HashMap<String, Set<MessageQueue>>();
    for (MessageQueue mq : this.processQueueTable.keySet()) {
        Set<MessageQueue> mqs = result.get(mq.getBrokerName());
        if (null == mqs) {
            mqs = new HashSet<MessageQueue>();
            result.put(mq.getBrokerName(), mqs);
        }
        mqs.add(mq);
    }
    return result;
}

LockBatchRequestBody requestBody = new LockBatchRequestBody();
requestBody.setConsumerGroup(this.consumerGroup);
requestBody.setClientId(this.mQClientFactory.getClientId());
requestBody.setMqSet(mqs);
Set<MessageQueue> lockOKMQSet = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);

for (MessageQueue mq : lockOKMQSet) {
    ProcessQueue processQueue = this.processQueueTable.get(mq);
    if (processQueue != null) {
        if (!processQueue.isLocked()) {
            log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);
        }
        processQueue.setLocked(true);
        processQueue.setLastLockTimestamp(System.currentTimeMillis());
    }
}

for (MessageQueue mq : mqs) {
    if (!lockOKMQSet.contains(mq)) {
        ProcessQueue processQueue = this.processQueueTable.get(mq);
        if (processQueue != null) {
            processQueue.setLocked(false);
            log.warn("the mes sage queue locked Failed, Group:{} {}", this.consumerGroup, mq);
        }
    }
}

public void submitConsumeRequest ( final List<MessageExt> msgs, final ProcessQueue processQueue,final MessageQueue messageQueue, final boolean dispathToConsume){
    if (dispathToConsume) {
        ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
        this.consumeExecutor.submit(consumeRequest);
    }
}

if (this.processQueue.isDropped()) {
    log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
    return;
}

final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
synchronized (objLock) {

if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) ||(this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
    // 消息消费逻辑
} else {
    if (this.processQueue.isDropped()) {
        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
        return;
    }
    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
}

final long beginTime = System.currentTimeMillis();
for (boolean continueConsume = true; continueConsume; ) {
    // 省略相关代码
    long interval = System.currentTimeMillis() - beginTime;
    if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
        ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
        break;
    }
}

final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);

long beginTimestamp = System.currentTimeMillis();
ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
boolean hasException = false;
try {
    this.processQueue.getLockConsume().lock();
    if (this.processQueue.isDropped()) {
        log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
        break;
    }
    status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);

} catch (Throwable e) {
    hasException = true;
} finally {
    this.processQueue.getLockConsume().unlock();
}

public long commit() {
    try {
        this.lockTreeMap.writeLock().lockInterruptibly();
        try {
            Long offset = this.msgTreeMapTemp.lastKey();
            msgCount.addAndGet(this.msgTreeMapTemp.size() * (-1));
            this.msgTreeMapTemp.clear();
            if (offset != null) {
                return offset + 1;
            }
        } finally {
            this.lockTreeMap.writeLock().unlock();
        }
    } catch (InterruptedException e) {
        log.error("commit exception", e);
    }
    return -1;
}

if (checkReconsumeTimes(msgs)) {
    consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
    this.submitConsumeRequestLater(consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue(), context.getSuspendCurrentQueueTimeMillis());
    continueConsume = false;
} else {
    commitOffset = consumeRequest.getProcessQueue().commit();
}