发布-订阅（Publish-Subscribe）模式 C++实现

#include "MessageCentrer.h" #include "Subscriber.h" #include "Publisher.h" #define MAX_PUBLISHES 10000 MessageCentrer* MessageCentrer::mSgMC = nullptr; std::mutex MessageCentrer::mMCMutex; MessageCentrer* MessageCentrer::GetMC() { if (mSgMC == nullptr) { std::unique_lock<std::mutex> lock(mMCMutex); if (mSgMC == nullptr) { volatile auto temp = new (std::nothrow) MessageCentrer(); mSgMC = temp; } } return mSgMC; } void MessageCentrer::Run() { mCoreProcss.reset(new std::thread(&MessageCentrer::CoreProcss, this)); } void MessageCentrer::RegistPublish(std::string tpcKey, void* msgdata, unsigned int datasize) { std::unique_lock<std::mutex> lock(mPublishMutex); if ((this->mSubscriber.find(tpcKey) != this->mSubscriber.end()) && (this->mPublisher[tpcKey].size() > MAX_PUBLISHES)) return; void* tmpdata = new char[datasize]; memcpy(tmpdata, msgdata, datasize); this->mPublisher[tpcKey].push_back(tmpdata); } void MessageCentrer::RegistSubscribe(std::string tpcKey, Subscriber* subscriber) { std::unique_lock<std::mutex> lock(mSubscribeMutex); this->mSubscriber[tpcKey].remove(subscriber); this->mSubscriber[tpcKey].push_back(subscriber); } void MessageCentrer::CancelSubscribe(std::string tpcKey, Subscriber* subscriber) { std::unique_lock<std::mutex> lock(mSubscribeMutex); if (this->mSubscriber.find(tpcKey) != this->mSubscriber.end()) this->mSubscriber.find(tpcKey)->second.remove(subscriber); } MessageCentrer::MessageCentrer() { this->mPublisher.clear(); this->mSubscriber.clear(); } MessageCentrer::~MessageCentrer() { } void MessageCentrer::CoreProcss() { while (true) { auto it = this->mSubscriber.begin(); while (it != this->mSubscriber.end()) { if (this->mPublisher.find(it->first) != this->mPublisher.end()) { auto itt = it->second.begin(); while (itt != it->second.end()) { auto mpitr = this->mPublisher.find(it->first)->second.begin(); auto mpitrend = this->mPublisher.find(it->first)->second.end(); while (mpitr != mpitrend) { (*itt)->HandeEnvent(it->first, *mpitr); ++mpitr; } ++itt; } mPublishMutex.lock(); auto mpitr = this->mPublisher.find(it->first)->second.begin(); auto mpitrend = this->mPublisher.find(it->first)->second.end(); while (mpitr != mpitrend) { delete[](*mpitr); ++mpitr; } this->mPublisher.find(it->first)->second.clear(); this->mPublisher.erase(it->first); mPublishMutex.unlock(); } ++it; } } }