| /** |
| * 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 "DefaultMQPushConsumerImpl.h" |
| |
| #include <string> |
| #include <set> |
| #include "DefaultMQPushConsumer.h" |
| #include "ConsumerStatManage.h" |
| #include "DefaultMQPullConsumer.h" |
| #include "DefaultMQProducer.h" |
| #include "MQClientFactory.h" |
| #include "MQAdminImpl.h" |
| #include "RebalancePushImpl.h" |
| #include "MQClientAPIImpl.h" |
| #include "OffsetStore.h" |
| #include "MixAll.h" |
| #include "MQClientManager.h" |
| #include "LocalFileOffsetStore.h" |
| #include "RemoteBrokerOffsetStore.h" |
| #include "PullSysFlag.h" |
| #include "FilterAPI.h" |
| #include "PullAPIWrapper.h" |
| #include "MQClientException.h" |
| #include "Validators.h" |
| #include "MessageListener.h" |
| #include "ConsumeMessageHook.h" |
| #include "PullMessageService.h" |
| #include "ConsumeMessageOrderlyService.h" |
| #include "ConsumeMessageConcurrentlyService.h" |
| #include "KPRUtil.h" |
| #include "TimerThread.h" |
| |
| namespace rmq |
| { |
| |
| /* RemoveProcessQueueLater */ |
| class RemoveProcessQueueLater : public kpr::TimerHandler |
| { |
| public: |
| RemoveProcessQueueLater(DefaultMQPushConsumerImpl* pConsumerImp, PullRequest* pPullRequest) |
| : m_pConsumerImp(pConsumerImp), m_pPullRequest(pPullRequest) |
| { |
| } |
| |
| void OnTimeOut(unsigned int timerID) |
| { |
| try |
| { |
| m_pConsumerImp->getOffsetStore()->updateOffset(m_pPullRequest->getMessageQueue(), m_pPullRequest->getNextOffset(), false); |
| m_pConsumerImp->getOffsetStore()->persist(m_pPullRequest->getMessageQueue()); |
| m_pConsumerImp->getRebalanceImpl()->removeProcessQueue(m_pPullRequest->getMessageQueue()); |
| |
| RMQ_WARN("fix the pull request offset, {%s}", m_pPullRequest->toString().c_str()); |
| } |
| catch(...) |
| { |
| RMQ_ERROR("RemoveProcessQueueLater OnTimeOut Exception"); |
| } |
| |
| delete this; |
| } |
| |
| private: |
| DefaultMQPushConsumerImpl* m_pConsumerImp; |
| PullRequest* m_pPullRequest; |
| }; |
| |
| |
| /* DefaultMQPushConsumerImplCallback */ |
| class DefaultMQPushConsumerImplCallback : public PullCallback |
| { |
| public: |
| DefaultMQPushConsumerImplCallback(SubscriptionData& subscriptionData, |
| DefaultMQPushConsumerImpl* pDefaultMQPushConsumerImpl, |
| PullRequest* pPullRequest) |
| : m_subscriptionData(subscriptionData), |
| m_pDefaultMQPushConsumerImpl(pDefaultMQPushConsumerImpl), |
| m_pPullRequest(pPullRequest) |
| { |
| m_beginTimestamp = KPRUtil::GetCurrentTimeMillis(); |
| } |
| |
| void onSuccess(PullResult& pullResult) |
| { |
| RMQ_DEBUG("onSuccess begin: %s", pullResult.toString().c_str()); |
| PullResult* pPullResult = &pullResult; |
| if (pPullResult != NULL) |
| { |
| pPullResult = |
| m_pDefaultMQPushConsumerImpl->m_pPullAPIWrapper->processPullResult( |
| m_pPullRequest->getMessageQueue(), *pPullResult, m_subscriptionData); |
| |
| switch (pPullResult->pullStatus) |
| { |
| case FOUND: |
| { |
| m_pPullRequest->setNextOffset(pPullResult->nextBeginOffset); |
| |
| long long pullRT = KPRUtil::GetCurrentTimeMillis() - m_beginTimestamp; |
| m_pDefaultMQPushConsumerImpl->getConsumerStatManager()->getConsumertat() |
| .pullTimesTotal++; |
| m_pDefaultMQPushConsumerImpl->getConsumerStatManager()->getConsumertat() |
| .pullRTTotal.fetchAndAdd(pullRT); |
| |
| ProcessQueue* processQueue = m_pPullRequest->getProcessQueue(); |
| bool dispatchToConsume = processQueue->putMessage(pPullResult->msgFoundList); |
| |
| m_pDefaultMQPushConsumerImpl->m_pConsumeMessageService->submitConsumeRequest(// |
| pPullResult->msgFoundList, // |
| processQueue, // |
| m_pPullRequest->getMessageQueue(), // |
| dispatchToConsume); |
| |
| if (m_pDefaultMQPushConsumerImpl->m_pDefaultMQPushConsumer->getPullInterval() > 0) |
| { |
| m_pDefaultMQPushConsumerImpl->executePullRequestLater(m_pPullRequest, |
| m_pDefaultMQPushConsumerImpl->m_pDefaultMQPushConsumer->getPullInterval()); |
| } |
| else |
| { |
| m_pDefaultMQPushConsumerImpl->executePullRequestImmediately(m_pPullRequest); |
| } |
| } |
| break; |
| case NO_NEW_MSG: |
| m_pPullRequest->setNextOffset(pPullResult->nextBeginOffset); |
| m_pDefaultMQPushConsumerImpl->correctTagsOffset(*m_pPullRequest); |
| m_pDefaultMQPushConsumerImpl->executePullRequestImmediately(m_pPullRequest); |
| break; |
| case NO_MATCHED_MSG: |
| m_pPullRequest->setNextOffset(pPullResult->nextBeginOffset); |
| m_pDefaultMQPushConsumerImpl->correctTagsOffset(*m_pPullRequest); |
| m_pDefaultMQPushConsumerImpl->executePullRequestImmediately(m_pPullRequest); |
| break; |
| case OFFSET_ILLEGAL: |
| RMQ_WARN("the pull request offset illegal, %s, %s", |
| m_pPullRequest->toString().c_str(), pPullResult->toString().c_str()); |
| |
| /* |
| if (m_pPullRequest->getNextOffset() < pPullResult->minOffset) |
| { |
| m_pPullRequest->setNextOffset(pPullResult->minOffset); |
| } |
| else if (m_pPullRequest->getNextOffset() > pPullResult->maxOffset) |
| { |
| m_pPullRequest->setNextOffset(pPullResult->maxOffset); |
| } |
| m_pDefaultMQPushConsumerImpl->m_pOffsetStore->updateOffset( |
| m_pPullRequest->getMessageQueue(), m_pPullRequest->getNextOffset(), false); |
| m_pDefaultMQPushConsumerImpl->executePullRequestImmediately(m_pPullRequest); |
| */ |
| |
| // todo |
| m_pPullRequest->setNextOffset(pPullResult->nextBeginOffset); |
| m_pPullRequest->getProcessQueue()->setDropped(true); |
| |
| m_pDefaultMQPushConsumerImpl->executeTaskLater(new RemoveProcessQueueLater( |
| m_pDefaultMQPushConsumerImpl, m_pPullRequest), 10000); |
| break; |
| default: |
| break; |
| } |
| } |
| else |
| { |
| RMQ_WARN("Warning: PullRequest is null!"); |
| } |
| RMQ_DEBUG("onSuccess end"); |
| } |
| |
| void onException(MQException& e) |
| { |
| std::string topic = m_pPullRequest->getMessageQueue().getTopic(); |
| if (topic.find(MixAll::RETRY_GROUP_TOPIC_PREFIX) != std::string::npos) |
| { |
| RMQ_WARN("execute the pull request exception:%s", e.what()); |
| } |
| |
| m_pDefaultMQPushConsumerImpl->executePullRequestLater(m_pPullRequest, |
| DefaultMQPushConsumerImpl::s_PullTimeDelayMillsWhenException); |
| } |
| |
| private: |
| SubscriptionData m_subscriptionData; |
| DefaultMQPushConsumerImpl* m_pDefaultMQPushConsumerImpl; |
| PullRequest* m_pPullRequest; |
| unsigned long long m_beginTimestamp; |
| }; |
| |
| |
| DefaultMQPushConsumerImpl::DefaultMQPushConsumerImpl(DefaultMQPushConsumer* pDefaultMQPushConsumer) |
| { |
| m_pDefaultMQPushConsumer = pDefaultMQPushConsumer; |
| m_serviceState = CREATE_JUST; |
| flowControlTimes1 = 0; |
| flowControlTimes2 = 0; |
| m_pause = false; |
| m_consumeOrderly = false; |
| |
| m_pMQClientFactory = NULL; |
| m_pPullAPIWrapper = NULL; |
| m_pMessageListenerInner = NULL; |
| m_pOffsetStore = NULL; |
| m_pRebalanceImpl = new RebalancePushImpl(this); |
| m_pConsumerStatManager = new ConsumerStatManager(); |
| m_pConsumeMessageService = NULL; |
| } |
| |
| DefaultMQPushConsumerImpl::~DefaultMQPushConsumerImpl() |
| { |
| //delete m_pMessageListenerInner; |
| if (m_pPullAPIWrapper) |
| delete m_pPullAPIWrapper; |
| if (m_pRebalanceImpl) |
| delete m_pRebalanceImpl; |
| if (m_pConsumerStatManager) |
| delete m_pConsumerStatManager; |
| if (m_pConsumeMessageService) |
| delete m_pConsumeMessageService; |
| if (m_pOffsetStore) |
| delete m_pOffsetStore; |
| //delete m_pMQClientFactory; |
| } |
| |
| void DefaultMQPushConsumerImpl::start() |
| { |
| RMQ_DEBUG("DefaultMQPushConsumerImpl::start()"); |
| switch (m_serviceState) |
| { |
| case CREATE_JUST: |
| { |
| RMQ_INFO("the consumer [{%s}] start beginning. messageModel={%s}", |
| m_pDefaultMQPushConsumer->getConsumerGroup().c_str(), |
| getMessageModelString(m_pDefaultMQPushConsumer->getMessageModel())); |
| |
| m_serviceState = START_FAILED; |
| checkConfig(); |
| copySubscription(); |
| |
| if (m_pDefaultMQPushConsumer->getMessageModel() == CLUSTERING) |
| { |
| m_pDefaultMQPushConsumer->changeInstanceNameToPID(); |
| } |
| |
| m_pMQClientFactory = MQClientManager::getInstance()->getAndCreateMQClientFactory(*m_pDefaultMQPushConsumer); |
| |
| m_pRebalanceImpl->setConsumerGroup(m_pDefaultMQPushConsumer->getConsumerGroup()); |
| m_pRebalanceImpl->setMessageModel(m_pDefaultMQPushConsumer->getMessageModel()); |
| m_pRebalanceImpl->setAllocateMessageQueueStrategy(m_pDefaultMQPushConsumer->getAllocateMessageQueueStrategy()); |
| m_pRebalanceImpl->setmQClientFactory(m_pMQClientFactory); |
| |
| m_pPullAPIWrapper = new PullAPIWrapper(m_pMQClientFactory, m_pDefaultMQPushConsumer->getConsumerGroup()); |
| |
| if (m_pDefaultMQPushConsumer->getOffsetStore() != NULL) |
| { |
| m_pOffsetStore = m_pDefaultMQPushConsumer->getOffsetStore(); |
| } |
| else |
| { |
| switch (m_pDefaultMQPushConsumer->getMessageModel()) |
| { |
| case BROADCASTING: |
| m_pOffsetStore = new LocalFileOffsetStore(m_pMQClientFactory, m_pDefaultMQPushConsumer->getConsumerGroup()); |
| break; |
| case CLUSTERING: |
| m_pOffsetStore = new RemoteBrokerOffsetStore(m_pMQClientFactory, m_pDefaultMQPushConsumer->getConsumerGroup()); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| m_pOffsetStore->load(); |
| |
| if (dynamic_cast<MessageListenerOrderly*>(m_pMessageListenerInner) != NULL) |
| { |
| m_consumeOrderly = true; |
| m_pConsumeMessageService = |
| new ConsumeMessageOrderlyService(this, (MessageListenerOrderly*)m_pMessageListenerInner); |
| } |
| else if (dynamic_cast<MessageListenerConcurrently*>(m_pMessageListenerInner) != NULL) |
| { |
| m_consumeOrderly = false; |
| m_pConsumeMessageService = |
| new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently*)m_pMessageListenerInner); |
| } |
| m_pConsumeMessageService->start(); |
| |
| bool registerOK = m_pMQClientFactory->registerConsumer(m_pDefaultMQPushConsumer->getConsumerGroup(), this); |
| if (!registerOK) |
| { |
| m_serviceState = CREATE_JUST; |
| m_pConsumeMessageService->shutdown(); |
| std::string str = "The consumer group[" + m_pDefaultMQPushConsumer->getConsumerGroup(); |
| str += "] has been created before, specify another name please."; |
| THROW_MQEXCEPTION(MQClientException, str, -1); |
| } |
| m_pMQClientFactory->start(); |
| |
| RMQ_INFO("the consumer [%s] start OK.", m_pDefaultMQPushConsumer->getConsumerGroup().c_str()); |
| m_serviceState = RUNNING; |
| } |
| break; |
| case RUNNING: |
| case START_FAILED: |
| case SHUTDOWN_ALREADY: |
| THROW_MQEXCEPTION(MQClientException, "The PullConsumer service state not OK, maybe started once, ", -1); |
| default: |
| break; |
| } |
| |
| updateTopicSubscribeInfoWhenSubscriptionChanged(); |
| m_pMQClientFactory->sendHeartbeatToAllBrokerWithLock(); |
| m_pMQClientFactory->rebalanceImmediately(); |
| } |
| |
| |
| void DefaultMQPushConsumerImpl::shutdown() |
| { |
| RMQ_DEBUG("DefaultMQPushConsumerImpl::shutdown()"); |
| switch (m_serviceState) |
| { |
| case CREATE_JUST: |
| break; |
| case RUNNING: |
| m_pConsumeMessageService->shutdown(); |
| persistConsumerOffset(); |
| m_pMQClientFactory->unregisterConsumer(m_pDefaultMQPushConsumer->getConsumerGroup()); |
| m_pMQClientFactory->shutdown(); |
| |
| m_serviceState = SHUTDOWN_ALREADY; |
| break; |
| case SHUTDOWN_ALREADY: |
| break; |
| default: |
| break; |
| } |
| } |
| |
| |
| |
| |
| bool DefaultMQPushConsumerImpl::hasHook() |
| { |
| return !m_hookList.empty(); |
| } |
| |
| void DefaultMQPushConsumerImpl::registerHook(ConsumeMessageHook* pHook) |
| { |
| m_hookList.push_back(pHook); |
| } |
| |
| void DefaultMQPushConsumerImpl::executeHookBefore(ConsumeMessageContext& context) |
| { |
| std::list<ConsumeMessageHook*>::iterator it = m_hookList.begin(); |
| for (; it != m_hookList.end(); it++) |
| { |
| try |
| { |
| (*it)->consumeMessageBefore(context); |
| } |
| catch (...) |
| { |
| RMQ_WARN("consumeMessageBefore exception"); |
| } |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::executeHookAfter(ConsumeMessageContext& context) |
| { |
| std::list<ConsumeMessageHook*>::iterator it = m_hookList.begin(); |
| for (; it != m_hookList.end(); it++) |
| { |
| try |
| { |
| (*it)->consumeMessageAfter(context); |
| } |
| catch (...) |
| { |
| RMQ_WARN("consumeMessageAfter exception"); |
| } |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::createTopic(const std::string& key, const std::string& newTopic, int queueNum) |
| { |
| m_pMQClientFactory->getMQAdminImpl()->createTopic(key, newTopic, queueNum); |
| } |
| |
| std::set<MessageQueue>* DefaultMQPushConsumerImpl::fetchSubscribeMessageQueues(const std::string& topic) |
| { |
| std::map<std::string, std::set<MessageQueue> >& mqs = m_pRebalanceImpl->getTopicSubscribeInfoTable(); |
| std::map<std::string, std::set<MessageQueue> >::iterator it = mqs.find(topic); |
| |
| if (it == mqs.end()) |
| { |
| m_pMQClientFactory->updateTopicRouteInfoFromNameServer(topic); |
| mqs = m_pRebalanceImpl->getTopicSubscribeInfoTable(); |
| it = mqs.find(topic); |
| } |
| |
| if (it == mqs.end()) |
| { |
| THROW_MQEXCEPTION(MQClientException, "The topic[" + topic + "] not exist", -1); |
| } |
| |
| std::set<MessageQueue>* result = new std::set<MessageQueue>(it->second.begin(), it->second.end()); |
| return result; |
| } |
| |
| DefaultMQPushConsumer* DefaultMQPushConsumerImpl::getDefaultMQPushConsumer() |
| { |
| return m_pDefaultMQPushConsumer; |
| } |
| |
| long long DefaultMQPushConsumerImpl::earliestMsgStoreTime(const MessageQueue& mq) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->earliestMsgStoreTime(mq); |
| } |
| |
| long long DefaultMQPushConsumerImpl::maxOffset(const MessageQueue& mq) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->maxOffset(mq); |
| } |
| |
| long long DefaultMQPushConsumerImpl::minOffset(const MessageQueue& mq) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->minOffset(mq); |
| } |
| |
| OffsetStore* DefaultMQPushConsumerImpl::getOffsetStore() |
| { |
| return m_pOffsetStore; |
| } |
| |
| void DefaultMQPushConsumerImpl::setOffsetStore(OffsetStore* pOffsetStore) |
| { |
| m_pOffsetStore = pOffsetStore; |
| } |
| |
| //MQConsumerInner |
| std::string DefaultMQPushConsumerImpl::groupName() |
| { |
| return m_pDefaultMQPushConsumer->getConsumerGroup(); |
| } |
| |
| MessageModel DefaultMQPushConsumerImpl::messageModel() |
| { |
| return m_pDefaultMQPushConsumer->getMessageModel(); |
| } |
| |
| ConsumeType DefaultMQPushConsumerImpl::consumeType() |
| { |
| return CONSUME_PASSIVELY; |
| } |
| |
| ConsumeFromWhere DefaultMQPushConsumerImpl::consumeFromWhere() |
| { |
| return m_pDefaultMQPushConsumer->getConsumeFromWhere(); |
| } |
| |
| std::set<SubscriptionData> DefaultMQPushConsumerImpl::subscriptions() |
| { |
| std::set<SubscriptionData> sds; |
| std::map<std::string, SubscriptionData>& subscription = m_pRebalanceImpl->getSubscriptionInner(); |
| std::map<std::string, SubscriptionData>::iterator it = subscription.begin(); |
| for (; it != subscription.end(); it++) |
| { |
| sds.insert(it->second); |
| } |
| |
| return sds; |
| } |
| |
| void DefaultMQPushConsumerImpl::doRebalance() |
| { |
| if (m_pRebalanceImpl != NULL) |
| { |
| m_pRebalanceImpl->doRebalance(); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::persistConsumerOffset() |
| { |
| try |
| { |
| makeSureStateOK(); |
| |
| std::set<MessageQueue> mqs; |
| { |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_pRebalanceImpl->getProcessQueueTableLock()); |
| std::map<MessageQueue, ProcessQueue*>& processQueueTable = m_pRebalanceImpl->getProcessQueueTable(); |
| RMQ_FOR_EACH(processQueueTable, it) |
| { |
| mqs.insert(it->first); |
| } |
| } |
| |
| m_pOffsetStore->persistAll(mqs); |
| } |
| catch (...) |
| { |
| RMQ_ERROR("persistConsumerOffset exception, group: %s", |
| m_pDefaultMQPushConsumer->getConsumerGroup().c_str()); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::updateTopicSubscribeInfo(const std::string& topic, const std::set<MessageQueue>& info) |
| { |
| std::map<std::string, SubscriptionData>& subTable = getSubscriptionInner(); |
| |
| if (subTable.find(topic) != subTable.end()) |
| { |
| m_pRebalanceImpl->getTopicSubscribeInfoTable().insert(std::pair<std::string, std::set<MessageQueue> >(topic, info)); |
| } |
| } |
| |
| std::map<std::string, SubscriptionData>& DefaultMQPushConsumerImpl::getSubscriptionInner() |
| { |
| return m_pRebalanceImpl->getSubscriptionInner(); |
| } |
| |
| bool DefaultMQPushConsumerImpl::isSubscribeTopicNeedUpdate(const std::string& topic) |
| { |
| std::map<std::string, SubscriptionData>& subTable = getSubscriptionInner(); |
| |
| if (subTable.find(topic) != subTable.end()) |
| { |
| std::map<std::string, std::set<MessageQueue> >& mqs = |
| m_pRebalanceImpl->getTopicSubscribeInfoTable(); |
| |
| return mqs.find(topic) == mqs.end(); |
| } |
| |
| return false; |
| } |
| |
| bool DefaultMQPushConsumerImpl::isPause() |
| { |
| return m_pause; |
| } |
| |
| void DefaultMQPushConsumerImpl::setPause(bool pause) |
| { |
| m_pause = pause; |
| } |
| |
| |
| void DefaultMQPushConsumerImpl::correctTagsOffset(PullRequest& pullRequest) |
| { |
| if (pullRequest.getProcessQueue()->getMsgCount().get() == 0) |
| { |
| m_pOffsetStore->updateOffset(pullRequest.getMessageQueue(), pullRequest.getNextOffset(), true); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::pullMessage(PullRequest* pPullRequest) |
| { |
| RMQ_DEBUG("pullMessage begin: %s", pPullRequest->toString().c_str()); |
| |
| ProcessQueue* processQueue = pPullRequest->getProcessQueue(); |
| if (processQueue->isDropped()) |
| { |
| RMQ_WARN("the pull request[%s] is dropped.", pPullRequest->toString().c_str()); |
| delete pPullRequest; |
| return; |
| } |
| |
| pPullRequest->getProcessQueue()->setLastPullTimestamp(KPRUtil::GetCurrentTimeMillis()); |
| |
| try |
| { |
| makeSureStateOK(); |
| } |
| catch (const MQException& e) |
| { |
| RMQ_WARN("pullMessage exception [%s], consumer state not ok", e.what()); |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenException); |
| return; |
| } |
| |
| if (isPause()) |
| { |
| RMQ_WARN("consumer was paused, execute pull request later. instanceName={%s}", |
| m_pDefaultMQPushConsumer->getInstanceName().c_str()); |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenSuspend); |
| return; |
| } |
| |
| long size = processQueue->getMsgCount().get(); |
| if (size > m_pDefaultMQPushConsumer->getPullThresholdForQueue()) |
| { |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenFlowControl); |
| if ((flowControlTimes1++ % 3000) == 0) |
| { |
| RMQ_WARN("the consumer message buffer is full, so do flow control, {%ld} {%s} {%lld}", size, |
| pPullRequest->toString().c_str(), flowControlTimes1); |
| } |
| return; |
| } |
| |
| if (!m_consumeOrderly) |
| { |
| if (processQueue->getMaxSpan() > m_pDefaultMQPushConsumer->getConsumeConcurrentlyMaxSpan()) |
| { |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenFlowControl); |
| if ((flowControlTimes2++ % 3000) == 0) |
| { |
| RMQ_WARN("the queue's messages, span too long, so do flow control, size: {%ld}, pullRequest: {%s}, times: {%lld}, maxspan: {%lld}", |
| size, pPullRequest->toString().c_str(), flowControlTimes2, processQueue->getMaxSpan()); |
| } |
| return; |
| } |
| } |
| |
| std::map<std::string, SubscriptionData>& subTable = getSubscriptionInner(); |
| std::string topic = pPullRequest->getMessageQueue().getTopic(); |
| std::map<std::string, SubscriptionData>::iterator it = subTable.find(topic); |
| if (it == subTable.end()) |
| { |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenException); |
| RMQ_WARN("find the consumer's subscription failed, {%s}", pPullRequest->toString().c_str()); |
| return; |
| } |
| |
| SubscriptionData subscriptionData = it->second; |
| PullCallback* pullCallback = new DefaultMQPushConsumerImplCallback(subTable[topic], this, pPullRequest); |
| |
| bool commitOffsetEnable = false; |
| long commitOffsetValue = 0L; |
| if (CLUSTERING == m_pDefaultMQPushConsumer->getMessageModel()) |
| { |
| commitOffsetValue = m_pOffsetStore->readOffset(pPullRequest->getMessageQueue(), |
| READ_FROM_MEMORY); |
| if (commitOffsetValue > 0) |
| { |
| commitOffsetEnable = true; |
| } |
| } |
| |
| int sysFlag = PullSysFlag::buildSysFlag( |
| commitOffsetEnable, // commitOffset |
| true, // suspend |
| false// subscription |
| ); |
| try |
| { |
| m_pPullAPIWrapper->pullKernelImpl( |
| pPullRequest->getMessageQueue(), // 1 |
| "", // 2 |
| subscriptionData.getSubVersion(), // 3 |
| pPullRequest->getNextOffset(), // 4 |
| m_pDefaultMQPushConsumer->getPullBatchSize(), // 5 |
| sysFlag, // 6 |
| commitOffsetValue,// 7 |
| s_BrokerSuspendMaxTimeMillis, // 8 |
| s_ConsumerTimeoutMillisWhenSuspend, // 9 |
| ASYNC, // 10 |
| pullCallback// 11 |
| ); |
| } |
| catch (...) |
| { |
| RMQ_ERROR("pullKernelImpl exception"); |
| executePullRequestLater(pPullRequest, s_PullTimeDelayMillsWhenException); |
| } |
| |
| RMQ_DEBUG("pullMessage end"); |
| } |
| |
| void DefaultMQPushConsumerImpl::executePullRequestImmediately(PullRequest* pullRequest) |
| { |
| m_pMQClientFactory->getPullMessageService()->executePullRequestImmediately(pullRequest); |
| } |
| |
| void DefaultMQPushConsumerImpl::executePullRequestLater(PullRequest* pullRequest, long timeDelay) |
| { |
| m_pMQClientFactory->getPullMessageService()->executePullRequestLater(pullRequest, timeDelay); |
| } |
| |
| void DefaultMQPushConsumerImpl::executeTaskLater(kpr::TimerHandler* handler, long timeDelay) |
| { |
| m_pMQClientFactory->getPullMessageService()->executeTaskLater(handler, timeDelay); |
| } |
| |
| |
| void DefaultMQPushConsumerImpl::makeSureStateOK() |
| { |
| if (m_serviceState != RUNNING) |
| { |
| THROW_MQEXCEPTION(MQClientException, "The consumer service state not OK, ", -1); |
| } |
| } |
| |
| ConsumerStatManager* DefaultMQPushConsumerImpl::getConsumerStatManager() |
| { |
| return m_pConsumerStatManager; |
| } |
| |
| QueryResult DefaultMQPushConsumerImpl::queryMessage(const std::string& topic, |
| const std::string& key, |
| int maxNum, |
| long long begin, |
| long long end) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->queryMessage(topic, key, maxNum, begin, end); |
| } |
| |
| void DefaultMQPushConsumerImpl::registerMessageListener(MessageListener* pMessageListener) |
| { |
| m_pMessageListenerInner = pMessageListener; |
| } |
| |
| void DefaultMQPushConsumerImpl::resume() |
| { |
| m_pause = false; |
| } |
| |
| long long DefaultMQPushConsumerImpl::searchOffset(const MessageQueue& mq, long long timestamp) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->searchOffset(mq, timestamp); |
| } |
| |
| void DefaultMQPushConsumerImpl::sendMessageBack(MessageExt& msg, int delayLevel, const std::string& brokerName) |
| { |
| try |
| { |
| std::string brokerAddr = brokerName.empty() ? |
| socketAddress2IPPort(msg.getStoreHost()) : m_pMQClientFactory->findBrokerAddressInPublish(brokerName); |
| |
| m_pMQClientFactory->getMQClientAPIImpl()->consumerSendMessageBack(brokerAddr, msg, |
| m_pDefaultMQPushConsumer->getConsumerGroup(), |
| delayLevel, |
| 5000); |
| } |
| catch (...) |
| { |
| RMQ_ERROR("sendMessageBack Exception, group: %s", m_pDefaultMQPushConsumer->getConsumerGroup().c_str()); |
| 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_pMQClientFactory->getDefaultMQProducer()->send(newMsg); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::checkConfig() |
| { |
| // consumerGroup check |
| Validators::checkGroup(m_pDefaultMQPushConsumer->getConsumerGroup()); |
| |
| // consumerGroup |
| if (m_pDefaultMQPushConsumer->getConsumerGroup() == MixAll::DEFAULT_CONSUMER_GROUP) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumerGroup can not equal " |
| + MixAll::DEFAULT_CONSUMER_GROUP // |
| + ", please specify another one.", -1); |
| } |
| |
| if (m_pDefaultMQPushConsumer->getMessageModel() != BROADCASTING |
| && m_pDefaultMQPushConsumer->getMessageModel() != CLUSTERING) |
| { |
| THROW_MQEXCEPTION(MQClientException, "messageModel is invalid ", -1); |
| } |
| |
| // allocateMessageQueueStrategy |
| if (m_pDefaultMQPushConsumer->getAllocateMessageQueueStrategy() == NULL) |
| { |
| THROW_MQEXCEPTION(MQClientException, "allocateMessageQueueStrategy is null", -1); |
| } |
| |
| // consumeFromWhereOffset |
| if (m_pDefaultMQPushConsumer->getConsumeFromWhere() < CONSUME_FROM_LAST_OFFSET |
| || m_pDefaultMQPushConsumer->getConsumeFromWhere() > CONSUME_FROM_MAX_OFFSET) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumeFromWhere is invalid", -1); |
| } |
| |
| // subscription |
| /* |
| if (m_pDefaultMQPushConsumer->getSubscription().size() == 0) |
| { |
| THROW_MQEXCEPTION(MQClientException,"subscription is null" ,-1); |
| } |
| */ |
| |
| // messageListener |
| if (m_pDefaultMQPushConsumer->getMessageListener() == NULL) |
| { |
| THROW_MQEXCEPTION(MQClientException, "messageListener is null", -1); |
| } |
| |
| MessageListener* listener = m_pDefaultMQPushConsumer->getMessageListener(); |
| MessageListener* orderly = (dynamic_cast<MessageListenerOrderly*>(listener)) ; |
| MessageListener* concurrently = (dynamic_cast<MessageListenerConcurrently*>(listener)) ; |
| |
| if (!orderly && !concurrently) |
| { |
| THROW_MQEXCEPTION(MQClientException, |
| "messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" , |
| -1); |
| } |
| |
| // consumeThreadMin |
| if (m_pDefaultMQPushConsumer->getConsumeThreadMin() < 1 |
| || m_pDefaultMQPushConsumer->getConsumeThreadMin() > 1000 |
| || m_pDefaultMQPushConsumer->getConsumeThreadMin() > m_pDefaultMQPushConsumer->getConsumeThreadMax() |
| ) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumeThreadMin Out of range [1, 1000]", -1); |
| } |
| |
| // consumeThreadMax |
| if (m_pDefaultMQPushConsumer->getConsumeThreadMax() < 1 |
| || m_pDefaultMQPushConsumer->getConsumeThreadMax() > 1000) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumeThreadMax Out of range [1, 1000]", -1); |
| } |
| |
| // consumeConcurrentlyMaxSpan |
| if (m_pDefaultMQPushConsumer->getConsumeConcurrentlyMaxSpan() < 1 |
| || m_pDefaultMQPushConsumer->getConsumeConcurrentlyMaxSpan() > 65535) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumeConcurrentlyMaxSpan Out of range [1, 65535]" , -1); |
| } |
| |
| // pullThresholdForQueue |
| if (m_pDefaultMQPushConsumer->getPullThresholdForQueue() < 1 |
| || m_pDefaultMQPushConsumer->getPullThresholdForQueue() > 65535) |
| { |
| THROW_MQEXCEPTION(MQClientException, "pullThresholdForQueue Out of range [1, 65535]", -1); |
| } |
| |
| // pullInterval |
| if (m_pDefaultMQPushConsumer->getPullInterval() < 0 |
| || m_pDefaultMQPushConsumer->getPullInterval() > 65535) |
| { |
| THROW_MQEXCEPTION(MQClientException, "pullInterval Out of range [0, 65535]", -1); |
| } |
| |
| // consumeMessageBatchMaxSize |
| if (m_pDefaultMQPushConsumer->getConsumeMessageBatchMaxSize() < 1 |
| || m_pDefaultMQPushConsumer->getConsumeMessageBatchMaxSize() > 1024) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumeMessageBatchMaxSize Out of range [1, 1024]", -1); |
| } |
| |
| // pullBatchSize |
| if (m_pDefaultMQPushConsumer->getPullBatchSize() < 1 |
| || m_pDefaultMQPushConsumer->getPullBatchSize() > 1024) |
| { |
| THROW_MQEXCEPTION(MQClientException, "pullBatchSize Out of range [1, 1024]", -1); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::copySubscription() |
| { |
| try |
| { |
| std::map<std::string, std::string>& sub = m_pDefaultMQPushConsumer->getSubscription(); |
| std::map<std::string, std::string>::iterator it = sub.begin(); |
| for (; it != sub.end(); it++) |
| { |
| SubscriptionDataPtr subscriptionData = FilterAPI::buildSubscriptionData(it->first, it->second); |
| m_pRebalanceImpl->getSubscriptionInner()[it->first] = *subscriptionData; |
| } |
| |
| if (m_pMessageListenerInner == NULL) |
| { |
| m_pMessageListenerInner = m_pDefaultMQPushConsumer->getMessageListener(); |
| } |
| |
| switch (m_pDefaultMQPushConsumer->getMessageModel()) |
| { |
| case BROADCASTING: |
| break; |
| case CLUSTERING: |
| { |
| std::string retryTopic = MixAll::getRetryTopic(m_pDefaultMQPushConsumer->getConsumerGroup()); |
| SubscriptionDataPtr subscriptionData = |
| FilterAPI::buildSubscriptionData(retryTopic, SubscriptionData::SUB_ALL); |
| m_pRebalanceImpl->getSubscriptionInner()[retryTopic] = *subscriptionData; |
| } |
| |
| break; |
| default: |
| break; |
| } |
| } |
| catch (...) |
| { |
| THROW_MQEXCEPTION(MQClientException, "subscription exception", -1); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::updateTopicSubscribeInfoWhenSubscriptionChanged() |
| { |
| std::map<std::string, SubscriptionData> subTable = getSubscriptionInner(); |
| std::map<std::string, SubscriptionData>::iterator it = subTable.begin(); |
| for (; it != subTable.end(); it++) |
| { |
| m_pMQClientFactory->updateTopicRouteInfoFromNameServer(it->first); |
| } |
| } |
| |
| MessageListener* DefaultMQPushConsumerImpl::getMessageListenerInner() |
| { |
| return m_pMessageListenerInner; |
| } |
| |
| void DefaultMQPushConsumerImpl::subscribe(const std::string& topic, const std::string& subExpression) |
| { |
| try |
| { |
| SubscriptionDataPtr subscriptionData = FilterAPI::buildSubscriptionData(topic, subExpression); |
| m_pRebalanceImpl->getSubscriptionInner()[topic] = *subscriptionData; |
| |
| if (m_pMQClientFactory) |
| { |
| m_pMQClientFactory->sendHeartbeatToAllBrokerWithLock(); |
| } |
| } |
| catch (...) |
| { |
| THROW_MQEXCEPTION(MQClientException, "subscription exception", -1); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::suspend() |
| { |
| m_pause = true; |
| } |
| |
| void DefaultMQPushConsumerImpl::unsubscribe(const std::string& topic) |
| { |
| m_pRebalanceImpl->getSubscriptionInner().erase(topic); |
| } |
| |
| void DefaultMQPushConsumerImpl::updateConsumeOffset(MessageQueue& mq, long long offset) |
| { |
| m_pOffsetStore->updateOffset(mq, offset, false); |
| } |
| |
| void DefaultMQPushConsumerImpl::updateCorePoolSize(int corePoolSize) |
| { |
| m_pConsumeMessageService->updateCorePoolSize(corePoolSize); |
| } |
| |
| MessageExt* DefaultMQPushConsumerImpl::viewMessage(const std::string& msgId) |
| { |
| return m_pMQClientFactory->getMQAdminImpl()->viewMessage(msgId); |
| } |
| |
| RebalanceImpl* DefaultMQPushConsumerImpl::getRebalanceImpl() |
| { |
| return m_pRebalanceImpl; |
| } |
| |
| bool DefaultMQPushConsumerImpl::isConsumeOrderly() |
| { |
| return m_consumeOrderly; |
| } |
| |
| void DefaultMQPushConsumerImpl::setConsumeOrderly(bool consumeOrderly) |
| { |
| m_consumeOrderly = consumeOrderly; |
| } |
| |
| |
| MQClientFactory* DefaultMQPushConsumerImpl::getmQClientFactory() |
| { |
| return m_pMQClientFactory; |
| } |
| |
| void DefaultMQPushConsumerImpl::setmQClientFactory(MQClientFactory* mQClientFactory) |
| { |
| m_pMQClientFactory = mQClientFactory; |
| } |
| |
| |
| ServiceState DefaultMQPushConsumerImpl::getServiceState() |
| { |
| return m_serviceState; |
| } |
| |
| void DefaultMQPushConsumerImpl::setServiceState(ServiceState serviceState) |
| { |
| m_serviceState = serviceState; |
| } |
| |
| } |
| |
