近段日子在做一个比较复杂的项目，其中用到了开源软件ZMQ和MessagePack。ZMQ对底层网络通信进行了封装，是一个消息处理队列库， 使用起来非常方便。MessagePack是一个基于二进制的对象序列化类库，具有跨语言的特性，同样非常容易使用。在我做的项目中，消息类通过 MessagePack进行压包，然后写入ZMQ的消息结构体，通过ZMQ传递，最后接收者利用MessagePack进行解包，从而分析命令。由于我英 语水平实在不高，所以我并没有通过阅读它们的说明文档来对它们进行了解，而仅仅是通过它们的示例代码进行探索。虽然因此遇到了一些不解问题，但这种方式却 为我节省了很多时间。不过，对于英语好的人，还是应该通过阅读说明文档来去了解它们。
为了说明如何使用它们，在这里构造一个使用场景：有N个Client，一个Server，M个Agent，Client使用ZMQ的请求-响应 模式和Server通信，Server收到Client的命令后，通过ZMQ的发布-订阅模式与各个Agent进行通信。下面的代码封装并使用了ZMQ和 MessagePack，为了简便，我把类的定义和实现都写在了头文件。
1.对ZMQ的简单封装：
#include"Msgpack.h"
#include<zmq.h>
#include<string>
#include<cassert>
#include<iostream> namespace Tool
{
//网络工具类
class Network
{
public: // 功能 ：构造函数。
// 参数 ：无。
// 返回 ：无。
Network() : m_socket(NULL) { } // 功能 ：初始化socket。
// 参数 ：zmqType表示ZMQ的模式，address表示socket绑定或连接地址。
// 返回 ：true表示初始化成功，false表示失败。
bool Init(int zmqType,const std::string& address)
{
try
{
m_socket = zmq_socket(Context,zmqType);
return SetSocket(zmqType,address);
}
catch(...)
{
std::cout << "Network初始化失败。" << std::endl;
return false;
}
} // 功能 ：发送消息。
// 参数 ：指向Msgpack的指针，isRelease如果为true表示发送消息后即刻释放资源。
// 返回 ：true表示发送成功，false表示发送失败。
bool SendMessage(Msgpack *msgpack,bool isRelease = true) const
{
try
{
zmq_msg_t msg;
zmq_msg_init(&msg);
if(isRelease)
{
zmq_msg_init_data(&msg,msgpack->GetSbuf().data(),msgpack->GetSbuf().size(),Tool::Network::Release,msgpack);
}
else
{
zmq_msg_init_data(&msg,msgpack->GetSbuf().data(),msgpack->GetSbuf().size(),,);
}
zmq_msg_send(&msg,m_socket,);
return true;
}
catch(...)
{
std::cout << "Network发送失败。" << std::endl;
return false;
}
} // 功能 ：接收消息。
// 参数 ：无。
// 返回 ：指向消息的指针。
zmq_msg_t* ReceiveMessage() const
{
zmq_msg_t *reply = NULL;
try
{
reply = new zmq_msg_t();
zmq_msg_init(reply);
zmq_msg_recv(reply,m_socket,);
return reply;
}
catch(...)
{
if( reply != NULL )
{
delete reply;
}
return NULL;
}
} // 功能 ：关闭消息。
// 参数 ：指向消息的指针。
// 返回 ：无。
void CloseMsg(zmq_msg_t* msg)
{
try
{
zmq_msg_close(msg);
msg = NULL;
}
catch(...)
{
msg = NULL;
}
} // 功能 ：析构函数。
// 参数 ：无。
// 返回 ：无。
~Network()
{
if( m_socket != NULL )
{
zmq_close(m_socket);
m_socket = NULL;
}
} private: //通信socket
void *m_socket; //网络环境
static void *Context; private: // 功能 ：设置socket。
// 参数 ：zmqType表示ZMQ的模式，address表示socket绑定或连接地址。
// 返回 ：true表示设置成功，false表示设置失败。
bool SetSocket(int zmqType,const std::string& address)
{
int result = -;
switch(zmqType)
{
case ZMQ_REP:
case ZMQ_PUB:
result = zmq_bind(m_socket,address.c_str());
break;
case ZMQ_REQ:
result = zmq_connect(m_socket,address.c_str());
break;
case ZMQ_SUB:
result = zmq_connect(m_socket,address.c_str());
assert(result == );
result = zmq_setsockopt(m_socket,ZMQ_SUBSCRIBE,"",);
break;
default:
return false;
}
assert( result == );
return true;
} // 功能 ：发送完消息后，释放消息资源。
// 参数 ：function为函数地址，hint指向要释放资源的对象。
// 返回 ：无。
static void Release(void *function, void *hint)
{
Msgpack *msgpack = (Msgpack*)hint;
if( msgpack != NULL )
{
delete msgpack;
msgpack = NULL;
}
}
}; //整个程序共用一个context
void *Tool::Network::Context = zmq_ctx_new();
};
说明：
（1）由zmq_ctx_new创建出来的Context，整个应用程序共用一个就可以了，具体的通信是由zmq_socket创建的socket来完成的。上述代码中没有去释放Context指向的资源。
（2）在zmq_msg_init_data函数的参数中，需要传入一个释放资源的函数地址，在ZMQ发送完消息后就调用这个函数来释放资源。 如果没有传入这个参数，而且传入的信息是临时变量，那么接收方很有可能接收不到信息，甚至抛出异常。如果不传入这个参数，那么就要记得由自己去释放资源 了。
2.对MessagePack的简单封装：
#include"BaseMessage.h"
#include"ClientMessage.h"
#include"ServerMessage.h"
#include<zmq.h>
#include<msgpack.hpp> namespace Tool
{
using namespace Message; //压包/解包工具类
class Msgpack
{
public: // 功能 ：构造函数。
// 参数 ：无。
// 返回 ：无。
Msgpack(void) { } // 功能 ：析构函数。
// 参数 ：无。
// 返回 ：无。
~Msgpack(void) { } // 功能 ：压包数据。
// 参数 ：要压包的数据。
// 返回 ：true表示压包成功。
template<typename T>
bool Pack(const T& t)
{
try
{
Release();
msgpack::pack(m_sbuf,t);
return true;
}
catch(...)
{
std::cout << "Msgpack压包数据失败。" << std::endl;
return false;
}
} // 功能 ：解包数据。
// 参数 ：zmq消息体。
// 返回 ：返回指向基类消息的指针。
BaseMessage* Unpack(zmq_msg_t& msg)
{
try
{
int size = zmq_msg_size(&msg);
if( size > )
{
Release();
m_sbuf.write((char*)zmq_msg_data(&msg),size);
size_t offset = ;
msgpack::zone z;
msgpack::object obj;
msgpack::unpack(m_sbuf.data(),m_sbuf.size(),&offset,&z,&obj);
return GetMessage(obj);
}
}
catch(...)
{
//吃掉异常
}
return NULL;
} // 功能 ：获取压包/解包工具。
// 参数 ：无。
// 返回 ：压包/解包工具。
inline msgpack::sbuffer& GetSbuf()
{
return m_sbuf;
} private: //压包/解包工具
msgpack::sbuffer m_sbuf; private: // 功能 ：释放上一次的数据资源。
// 参数 ：无。
// 返回 ：无。
void Release()
{
m_sbuf.clear();
m_sbuf.release();
} // 功能 ：获取消息。
// 参数 ：用于转换的msgpack::object。
// 返回 ：指向消息基类的指针。
BaseMessage* GetMessage(const msgpack::object& obj)
{
BaseMessage bmessage;
obj.convert(&bmessage);
switch(bmessage.Type)
{
case :
return Convert<ClientMessage>(obj);
case :
return Convert<ServerMessage>(obj);
default:
return NULL;
}
} // 功能 ：将压包后的数据转换为具体的类。
// 参数 ：用于转换的msgpack::object。
// 返回 ：指向T的指针。
template<typename T>
T* Convert(const msgpack::object& obj)
{
T *t = new T();
obj.convert(t);
return t;
}
};
};
说明：
压包时将zmq_msg_t消息体压包到msgpack::sbuffer，然后就可以关闭这个消息体了。要将解包后的数据转换成具体的某一个类，需要知道这个类是什么类，这里有三种方法：
（1）可以先发送一个消息告知接收者即将收到什么消息，然后接收者将消息解包后转换成对应的类。这种方式需要额外的一次通信，不建议使用。
（2）所有的消息都继承自一个基类，这个基类存储有消息类型的字段。解包后，先将数据转换为基类，然后根据类型再转换为具体的派生类。这种方式需要多转换一次，上面的代码也正是采用这种方式。
（3）压包时先压包一个消息类，然后再压包一个标识这个消息是什么类型的标 识类，即压包两次。解包时，先解包标识类，得知消息类的具体类型，然后再解包消息类，即解包两次，转换两次。与（2）相比，除了要做更多的压包、解包工作 外，这里还需要对解包的偏移量进行计算，否则容易出错。
3.使用到的消息类：
namespace Message
{
//消息基类
class BaseMessage
{
public: MSGPACK_DEFINE(Type); //消息类型
int Type; //默认构造函数
BaseMessage()
{
Type = ;
}
}; //来自客户端的消息
class ClientMessage : public BaseMessage
{
public: MSGPACK_DEFINE(Type,Information); //信息
std::string Information; //默认构造函数
ClientMessage()
{
Type = ;
}
}; //来自服务端的消息
class ServerMessage : public BaseMessage
{
public: MSGPACK_DEFINE(Type,Information); //信息
std::vector<std::string> Information; //默认构造函数
ServerMessage()
{
Type = ;
}
};
};
说明：
（1）MSPACK_DEFINE标识了一个类的哪些成员可以进行压包/解包。派生类中的MSGPACK_DEFINE还需要写上基类的成员，否则无法使用对MessagePack封装说明的第二个方法。
（2）C++版本的MessagePack压/解包的数据成员，只能是一个类、结构或者联合体，不能使用指针（包括boost库的智能指针）、 数组，枚举值也不适用。因此，BaseMessage使用int值来标识派生类属于哪个类型。C#版本的MessagePack可以对枚举值进行压包。
4.Client的示例代码：
int _tmain(int argc, _TCHAR* argv[])
{
Network network;
bool result = network.Init(ZMQ_REQ,"tcp://192.168.10.179:8888");
if(result)
{
ClientMessage cmessage;
cmessage.Information = "I come form Client."; Msgpack msgpack;
result = msgpack.Pack<ClientMessage>(cmessage);
if(result)
{
result = network.SendMessageW(&msgpack,false);
if(result)
{
zmq_msg_t *msg = network.ReceiveMessage();
if( msg != NULL )
{
BaseMessage *bmessage = msgpack.Unpack(*msg);
network.CloseMsg(msg);
if( bmessage != NULL && bmessage->Type == )
{
ServerMessage *smessage = static_cast<ServerMessage*>(bmessage);
if( smessage != NULL && smessage->Information.size() > )
{
std::cout << smessage->Information[] << std::endl;
}
delete smessage;
smessage = NULL;
bmessage = NULL;
}
}
}
}
} system("pause");
return ;
}
5.Server的示例代码：
int _tmain(int argc, _TCHAR* argv[])
{
Network responder;
bool result = responder.Init(ZMQ_REP,"tcp://192.168.10.179:8888");
if(result)
{
Network publisher;
result = publisher.Init(ZMQ_PUB,"tcp://192.168.10.179:9999");
if(result)
{
Msgpack msgpack;
while(true)
{
zmq_msg_t *msg = responder.ReceiveMessage();
BaseMessage *bmessage = msgpack.Unpack(*msg);
responder.CloseMsg(msg); ServerMessage smessage;
smessage.Information.push_back("I come from Server.");
msgpack.Pack<ServerMessage>(smessage);
result = responder.SendMessageW(&msgpack,false); if( result )
{
if( bmessage != NULL && bmessage->Type == )
{
ClientMessage *cmessage = static_cast<ClientMessage*>(bmessage);
if( cmessage != NULL )
{
std::cout << cmessage->Information << std::endl;
for( int counter = ; counter < ; counter++ )
{
publisher.SendMessageW(&msgpack,false);
}
}
delete cmessage;
cmessage = NULL;
bmessage = NULL;
}
}
}
}
} return ;
}
6.Agent的示例代码：
int _tmain(int argc, _TCHAR* argv[])
{
Network network;
bool result = network.Init(ZMQ_SUB,"tcp://192.168.10.179:9999");
if(result)
{
zmq_msg_t *msg = network.ReceiveMessage();
if( msg != NULL )
{
Msgpack msgpack;
BaseMessage *bmessage = msgpack.Unpack(*msg);
network.CloseMsg(msg);
if( bmessage->Type == )
{
ServerMessage *smessage = static_cast<ServerMessage*>(bmessage);
if( smessage->Information.size() > )
{
std::cout << smessage->Information[] << std::endl;
}
delete smessage;
smessage = NULL;
bmessage = NULL;
}
}
} system("pause");
return ;
}
7.启动这三个程序，Client将要发送的消息压包后发给Server，Server接收到消息后反馈一个信息给Client，然后循环发布消息给Agent，Agent不需要回复Server。最后着重说明两点：
（1）ZMQ创建的socket发送数据和接收数据要处在同一条线程。Server接收到Client的数据后，不能通过开一条线程来给Client反馈信息，必须要在接收数据的线程中反馈信息。
（2）ZMQ并不要求发送者和接收者有一定的启动顺序，但在Server中如果只发布一次消息，那么Agent很有可能收不到信息。不管是 Agent先启动，还是Server先启动，Agent都有可能收不到信息。在Server的代码中，通过循环发布一百次，来让Agent收到信息。至于 实际应用中，可以结合请求-响应模式来保证订阅消息者都收到了发布者的消息。
参考资料：
ZMQ：http://zguide.zeromq.org/page:all
MessagePack：http://wiki.msgpack.org/pages/viewpage.action?pageId=1081387#QuickStartforC%2B%2B-ImplementationStatus