ZMQ和MessagePack的簡單使用

　　近段日子在做一個比較復雜的項目，其中用到了開源軟件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(),0,0);
                }
                zmq_msg_send(&msg,m_socket,0);
                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,0);
                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 = -1;
            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 == 0);
                result = zmq_setsockopt(m_socket,ZMQ_SUBSCRIBE,"",0);
                break;
            default:
                return false;
            }
            assert( result == 0 );
            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 > 0 )
                {
                    Release();
                    m_sbuf.write((char*)zmq_msg_data(&msg),size);
                    size_t offset = 0;
                    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 1024:
                return Convert<ClientMessage>(obj);
            case 2048:
                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 = 0;
        }
    };

    //來自客戶端的消息
    class ClientMessage : public BaseMessage
    {
    public:

        MSGPACK_DEFINE(Type,Information);

        //信息
        std::string Information;

        //默認構造函數
        ClientMessage()
        {
            Type = 1024;
        }
    };

    //來自服務端的消息
    class ServerMessage : public BaseMessage
    {
    public:

        MSGPACK_DEFINE(Type,Information);

        //信息
        std::vector<std::string> Information;

        //默認構造函數
        ServerMessage()
        {
            Type = 2048;
        }
    };
}; 

　　說明：

　　（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 == 2048 )
                    {
                        ServerMessage *smessage = static_cast<ServerMessage*>(bmessage);
                        if( smessage != NULL && smessage->Information.size() > 0 )
                        {
                            std::cout << smessage->Information[0] << std::endl;
                        }
                        delete smessage;
                        smessage = NULL;
                        bmessage = NULL;
                    }
                }
            }
        }
    }

    system("pause");
    return 0;
}

　　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 == 1024 )
                    {
                        ClientMessage *cmessage = static_cast<ClientMessage*>(bmessage);
                        if( cmessage != NULL )
                        {
                            std::cout << cmessage->Information << std::endl;
                            for( int counter = 0 ; counter < 100 ; counter++ )
                            {
                                publisher.SendMessageW(&msgpack,false);
                            }
                        }
                        delete cmessage;
                        cmessage = NULL;
                        bmessage = NULL;
                    }
                }
            }
        }
    }

    return 0;
}

　　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 == 2048 )
            {
                ServerMessage *smessage = static_cast<ServerMessage*>(bmessage);
                if( smessage->Information.size() > 0 )
                {
                    std::cout << smessage->Information[0] << std::endl;
                }
                delete smessage;
                smessage = NULL;
                bmessage = NULL;
            }
        }
    }

    system("pause");
    return 0;
}

　　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