| /** |
| * 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 "RemoteBrokerOffsetStore.h" |
| #include "MQClientFactory.h" |
| #include "ScopedLock.h" |
| #include "MQClientException.h" |
| #include "CommandCustomHeader.h" |
| #include "MQClientAPIImpl.h" |
| |
| namespace rmq |
| { |
| |
| RemoteBrokerOffsetStore::RemoteBrokerOffsetStore(MQClientFactory* pMQClientFactory, const std::string& groupName) |
| { |
| m_pMQClientFactory = pMQClientFactory; |
| m_groupName = groupName; |
| } |
| |
| void RemoteBrokerOffsetStore::load() |
| { |
| |
| } |
| |
| void RemoteBrokerOffsetStore::updateOffset(const MessageQueue& mq, long long offset, bool increaseOnly) |
| { |
| kpr::ScopedWLock<kpr::RWMutex> lock(m_tableMutex); |
| typeof(m_offsetTable.begin()) it = m_offsetTable.find(mq); |
| if (it == m_offsetTable.end()) |
| { |
| m_offsetTable[mq] = offset; |
| it = m_offsetTable.find(mq); |
| } |
| |
| kpr::AtomicLong& offsetOld = it->second; |
| if (increaseOnly) |
| { |
| MixAll::compareAndIncreaseOnly(offsetOld, offset); |
| } |
| else |
| { |
| offsetOld.set(offset); |
| } |
| } |
| |
| long long RemoteBrokerOffsetStore::readOffset(const MessageQueue& mq, ReadOffsetType type) |
| { |
| RMQ_DEBUG("readOffset, MQ:%s, type:%d", mq.toString().c_str(), type); |
| switch (type) |
| { |
| case MEMORY_FIRST_THEN_STORE: |
| case READ_FROM_MEMORY: |
| { |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_tableMutex); |
| typeof(m_offsetTable.begin()) it = m_offsetTable.find(mq); |
| if (it != m_offsetTable.end()) |
| { |
| return it->second.get(); |
| } |
| else if (READ_FROM_MEMORY == type) |
| { |
| RMQ_DEBUG("No offset in memory, MQ:%s", mq.toString().c_str()); |
| return -1; |
| } |
| } |
| case READ_FROM_STORE: |
| { |
| try |
| { |
| long long brokerOffset = this->fetchConsumeOffsetFromBroker(mq); |
| RMQ_DEBUG("fetchConsumeOffsetFromBroker, MQ:%s, brokerOffset:%lld", |
| mq.toString().c_str(), brokerOffset); |
| if (brokerOffset >= 0) |
| { |
| this->updateOffset(mq, brokerOffset, false); |
| } |
| return brokerOffset; |
| } |
| // No offset in broker |
| catch (const MQBrokerException& e) |
| { |
| RMQ_WARN("No offset in broker, MQ:%s, exception:%s", mq.toString().c_str(), e.what()); |
| return -1; |
| } |
| catch (const std::exception& e) |
| { |
| RMQ_ERROR("fetchConsumeOffsetFromBroker exception, MQ:%s, msg:%s", |
| mq.toString().c_str(), e.what()); |
| return -2; |
| } |
| catch (...) |
| { |
| RMQ_ERROR("fetchConsumeOffsetFromBroker unknow exception, MQ:%s", |
| mq.toString().c_str()); |
| return -2; |
| } |
| } |
| default: |
| break; |
| } |
| |
| return -1; |
| } |
| |
| void RemoteBrokerOffsetStore::persistAll(std::set<MessageQueue>& mqs) |
| { |
| if (mqs.empty()) |
| { |
| return; |
| } |
| |
| std::set<MessageQueue> unusedMQ; |
| long long times = m_storeTimesTotal.fetchAndAdd(1); |
| |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_tableMutex); |
| for (typeof(m_offsetTable.begin()) it = m_offsetTable.begin(); |
| it != m_offsetTable.end(); it++) |
| { |
| MessageQueue mq = it->first; |
| kpr::AtomicLong& offset = it->second; |
| if (mqs.find(mq) != mqs.end()) |
| { |
| try |
| { |
| this->updateConsumeOffsetToBroker(mq, offset.get()); |
| if ((times % 12) == 0) |
| { |
| RMQ_INFO("updateConsumeOffsetToBroker, Group: {%s} ClientId: {%s} mq:{%s} offset {%llu}", |
| m_groupName.c_str(), |
| m_pMQClientFactory->getClientId().c_str(), |
| mq.toString().c_str(), |
| offset.get()); |
| } |
| } |
| catch (...) |
| { |
| RMQ_ERROR("updateConsumeOffsetToBroker exception, mq=%s", mq.toString().c_str()); |
| } |
| } |
| else |
| { |
| unusedMQ.insert(mq); |
| } |
| } |
| |
| if (!unusedMQ.empty()) |
| { |
| for (typeof(unusedMQ.begin()) it = unusedMQ.begin(); it != unusedMQ.end(); it++) |
| { |
| m_offsetTable.erase(*it); |
| RMQ_INFO("remove unused mq, %s, %s", it->toString().c_str(), m_groupName.c_str()); |
| } |
| } |
| } |
| |
| void RemoteBrokerOffsetStore::persist(const MessageQueue& mq) |
| { |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_tableMutex); |
| typeof(m_offsetTable.begin()) it = m_offsetTable.find(mq); |
| if (it != m_offsetTable.end()) |
| { |
| try |
| { |
| this->updateConsumeOffsetToBroker(mq, it->second.get()); |
| RMQ_DEBUG("updateConsumeOffsetToBroker ok, mq=%s, offset=%lld", mq.toString().c_str(), it->second.get()); |
| } |
| catch (...) |
| { |
| RMQ_ERROR("updateConsumeOffsetToBroker exception, mq=%s", mq.toString().c_str()); |
| } |
| } |
| } |
| |
| void RemoteBrokerOffsetStore::removeOffset(const MessageQueue& mq) |
| { |
| kpr::ScopedWLock<kpr::RWMutex> lock(m_tableMutex); |
| m_offsetTable.erase(mq); |
| RMQ_INFO("remove unnecessary messageQueue offset. mq=%s, offsetTableSize=%u", |
| mq.toString().c_str(), (unsigned)m_offsetTable.size()); |
| } |
| |
| |
| std::map<MessageQueue, long long> RemoteBrokerOffsetStore::cloneOffsetTable(const std::string& topic) |
| { |
| kpr::ScopedRLock<kpr::RWMutex> lock(m_tableMutex); |
| std::map<MessageQueue, long long> cloneOffsetTable; |
| RMQ_FOR_EACH(m_offsetTable, it) |
| { |
| MessageQueue mq = it->first; |
| kpr::AtomicLong& offset = it->second; |
| if (topic == mq.getTopic()) |
| { |
| cloneOffsetTable[mq] = offset.get(); |
| } |
| } |
| |
| return cloneOffsetTable; |
| } |
| |
| |
| void RemoteBrokerOffsetStore::updateConsumeOffsetToBroker(const MessageQueue& mq, long long offset) |
| { |
| FindBrokerResult findBrokerResult = m_pMQClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()); |
| if (findBrokerResult.brokerAddr.empty()) |
| { |
| m_pMQClientFactory->updateTopicRouteInfoFromNameServer(mq.getTopic()); |
| findBrokerResult = m_pMQClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()); |
| } |
| |
| if (!findBrokerResult.brokerAddr.empty()) |
| { |
| UpdateConsumerOffsetRequestHeader* requestHeader = new UpdateConsumerOffsetRequestHeader(); |
| requestHeader->topic = mq.getTopic(); |
| requestHeader->consumerGroup = this->m_groupName; |
| requestHeader->queueId = mq.getQueueId(); |
| requestHeader->commitOffset = offset; |
| |
| m_pMQClientFactory->getMQClientAPIImpl()->updateConsumerOffsetOneway( |
| findBrokerResult.brokerAddr, requestHeader, 1000 * 5); |
| } |
| else |
| { |
| THROW_MQEXCEPTION(MQClientException, "The broker[" + mq.getBrokerName() + "] not exist", -1); |
| } |
| } |
| |
| long long RemoteBrokerOffsetStore::fetchConsumeOffsetFromBroker(const MessageQueue& mq) |
| { |
| FindBrokerResult findBrokerResult = m_pMQClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()); |
| if (findBrokerResult.brokerAddr.empty()) |
| { |
| // TODO Here may be heavily overhead for Name Server,need tuning |
| m_pMQClientFactory->updateTopicRouteInfoFromNameServer(mq.getTopic()); |
| findBrokerResult = m_pMQClientFactory->findBrokerAddressInAdmin(mq.getBrokerName()); |
| } |
| |
| if (!findBrokerResult.brokerAddr.empty()) |
| { |
| QueryConsumerOffsetRequestHeader* requestHeader = new QueryConsumerOffsetRequestHeader(); |
| requestHeader->topic = mq.getTopic(); |
| requestHeader->consumerGroup = this->m_groupName; |
| requestHeader->queueId = mq.getQueueId(); |
| |
| return m_pMQClientFactory->getMQClientAPIImpl()->queryConsumerOffset( |
| findBrokerResult.brokerAddr, requestHeader, 1000 * 5); |
| } |
| else |
| { |
| THROW_MQEXCEPTION(MQClientException, "The broker[" + mq.getBrokerName() + "] not exist", -1); |
| } |
| } |
| |
| } |
