zeromq 发布订阅 无数据丢失性能测试

sub接收端代码：

#include <zmq.hpp>
#include <assert.h>
#include <iostream>
#include <string>

using namespace std;
#define BAG_LARGE_NUM			14

long long int gettime(){  
	struct timeval tv = {0, 0};  
	gettimeofday(&tv, NULL);  
	return tv.tv_sec * 1000000 + tv.tv_usec;  
}

int main(int vArgc, char** vArgv)
{
	char* ProtocoltConSub			= vArgv[1];
	char* ProtocoltConReq			= vArgv[2];
	cout<<"ProtocoltConSub: "<<ProtocoltConSub<<endl;
	cout<<"ProtocoltConReq: "<<ProtocoltConReq<<endl;
	cout<<"***********************Client Start********************"<<endl;
	zmq::context_t		tContext(1);
	zmq::socket_t		tSocketSub(tContext, ZMQ_SUB);
	zmq::socket_t		tSocketReq(tContext, ZMQ_REQ);
	tSocketSub.setsockopt(ZMQ_SUBSCRIBE, "", 0);
	tSocketSub.connect(ProtocoltConSub);
	tSocketReq.connect(ProtocoltConReq);

	long long				tTimeStart, tTimeEnd;
	zmq::message_t		tMsgReq(32), tMsgSub(32);
	//同步pub sub
	tSocketReq.send(tMsgReq);
	tMsgReq.rebuild();
	tSocketReq.recv(&tMsgReq);
	tMsgSub.rebuild();
	tSocketSub.recv(&tMsgSub);  //等待pub端所有req
	for(int i=0; i<BAG_LARGE_NUM; i++)
	{
		long long tLoopTimes = atoi((char*)tMsgReq.data());
		tTimeStart = gettime();
		for (int k=0; tLoopTimes; k++)
		{
			if(0 == k%500)
			{//sub pub同步
				tMsgReq.rebuild(32);
				tSocketReq.send(tMsgReq);
				tMsgReq.rebuild();
				tSocketReq.recv(&tMsgReq);
			}
			tMsgSub.rebuild();
			tSocketSub.recv(&tMsgSub);
		}
		tTimeEnd	= gettime();
		cout<<"bag large: "<<tMsgSub.size()<<" take time: "<<tTimeEnd-tTimeStart<<" tps: "<<tLoopTimes/(tTimeEnd-tTimeStart) <<endl;
		//更新tLoopTimes
		tMsgReq.rebuild(32);
		tSocketReq.send(tMsgReq);
		tMsgReq.rebuild();
		tSocketReq.recv(&tMsgReq);
	}
	cout<<"finish"<<endl;
	return 0;
}
           

./Client tcp://localhost:5555 tcp://localhost:5556

pub发布端代码：

#include <zmq.hpp>
#include <assert.h>
#include <iostream>

using namespace std;
#define BAG_LARGE_NUM			14

long long int gettime(){  
	struct timeval tv = {0, 0};  
	gettimeofday(&tv, NULL);  
	return tv.tv_sec * 1000000 + tv.tv_usec;  
}

int main(int vArgc, char** vArgv)
{
	char* tProtocolPub		= vArgv[1];
	char* tProtocolRep		= vArgv[2];
	int tClientNum			= atoi(vArgv[3]);
	long long tLoopL		= atoi(vArgv[4]);
	long long tLoopB		= atoi(vArgv[5]);

	cout<<"tProtocolPub: "<<tProtocolPub<<endl;
	cout<<"tProtocolRep: "<<tProtocolRep<<endl;
	cout<<"tLoopL: "<<tLoopL<<endl;
	cout<<"tLoopB: "<<tLoopB<<endl;
	cout<<"ClientNum: "<<tClientNum<<endl;
	cout<<"**********************Server Start*******************"<<endl;
	zmq::context_t		tContext(1);
	zmq::socket_t		tSocketPub(tContext, ZMQ_PUB);
	zmq::socket_t		tSocketRep(tContext, ZMQ_REP);
	tSocketPub.bind(tProtocolPub);
	tSocketRep.bind(tProtocolRep);

	std::string tData;
	zmq::message_t tMsgRecv(32), tMsgSend(32), tMsg(32);
	long long tBagLarge[BAG_LARGE_NUM] = {1, 4, 8, 32, 64, 128, 256, 512,1024,1024*8, 1024*64, 1024*128, 1024*512, 1024*1024};
	long long tChansportLoop = tLoopL;
	long long			tTimeStart, tTimeEnd;
	for (int i=0; i<tClientNum; i++)
	{//pub sub同步
		tMsgRecv.rebuild();
		tSocketRep.recv(&tMsgRecv, 0);
		tMsgSend.rebuild(sizeof(tBagLarge[0]));
		tSocketRep.send(tMsgSend);
	}
	
	for (int i = 0; i<BAG_LARGE_NUM; i++)
	{
		tTimeStart = gettime();
		for (long long t=0; t<tChansportLoop; t++)
		{
			if(0 == t%500)
			{
				for (int j=0; j<tClientNum; j++)
				{//pub sub同步
					tMsg.rebuild();
					tSocketRep.recv(&tMsg, 0);
					tMsg.rebuild(32);
					tSocketRep.send(tMsg, 0);
				}
			}
			tMsgSend.rebuild(tBagLarge[i]);
			tSocketPub.send(tMsgSend, 0);
		}
		tTimeEnd = gettime();
		cout<<"bag large: "<<tBagLarge[i]<<" take time: "<<tTimeEnd-tTimeStart<<" tps: "<<tChansportLoop/(tTimeEnd-tTimeStart) <<endl;
		//next chansport loop
		tMsg.rebuild();
		tSocketRep.recv(&tMsg);
		tMsg.rebuild(tBagLarge[i+1]);
		tSocketRep.send(tMsg);
	}

	cout<<"finish"<<endl;
	return 0;
}
           

脚本命令：./Server tcp://*:5555 tcp://*:5556 3 10000 100000