| /** |
| * 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 "DefaultMQPullConsumerImpl.h" |
| |
| #include <iostream> |
| #include <string> |
| #include <set> |
| #include "DefaultMQPullConsumer.h" |
| #include "DefaultMQProducer.h" |
| #include "MQClientFactory.h" |
| #include "MQAdminImpl.h" |
| #include "RebalancePullImpl.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 "ScopedLock.h" |
| |
| namespace rmq |
| { |
| |
| DefaultMQPullConsumerImpl::DefaultMQPullConsumerImpl(DefaultMQPullConsumer* pDefaultMQPullConsumer) |
| : m_pDefaultMQPullConsumer(pDefaultMQPullConsumer), |
| m_serviceState(CREATE_JUST) |
| { |
| m_pMQClientFactory = NULL; |
| m_pPullAPIWrapper = NULL; |
| m_pOffsetStore = NULL; |
| m_pRebalanceImpl = new RebalancePullImpl(this); |
| } |
| |
| DefaultMQPullConsumerImpl::~DefaultMQPullConsumerImpl() |
| { |
| if (m_pRebalanceImpl) |
| delete m_pRebalanceImpl; |
| if (m_pPullAPIWrapper) |
| delete m_pPullAPIWrapper; |
| if (m_pOffsetStore) |
| delete m_pOffsetStore; |
| //delete m_pMQClientFactory; |
| } |
| |
| void DefaultMQPullConsumerImpl::start() |
| { |
| RMQ_INFO("DefaultMQPullConsumerImpl::start()"); |
| switch (m_serviceState) |
| { |
| case CREATE_JUST: |
| { |
| RMQ_INFO("the consumer [{%s}] start beginning. messageModel={%s}", |
| m_pDefaultMQPullConsumer->getConsumerGroup().c_str(), |
| getMessageModelString(m_pDefaultMQPullConsumer->getMessageModel())); |
| |
| m_serviceState = START_FAILED; |
| checkConfig(); |
| copySubscription(); |
| |
| if (m_pDefaultMQPullConsumer->getMessageModel() == CLUSTERING) |
| { |
| m_pDefaultMQPullConsumer->changeInstanceNameToPID(); |
| } |
| |
| m_pMQClientFactory = MQClientManager::getInstance()->getAndCreateMQClientFactory(*m_pDefaultMQPullConsumer); |
| |
| m_pRebalanceImpl->setConsumerGroup(m_pDefaultMQPullConsumer->getConsumerGroup()); |
| m_pRebalanceImpl->setMessageModel(m_pDefaultMQPullConsumer->getMessageModel()); |
| m_pRebalanceImpl->setAllocateMessageQueueStrategy(m_pDefaultMQPullConsumer->getAllocateMessageQueueStrategy()); |
| m_pRebalanceImpl->setmQClientFactory(m_pMQClientFactory); |
| |
| m_pPullAPIWrapper = new PullAPIWrapper(m_pMQClientFactory, m_pDefaultMQPullConsumer->getConsumerGroup()); |
| |
| if (m_pDefaultMQPullConsumer->getOffsetStore() != NULL) |
| { |
| m_pOffsetStore = m_pDefaultMQPullConsumer->getOffsetStore(); |
| } |
| else |
| { |
| switch (m_pDefaultMQPullConsumer->getMessageModel()) |
| { |
| case BROADCASTING: |
| m_pOffsetStore = new LocalFileOffsetStore(m_pMQClientFactory, m_pDefaultMQPullConsumer->getConsumerGroup()); |
| break; |
| case CLUSTERING: |
| m_pOffsetStore = new RemoteBrokerOffsetStore(m_pMQClientFactory, m_pDefaultMQPullConsumer->getConsumerGroup()); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| m_pOffsetStore->load(); |
| |
| bool registerOK = |
| m_pMQClientFactory->registerConsumer(m_pDefaultMQPullConsumer->getConsumerGroup(), this); |
| if (!registerOK) |
| { |
| m_serviceState = CREATE_JUST; |
| std::string str = "The consumer group[" + m_pDefaultMQPullConsumer->getConsumerGroup(); |
| str += "] has been created before, specify another name please."; |
| THROW_MQEXCEPTION(MQClientException, str, -1); |
| } |
| |
| m_pMQClientFactory->start(); |
| |
| 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; |
| } |
| } |
| |
| |
| void DefaultMQPullConsumerImpl::shutdown() |
| { |
| RMQ_DEBUG("DefaultMQPullConsumerImpl::shutdown()"); |
| switch (m_serviceState) |
| { |
| case CREATE_JUST: |
| break; |
| case RUNNING: |
| persistConsumerOffset(); |
| m_pMQClientFactory->unregisterConsumer(m_pDefaultMQPullConsumer->getConsumerGroup()); |
| m_pMQClientFactory->shutdown(); |
| |
| m_serviceState = SHUTDOWN_ALREADY; |
| break; |
| case SHUTDOWN_ALREADY: |
| break; |
| default: |
| break; |
| } |
| } |
| |
| |
| void DefaultMQPullConsumerImpl::createTopic(const std::string& key, const std::string& newTopic, int queueNum) |
| { |
| makeSureStateOK(); |
| m_pMQClientFactory->getMQAdminImpl()->createTopic(key, newTopic, queueNum); |
| } |
| |
| long long DefaultMQPullConsumerImpl::fetchConsumeOffset(MessageQueue& mq, bool fromStore) |
| { |
| makeSureStateOK(); |
| return m_pOffsetStore->readOffset(mq, fromStore ? READ_FROM_STORE : MEMORY_FIRST_THEN_STORE); |
| } |
| |
| std::set<MessageQueue>* DefaultMQPullConsumerImpl::fetchMessageQueuesInBalance(const std::string& topic) |
| { |
| makeSureStateOK(); |
| std::set<MessageQueue>* mqResult = new std::set<MessageQueue>; |
| |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_pRebalanceImpl->getProcessQueueTableLock()); |
| std::map<MessageQueue, ProcessQueue*>& mqTable = m_pRebalanceImpl->getProcessQueueTable(); |
| RMQ_FOR_EACH(mqTable, it) |
| { |
| if (it->first.getTopic() == topic) |
| { |
| mqResult->insert(it->first); |
| } |
| } |
| |
| return mqResult; |
| } |
| |
| std::vector<MessageQueue>* DefaultMQPullConsumerImpl::fetchPublishMessageQueues(const std::string& topic) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->fetchPublishMessageQueues(topic); |
| } |
| |
| std::set<MessageQueue>* DefaultMQPullConsumerImpl::fetchSubscribeMessageQueues(const std::string& topic) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->fetchSubscribeMessageQueues(topic); |
| } |
| |
| long long DefaultMQPullConsumerImpl::earliestMsgStoreTime(const MessageQueue& mq) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->earliestMsgStoreTime(mq); |
| } |
| |
| std::string DefaultMQPullConsumerImpl::groupName() |
| { |
| return m_pDefaultMQPullConsumer->getConsumerGroup(); |
| } |
| |
| MessageModel DefaultMQPullConsumerImpl::messageModel() |
| { |
| return m_pDefaultMQPullConsumer->getMessageModel(); |
| } |
| |
| ConsumeType DefaultMQPullConsumerImpl::consumeType() |
| { |
| return CONSUME_ACTIVELY; |
| } |
| |
| ConsumeFromWhere DefaultMQPullConsumerImpl::consumeFromWhere() |
| { |
| return CONSUME_FROM_LAST_OFFSET; |
| } |
| |
| std::set<SubscriptionData> DefaultMQPullConsumerImpl::subscriptions() |
| { |
| //TODO |
| std::set<SubscriptionData> result; |
| return result; |
| } |
| |
| void DefaultMQPullConsumerImpl::doRebalance() |
| { |
| if (m_pRebalanceImpl != NULL) |
| { |
| m_pRebalanceImpl->doRebalance(); |
| } |
| } |
| |
| void DefaultMQPullConsumerImpl::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("group {%s} persistConsumerOffset exception", |
| m_pDefaultMQPullConsumer->getConsumerGroup().c_str()); |
| } |
| } |
| |
| void DefaultMQPullConsumerImpl::updateTopicSubscribeInfo(const std::string& topic, const std::set<MessageQueue>& info) |
| { |
| std::map<std::string, SubscriptionData>& subTable = m_pRebalanceImpl->getSubscriptionInner(); |
| |
| if (subTable.find(topic) != subTable.end()) |
| { |
| m_pRebalanceImpl->getTopicSubscribeInfoTable().insert(std::pair<std::string, std::set<MessageQueue> >(topic, info)); |
| } |
| } |
| |
| bool DefaultMQPullConsumerImpl::isSubscribeTopicNeedUpdate(const std::string& topic) |
| { |
| std::map<std::string, SubscriptionData>& subTable = m_pRebalanceImpl->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; |
| } |
| |
| long long DefaultMQPullConsumerImpl::maxOffset(const MessageQueue& mq) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->maxOffset(mq); |
| } |
| |
| long long DefaultMQPullConsumerImpl::minOffset(const MessageQueue& mq) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->minOffset(mq); |
| } |
| |
| PullResult* DefaultMQPullConsumerImpl::pull(MessageQueue& mq, |
| const std::string& subExpression, |
| long long offset, |
| int maxNums) |
| { |
| return pullSyncImpl(mq, subExpression, offset, maxNums, false); |
| } |
| |
| void DefaultMQPullConsumerImpl::pull(MessageQueue& mq, |
| const std::string& subExpression, |
| long long offset, |
| int maxNums, |
| PullCallback* pPullCallback) |
| { |
| pullAsyncImpl(mq, subExpression, offset, maxNums, pPullCallback, false); |
| } |
| |
| PullResult* DefaultMQPullConsumerImpl::pullBlockIfNotFound(MessageQueue& mq, |
| const std::string& subExpression, |
| long long offset, |
| int maxNums) |
| { |
| return pullSyncImpl(mq, subExpression, offset, maxNums, true); |
| } |
| |
| void DefaultMQPullConsumerImpl::pullBlockIfNotFound(MessageQueue& mq, |
| const std::string& subExpression, |
| long long offset, |
| int maxNums, |
| PullCallback* pPullCallback) |
| { |
| pullAsyncImpl(mq, subExpression, offset, maxNums, pPullCallback, true); |
| } |
| |
| QueryResult DefaultMQPullConsumerImpl::queryMessage(const std::string& topic, |
| const std::string& key, |
| int maxNum, |
| long long begin, |
| long long end) |
| { |
| makeSureStateOK(); |
| |
| QueryResult result(0, std::list<MessageExt*>()); |
| return m_pMQClientFactory->getMQAdminImpl()->queryMessage(topic, key, maxNum, begin, end); |
| } |
| |
| long long DefaultMQPullConsumerImpl::searchOffset(const MessageQueue& mq, long long timestamp) |
| { |
| makeSureStateOK(); |
| return m_pMQClientFactory->getMQAdminImpl()->searchOffset(mq, timestamp); |
| } |
| |
| void DefaultMQPullConsumerImpl::sendMessageBack(MessageExt& msg, int delayLevel, const std::string& brokerName) |
| { |
| return sendMessageBack(msg, delayLevel, brokerName, m_pDefaultMQPullConsumer->getConsumerGroup()); |
| } |
| |
| |
| void DefaultMQPullConsumerImpl::sendMessageBack(MessageExt& msg, int delayLevel, const std::string& brokerName, |
| const std::string& consumerGroup) |
| { |
| try |
| { |
| std::string brokerAddr = brokerName.empty() ? |
| socketAddress2IPPort(msg.getStoreHost()) : m_pMQClientFactory->findBrokerAddressInPublish(brokerName); |
| |
| m_pMQClientFactory->getMQClientAPIImpl()->consumerSendMessageBack(brokerAddr, msg, |
| consumerGroup.empty() ? m_pDefaultMQPullConsumer->getConsumerGroup() : consumerGroup, |
| delayLevel, |
| 3000); |
| } |
| catch (...) |
| { |
| RMQ_ERROR("sendMessageBack Exception, group: %s", m_pDefaultMQPullConsumer->getConsumerGroup().c_str()); |
| Message newMsg(MixAll::getRetryTopic(m_pDefaultMQPullConsumer->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_pDefaultMQPullConsumer->getMaxReconsumeTimes())); |
| newMsg.setDelayTimeLevel(3 + reTimes); |
| |
| m_pMQClientFactory->getDefaultMQProducer()->send(newMsg); |
| } |
| } |
| |
| void DefaultMQPullConsumerImpl::updateConsumeOffset(MessageQueue& mq, long long offset) |
| { |
| makeSureStateOK(); |
| m_pOffsetStore->updateOffset(mq, offset, false); |
| } |
| |
| MessageExt* DefaultMQPullConsumerImpl::viewMessage(const std::string& msgId) |
| { |
| makeSureStateOK(); |
| |
| return m_pMQClientFactory->getMQAdminImpl()->viewMessage(msgId); |
| } |
| |
| DefaultMQPullConsumer* DefaultMQPullConsumerImpl::getDefaultMQPullConsumer() |
| { |
| return m_pDefaultMQPullConsumer; |
| } |
| |
| OffsetStore* DefaultMQPullConsumerImpl::getOffsetStore() |
| { |
| return m_pOffsetStore; |
| } |
| |
| void DefaultMQPullConsumerImpl::setOffsetStore(OffsetStore* pOffsetStore) |
| { |
| m_pOffsetStore = pOffsetStore; |
| } |
| |
| void DefaultMQPullConsumerImpl::makeSureStateOK() |
| { |
| if (m_serviceState != RUNNING) |
| { |
| THROW_MQEXCEPTION(MQClientException, "The consumer service state not OK, ", -1); |
| } |
| } |
| |
| PullResult* DefaultMQPullConsumerImpl::pullSyncImpl(MessageQueue& mq, |
| const std::string& subExpression, |
| long long offset, |
| int maxNums, |
| bool block) |
| { |
| makeSureStateOK(); |
| |
| if (offset < 0) |
| { |
| THROW_MQEXCEPTION(MQClientException, "offset < 0", -1); |
| } |
| |
| if (maxNums <= 0) |
| { |
| THROW_MQEXCEPTION(MQClientException, "maxNums <= 0", -1); |
| } |
| |
| subscriptionAutomatically(mq.getTopic()); |
| |
| int sysFlag = PullSysFlag::buildSysFlag(false, block, true); |
| |
| SubscriptionDataPtr subscriptionData = NULL; |
| try |
| { |
| subscriptionData = FilterAPI::buildSubscriptionData(mq.getTopic(), subExpression); |
| } |
| catch (...) |
| { |
| THROW_MQEXCEPTION(MQClientException, "parse subscription error", -1); |
| } |
| |
| int timeoutMillis = |
| block ? m_pDefaultMQPullConsumer->getConsumerTimeoutMillisWhenSuspend() |
| : m_pDefaultMQPullConsumer->getConsumerPullTimeoutMillis(); |
| |
| PullResult* pullResult = m_pPullAPIWrapper->pullKernelImpl(// |
| mq, // 1 |
| subscriptionData->getSubString(), // 2 |
| 0L, // 3 |
| offset, // 4 |
| maxNums, // 5 |
| sysFlag, // 6 |
| 0, // 7 |
| m_pDefaultMQPullConsumer->getBrokerSuspendMaxTimeMillis(), // 8 |
| timeoutMillis, // 9 |
| SYNC, // 10 |
| NULL// 11 |
| ); |
| |
| return m_pPullAPIWrapper->processPullResult(mq, *pullResult, *subscriptionData); |
| } |
| |
| void DefaultMQPullConsumerImpl::subscriptionAutomatically(const std::string& topic) |
| { |
| std::map<std::string, SubscriptionData>& sd = m_pRebalanceImpl->getSubscriptionInner(); |
| std::map<std::string, SubscriptionData>::iterator it = sd.find(topic); |
| |
| if (it == sd.end()) |
| { |
| try |
| { |
| SubscriptionDataPtr subscriptionData = |
| FilterAPI::buildSubscriptionData(topic, SubscriptionData::SUB_ALL); |
| sd[topic] = *subscriptionData; |
| } |
| catch (...) |
| { |
| RMQ_WARN("FilterAPI::buildSubscriptionData exception"); |
| } |
| } |
| } |
| |
| void DefaultMQPullConsumerImpl::pullAsyncImpl(// |
| MessageQueue& mq, const std::string& subExpression, long long offset, int maxNums, |
| PullCallback* pPullCallback,// |
| bool block) |
| { |
| makeSureStateOK(); |
| |
| if (offset < 0) |
| { |
| THROW_MQEXCEPTION(MQClientException, "offset < 0", -1); |
| } |
| |
| if (maxNums <= 0) |
| { |
| THROW_MQEXCEPTION(MQClientException, "maxNums <= 0", -1); |
| } |
| |
| if (pPullCallback == NULL) |
| { |
| THROW_MQEXCEPTION(MQClientException, "pullCallback is null", -1); |
| } |
| |
| subscriptionAutomatically(mq.getTopic()); |
| try |
| { |
| int sysFlag = PullSysFlag::buildSysFlag(false, block, true); |
| |
| SubscriptionDataPtr subscriptionData = NULL; |
| try |
| { |
| subscriptionData = FilterAPI::buildSubscriptionData(mq.getTopic(), subExpression); |
| } |
| catch (...) |
| { |
| THROW_MQEXCEPTION(MQClientException, "parse subscription error", -1); |
| } |
| |
| int timeoutMillis = |
| block ? m_pDefaultMQPullConsumer->getConsumerTimeoutMillisWhenSuspend() |
| : m_pDefaultMQPullConsumer->getConsumerPullTimeoutMillis(); |
| DefaultMQPullConsumerImplCallback* callback = |
| new DefaultMQPullConsumerImplCallback(*subscriptionData, |
| mq, this, pPullCallback); |
| |
| m_pPullAPIWrapper->pullKernelImpl( |
| mq, // 1 |
| subscriptionData->getSubString(), // 2 |
| 0L, // 3 |
| offset, // 4 |
| maxNums, // 5 |
| sysFlag, // 6 |
| 0, // 7 |
| m_pDefaultMQPullConsumer->getBrokerSuspendMaxTimeMillis(), // 8 |
| timeoutMillis, // 9 |
| ASYNC, // 10 |
| callback// 11 |
| ); |
| } |
| catch (const MQBrokerException& e) |
| { |
| THROW_MQEXCEPTION(MQClientException, "pullAsync unknow exception", -1); |
| } |
| } |
| |
| |
| void DefaultMQPullConsumerImpl::copySubscription() |
| { |
| try |
| { |
| std::set<std::string> registerTopics = m_pDefaultMQPullConsumer->getRegisterTopics(); |
| std::set<std::string>::iterator it = registerTopics.begin(); |
| |
| for (; it != registerTopics.end(); it++) |
| { |
| SubscriptionDataPtr subscriptionData = |
| FilterAPI::buildSubscriptionData(*it, SubscriptionData::SUB_ALL); |
| m_pRebalanceImpl->getSubscriptionInner()[*it] = *subscriptionData; |
| } |
| } |
| catch (...) |
| { |
| THROW_MQEXCEPTION(MQClientException, "subscription exception", -1); |
| } |
| } |
| |
| |
| void DefaultMQPullConsumerImpl::checkConfig() |
| { |
| // check consumerGroup |
| Validators::checkGroup(m_pDefaultMQPullConsumer->getConsumerGroup()); |
| |
| // consumerGroup |
| if (m_pDefaultMQPullConsumer->getConsumerGroup() == MixAll::DEFAULT_CONSUMER_GROUP) |
| { |
| THROW_MQEXCEPTION(MQClientException, "consumerGroup can not equal " |
| + MixAll::DEFAULT_CONSUMER_GROUP // |
| + ", please specify another one.", -1); |
| } |
| |
| if (m_pDefaultMQPullConsumer->getMessageModel() != BROADCASTING |
| && m_pDefaultMQPullConsumer->getMessageModel() != CLUSTERING) |
| { |
| THROW_MQEXCEPTION(MQClientException, "messageModel is valid ", -1); |
| } |
| |
| // allocateMessageQueueStrategy |
| if (m_pDefaultMQPullConsumer->getAllocateMessageQueueStrategy() == NULL) |
| { |
| THROW_MQEXCEPTION(MQClientException, "allocateMessageQueueStrategy is null", -1); |
| } |
| } |
| |
| ServiceState DefaultMQPullConsumerImpl::getServiceState() |
| { |
| return m_serviceState; |
| } |
| |
| void DefaultMQPullConsumerImpl::setServiceState(ServiceState serviceState) |
| { |
| m_serviceState = serviceState; |
| } |
| |
| |
| } |
