发布-订阅（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;
		}
	}
}