| /** |
| * Copyright (C) 2013 kangliqiang ,kangliq@163.com |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| #include <list> |
| #include <string> |
| |
| #include "ConsumeMessageOrderlyService.h" |
| #include "DefaultMQPushConsumerImpl.h" |
| #include "MQClientFactory.h" |
| #include "DefaultMQProducer.h" |
| #include "MessageListener.h" |
| #include "MessageQueue.h" |
| #include "RebalanceImpl.h" |
| #include "DefaultMQPushConsumer.h" |
| #include "OffsetStore.h" |
| #include "ScopedLock.h" |
| #include "KPRUtil.h" |
| #include "MixAll.h" |
| #include "UtilAll.h" |
| |
| namespace rmq |
| { |
| |
| class LockMq : public kpr::TimerHandler |
| { |
| public: |
| LockMq(ConsumeMessageOrderlyService* pService) |
| : m_pService(pService) |
| { |
| |
| } |
| |
| void OnTimeOut(unsigned int timerID) |
| { |
| m_pService->lockMQPeriodically(); |
| |
| // can not delete |
| //delete this; |
| } |
| |
| private: |
| ConsumeMessageOrderlyService* m_pService; |
| }; |
| |
| class SubmitConsumeRequestLaterOrderly : public kpr::TimerHandler |
| { |
| public: |
| SubmitConsumeRequestLaterOrderly(ProcessQueue* pProcessQueue, |
| const MessageQueue& messageQueue, |
| ConsumeMessageOrderlyService* pService) |
| : m_pProcessQueue(pProcessQueue), |
| m_messageQueue(messageQueue), |
| m_pService(pService) |
| { |
| |
| } |
| |
| void OnTimeOut(unsigned int timerID) |
| { |
| try |
| { |
| std::list<MessageExt*> msgs; |
| m_pService->submitConsumeRequest(msgs, m_pProcessQueue, m_messageQueue, true); |
| } |
| catch(...) |
| { |
| RMQ_ERROR("SubmitConsumeRequestLaterOrderly OnTimeOut exception"); |
| } |
| |
| delete this; |
| } |
| |
| private: |
| ProcessQueue* m_pProcessQueue; |
| MessageQueue m_messageQueue; |
| ConsumeMessageOrderlyService* m_pService; |
| }; |
| |
| |
| class TryLockLaterAndReconsume : public kpr::TimerHandler |
| { |
| public: |
| TryLockLaterAndReconsume(ProcessQueue* pProcessQueue, |
| MessageQueue& messageQueue, |
| ConsumeMessageOrderlyService* pService) |
| : m_pProcessQueue(pProcessQueue), |
| m_messageQueue(messageQueue), |
| m_pService(pService) |
| { |
| |
| } |
| |
| void OnTimeOut(unsigned int timerID) |
| { |
| try |
| { |
| bool lockOK = m_pService->lockOneMQ(m_messageQueue); |
| if (lockOK) |
| { |
| m_pService->submitConsumeRequestLater(m_pProcessQueue, m_messageQueue, 10); |
| } |
| else |
| { |
| m_pService->submitConsumeRequestLater(m_pProcessQueue, m_messageQueue, 3000); |
| } |
| } |
| catch(...) |
| { |
| RMQ_ERROR("TryLockLaterAndReconsume OnTimeOut exception"); |
| } |
| |
| delete this; |
| } |
| |
| private: |
| ProcessQueue* m_pProcessQueue; |
| MessageQueue m_messageQueue; |
| ConsumeMessageOrderlyService* m_pService; |
| }; |
| |
| |
| |
| ConsumeMessageOrderlyService::ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl* pDefaultMQPushConsumerImpl, |
| MessageListenerOrderly* pMessageListener) |
| { |
| m_stoped = false; |
| m_pDefaultMQPushConsumerImpl = pDefaultMQPushConsumerImpl; |
| m_pMessageListener = pMessageListener; |
| m_pDefaultMQPushConsumer = m_pDefaultMQPushConsumerImpl->getDefaultMQPushConsumer(); |
| m_consumerGroup = m_pDefaultMQPushConsumer->getConsumerGroup(); |
| m_pConsumeExecutor = new kpr::ThreadPool("ConsumeMessageThreadPool", 1, |
| m_pDefaultMQPushConsumer->getConsumeThreadMin(), m_pDefaultMQPushConsumer->getConsumeThreadMax()); |
| m_scheduledExecutorService = new kpr::TimerThread("ConsumeMessageConcurrentlyService", 10); |
| } |
| |
| ConsumeMessageOrderlyService::~ConsumeMessageOrderlyService() |
| { |
| } |
| |
| |
| void ConsumeMessageOrderlyService::start() |
| { |
| m_scheduledExecutorService->Start(); |
| |
| LockMq* lm = new LockMq(this); |
| m_scheduledExecutorService->RegisterTimer(0, ProcessQueue::s_RebalanceLockInterval, lm, true); |
| } |
| |
| void ConsumeMessageOrderlyService::shutdown() |
| { |
| m_stoped = true; |
| m_pConsumeExecutor->Destroy(); |
| m_scheduledExecutorService->Stop(); |
| m_scheduledExecutorService->Join(); |
| unlockAllMQ(); |
| } |
| |
| void ConsumeMessageOrderlyService::unlockAllMQ() |
| { |
| m_pDefaultMQPushConsumerImpl->getRebalanceImpl()->unlockAll(false); |
| } |
| |
| void ConsumeMessageOrderlyService::lockMQPeriodically() |
| { |
| if (!m_stoped) |
| { |
| m_pDefaultMQPushConsumerImpl->getRebalanceImpl()->lockAll(); |
| } |
| } |
| |
| bool ConsumeMessageOrderlyService::lockOneMQ(MessageQueue& mq) |
| { |
| if (!m_stoped) |
| { |
| return m_pDefaultMQPushConsumerImpl->getRebalanceImpl()->lock(mq); |
| } |
| |
| return false; |
| } |
| |
| void ConsumeMessageOrderlyService::tryLockLaterAndReconsume(MessageQueue& messageQueue, |
| ProcessQueue* pProcessQueue, |
| long long delayMills) |
| { |
| TryLockLaterAndReconsume* consume = new TryLockLaterAndReconsume(pProcessQueue, messageQueue, this); |
| m_scheduledExecutorService->RegisterTimer(0, int(delayMills), consume, false); |
| } |
| |
| ConsumerStat& ConsumeMessageOrderlyService::getConsumerStat() |
| { |
| return m_pDefaultMQPushConsumerImpl->getConsumerStatManager()->getConsumertat(); |
| } |
| |
| void ConsumeMessageOrderlyService::submitConsumeRequestLater(ProcessQueue* pProcessQueue, |
| const MessageQueue& messageQueue, |
| long long suspendTimeMillis) |
| { |
| long timeMillis = long(suspendTimeMillis); |
| if (timeMillis < 10) |
| { |
| timeMillis = 10; |
| } |
| else if (timeMillis > 30000) |
| { |
| timeMillis = 30000; |
| } |
| |
| SubmitConsumeRequestLaterOrderly* sc = new SubmitConsumeRequestLaterOrderly(pProcessQueue, messageQueue, this); |
| m_scheduledExecutorService->RegisterTimer(0, timeMillis, sc, false); |
| } |
| |
| void ConsumeMessageOrderlyService::submitConsumeRequest(std::list<MessageExt*>& msgs, |
| ProcessQueue* pProcessQueue, |
| const MessageQueue& messageQueue, |
| bool dispathToConsume) |
| { |
| if (dispathToConsume) |
| { |
| kpr::ThreadPoolWorkPtr consumeRequest = new ConsumeOrderlyRequest(pProcessQueue, messageQueue, this); |
| m_pConsumeExecutor->AddWork(consumeRequest); |
| } |
| } |
| |
| void ConsumeMessageOrderlyService::updateCorePoolSize(int corePoolSize) |
| { |
| } |
| |
| |
| std::string& ConsumeMessageOrderlyService::getConsumerGroup() |
| { |
| return m_consumerGroup; |
| } |
| |
| MessageListenerOrderly* ConsumeMessageOrderlyService::getMessageListener() |
| { |
| return m_pMessageListener; |
| } |
| |
| DefaultMQPushConsumerImpl* ConsumeMessageOrderlyService::getDefaultMQPushConsumerImpl() |
| { |
| return m_pDefaultMQPushConsumerImpl; |
| } |
| |
| bool ConsumeMessageOrderlyService::processConsumeResult(std::list<MessageExt*>& msgs, |
| ConsumeOrderlyStatus status, |
| ConsumeOrderlyContext& context, |
| ConsumeOrderlyRequest& consumeRequest) |
| { |
| bool continueConsume = true; |
| long long commitOffset = -1L; |
| int msgsSize = msgs.size(); |
| |
| if (context.autoCommit) |
| { |
| switch (status) |
| { |
| case COMMIT: |
| case ROLLBACK: |
| RMQ_WARN("the message queue consume result is illegal, we think you want to ack these message: %s", |
| consumeRequest.getMessageQueue().toString().c_str()); |
| case SUCCESS: |
| getConsumerStat().consumeMsgOKTotal.fetchAndAdd(msgsSize); |
| commitOffset = consumeRequest.getProcessQueue()->commit(); |
| break; |
| case SUSPEND_CURRENT_QUEUE_A_MOMENT: |
| getConsumerStat().consumeMsgFailedTotal.fetchAndAdd(msgsSize); |
| if (checkReconsumeTimes(msgs)) |
| { |
| consumeRequest.getProcessQueue()->makeMessageToCosumeAgain(msgs); |
| submitConsumeRequestLater(consumeRequest.getProcessQueue(), |
| consumeRequest.getMessageQueue(), |
| context.suspendCurrentQueueTimeMillis); |
| continueConsume = false; |
| } |
| else |
| { |
| commitOffset = consumeRequest.getProcessQueue()->commit(); |
| } |
| |
| break; |
| default: |
| break; |
| } |
| } |
| else |
| { |
| switch (status) |
| { |
| case SUCCESS: |
| getConsumerStat().consumeMsgOKTotal.fetchAndAdd(msgsSize); |
| break; |
| case COMMIT: |
| commitOffset = consumeRequest.getProcessQueue()->commit(); |
| break; |
| case ROLLBACK: |
| consumeRequest.getProcessQueue()->rollback(); |
| submitConsumeRequestLater(consumeRequest.getProcessQueue(), |
| consumeRequest.getMessageQueue(), |
| context.suspendCurrentQueueTimeMillis); |
| continueConsume = false; |
| break; |
| case SUSPEND_CURRENT_QUEUE_A_MOMENT: |
| getConsumerStat().consumeMsgFailedTotal.fetchAndAdd(msgsSize); |
| if (checkReconsumeTimes(msgs)) |
| { |
| consumeRequest.getProcessQueue()->makeMessageToCosumeAgain(msgs); |
| submitConsumeRequestLater(consumeRequest.getProcessQueue(), |
| consumeRequest.getMessageQueue(), |
| context.suspendCurrentQueueTimeMillis); |
| continueConsume = false; |
| } |
| break; |
| default: |
| break; |
| } |
| } |
| |
| if (commitOffset >= 0 && !consumeRequest.getProcessQueue()->isDropped()) |
| { |
| m_pDefaultMQPushConsumerImpl->getOffsetStore()->updateOffset(consumeRequest.getMessageQueue(), |
| commitOffset, false); |
| } |
| |
| return continueConsume; |
| } |
| |
| bool ConsumeMessageOrderlyService::checkReconsumeTimes(std::list<MessageExt*>& msgs) |
| { |
| bool suspend = false; |
| |
| if (!msgs.empty()) |
| { |
| std::list<MessageExt*>::iterator it = msgs.begin(); |
| for (; it != msgs.end(); it++) |
| { |
| MessageExt* msg = *it; |
| if (msg->getReconsumeTimes() >= m_pDefaultMQPushConsumer->getMaxReconsumeTimes()) |
| { |
| msg->putProperty(Message::PROPERTY_RECONSUME_TIME, UtilAll::toString(msg->getReconsumeTimes())); |
| |
| if (!sendMessageBack(*msg)) |
| { |
| suspend = true; |
| msg->setReconsumeTimes(msg->getReconsumeTimes() + 1); |
| } |
| } |
| else |
| { |
| suspend = true; |
| msg->setReconsumeTimes(msg->getReconsumeTimes() + 1); |
| } |
| } |
| } |
| |
| return suspend; |
| } |
| |
| bool ConsumeMessageOrderlyService::sendMessageBack(MessageExt& msg) |
| { |
| try |
| { |
| Message newMsg(MixAll::getRetryTopic(m_pDefaultMQPushConsumer->getConsumerGroup()), |
| msg.getBody(), msg.getBodyLen()); |
| |
| std::string originMsgId = msg.getProperty(Message::PROPERTY_ORIGIN_MESSAGE_ID); |
| newMsg.putProperty(Message::PROPERTY_ORIGIN_MESSAGE_ID, UtilAll::isBlank(originMsgId) ? msg.getMsgId() |
| : originMsgId); |
| |
| newMsg.setFlag(msg.getFlag()); |
| newMsg.setProperties(msg.getProperties()); |
| newMsg.putProperty(Message::PROPERTY_RETRY_TOPIC, msg.getTopic()); |
| |
| int reTimes = msg.getReconsumeTimes() + 1; |
| newMsg.putProperty(Message::PROPERTY_RECONSUME_TIME, UtilAll::toString(reTimes)); |
| newMsg.putProperty(Message::PROPERTY_MAX_RECONSUME_TIMES, UtilAll::toString(m_pDefaultMQPushConsumer->getMaxReconsumeTimes())); |
| newMsg.setDelayTimeLevel(3 + reTimes); |
| |
| m_pDefaultMQPushConsumerImpl->getmQClientFactory()->getDefaultMQProducer()->send(newMsg); |
| |
| return true; |
| } |
| catch (...) |
| { |
| RMQ_ERROR("sendMessageBack exception, group: %s, msg: %s", |
| m_consumerGroup.c_str(), msg.toString().c_str()); |
| } |
| |
| return false; |
| } |
| |
| |
| MessageQueueLock& ConsumeMessageOrderlyService::getMessageQueueLock() |
| { |
| return m_messageQueueLock; |
| } |
| |
| DefaultMQPushConsumer* ConsumeMessageOrderlyService::getDefaultMQPushConsumer() |
| { |
| return m_pDefaultMQPushConsumer; |
| } |
| |
| ConsumeOrderlyRequest::ConsumeOrderlyRequest(ProcessQueue* pProcessQueue, |
| const MessageQueue& messageQueue, |
| ConsumeMessageOrderlyService* pService) |
| { |
| m_pProcessQueue = pProcessQueue; |
| m_messageQueue = messageQueue; |
| m_pService = pService; |
| } |
| |
| ConsumeOrderlyRequest::~ConsumeOrderlyRequest() |
| { |
| } |
| |
| void ConsumeOrderlyRequest::Do() |
| { |
| if (m_pProcessQueue->isDropped()) |
| { |
| RMQ_WARN("run, the message queue not be able to consume, because it's dropped, MQ: %s", |
| m_messageQueue.toString().c_str()); |
| return; |
| } |
| |
| try |
| { |
| kpr::Mutex* objLock = m_pService->getMessageQueueLock().fetchLockObject(m_messageQueue); |
| { |
| kpr::ScopedLock<kpr::Mutex> lock(*objLock); |
| |
| MessageModel messageModel = m_pService->getDefaultMQPushConsumerImpl()->messageModel(); |
| if (BROADCASTING == messageModel |
| || (m_pProcessQueue->isLocked() || !m_pProcessQueue->isLockExpired())) |
| { |
| long long beginTime = KPRUtil::GetCurrentTimeMillis(); |
| for (bool continueConsume = true; continueConsume;) |
| { |
| if (m_pProcessQueue->isDropped()) |
| { |
| RMQ_INFO("the message queue not be able to consume, because it's droped, MQ: %s", |
| m_messageQueue.toString().c_str()); |
| break; |
| } |
| |
| if (CLUSTERING == messageModel |
| && !m_pProcessQueue->isLocked()) |
| { |
| RMQ_WARN("the message queue not locked, so consume later, MQ: %s", m_messageQueue.toString().c_str()); |
| m_pService->tryLockLaterAndReconsume(m_messageQueue, m_pProcessQueue, 10); |
| break; |
| } |
| |
| if (CLUSTERING == messageModel |
| && m_pProcessQueue->isLockExpired()) |
| { |
| RMQ_WARN("the message queue lock expired, so consume later, MQ: %s", m_messageQueue.toString().c_str()); |
| m_pService->tryLockLaterAndReconsume(m_messageQueue, m_pProcessQueue, 10); |
| break; |
| } |
| |
| long interval = long(KPRUtil::GetCurrentTimeMillis() - beginTime); |
| if (interval > ConsumeMessageOrderlyService::s_MaxTimeConsumeContinuously) |
| { |
| m_pService->submitConsumeRequestLater(m_pProcessQueue, m_messageQueue, 10); |
| break; |
| } |
| |
| int consumeBatchSize = |
| m_pService->getDefaultMQPushConsumer()->getConsumeMessageBatchMaxSize(); |
| |
| std::list<MessageExt*> msgs = m_pProcessQueue->takeMessages(consumeBatchSize); |
| if (!msgs.empty()) |
| { |
| ConsumeOrderlyContext context(m_messageQueue); |
| |
| ConsumeOrderlyStatus status = SUSPEND_CURRENT_QUEUE_A_MOMENT; |
| |
| ConsumeMessageContext consumeMessageContext; |
| if (m_pService->getDefaultMQPushConsumerImpl()->hasHook()) |
| { |
| consumeMessageContext.consumerGroup = m_pService->getConsumerGroup(); |
| consumeMessageContext.mq = m_messageQueue; |
| consumeMessageContext.msgList = msgs; |
| consumeMessageContext.success = false; |
| m_pService->getDefaultMQPushConsumerImpl()->executeHookBefore(consumeMessageContext); |
| } |
| |
| long long beginTimestamp = KPRUtil::GetCurrentTimeMillis(); |
| try |
| { |
| kpr::ScopedLock<kpr::Mutex> lock(m_pProcessQueue->getLockConsume()); |
| if (m_pProcessQueue->isDropped()) |
| { |
| RMQ_WARN("consumeMessage, the message queue not be able to consume, because it's dropped, MQ: %s", |
| m_messageQueue.toString().c_str()); |
| break; |
| } |
| |
| status = m_pService->getMessageListener()->consumeMessage(msgs, context); |
| } |
| catch (...) |
| { |
| RMQ_WARN("consumeMessage exception, Group: {%s}, Msgs: {%u}, MQ: %s",// |
| m_pService->getConsumerGroup().c_str(), |
| (unsigned)msgs.size(), |
| m_messageQueue.toString().c_str()); |
| } |
| |
| long long consumeRT = KPRUtil::GetCurrentTimeMillis() - beginTimestamp; |
| |
| if (SUSPEND_CURRENT_QUEUE_A_MOMENT == status |
| || ROLLBACK == status) |
| { |
| RMQ_WARN("consumeMessage Orderly return not OK, Group: {%s} Msgs: {%u} MQ: %s",// |
| m_pService->getConsumerGroup().c_str(), |
| (unsigned)msgs.size(), |
| m_messageQueue.toString().c_str()); |
| //status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; |
| } |
| |
| if (m_pService->getDefaultMQPushConsumerImpl()->hasHook()) |
| { |
| consumeMessageContext.success = (SUCCESS == status |
| || COMMIT == status); |
| m_pService->getDefaultMQPushConsumerImpl()->executeHookAfter(consumeMessageContext); |
| } |
| |
| m_pService->getConsumerStat().consumeMsgRTTotal.fetchAndAdd(consumeRT); |
| MixAll::compareAndIncreaseOnly(m_pService->getConsumerStat() |
| .consumeMsgRTMax, consumeRT); |
| |
| continueConsume = m_pService->processConsumeResult(msgs, status, context, *this); |
| } |
| else |
| { |
| continueConsume = false; |
| } |
| } |
| } |
| else |
| { |
| if (m_pProcessQueue->isDropped()) |
| { |
| RMQ_WARN("consumeMessage, the message queue not be able to consume, because it's dropped, MQ: %s", |
| m_messageQueue.toString().c_str()); |
| return; |
| } |
| |
| m_pService->tryLockLaterAndReconsume(m_messageQueue, m_pProcessQueue, 100); |
| } |
| } |
| } |
| catch(...) |
| { |
| RMQ_WARN("ConsumeOrderlyRequest exception"); |
| } |
| |
| return; |
| } |
| |
| } |
| |
