ZeroMQ发布-订阅模式的应用(C++)

我使用的ZeroMQ版本是4.2.0，应用的是其发布-订阅模式

应该知道的细节：
PUB-SUB套接字是慢连接，你无法得知SUB是何时开始接收消息的。就算你先打开了SUB套接字，后打开PUB发送消息，这时SUB还是会丢失一些消息的，因为建立连接是需要一些时间的。很少，但并不是零。参考文章：传送门1.传送门2.

先看代码，参考该文章：ZeroMQ的订阅发布（publish-subscribe）模式

//server--publish
#include <iostream>
#include <string>
#include <sstream>
#include <cstring>
#include <iomanip>
#include <cstdlib>
#include <ctime>
#include <assert.h>
#include <zmq.h>

using namespace std;

int main ()
{
    void *context = zmq_ctx_new ();
    void *publisher = zmq_socket (context, ZMQ_PUB);
    int rc = zmq_bind (publisher, "tcp://127.0.0.1:5556");
    assert (rc == 0);

    //  Initialize random number generator
    srand(time(0));
    while (1) {
        int zipcode, temperature, relhumidity;
        zipcode     = rand() % 100000;
        temperature = rand() % 215 - 80;
        relhumidity = rand() % 50 + 10;

        ostringstream os;
        os << setw(5) << setfill('0')<< zipcode <<" "
           << temperature <<" "<< relhumidity << "\0";

		int len = strlen(os.str().c_str());
		zmq_msg_t msg;
		zmq_msg_init_size(&msg, len);
		
		char *message = static_cast<char *>(zmq_msg_data(&msg));
		memcpy(message, os.str().c_str(), strlen(os.str().c_str()));
		message[len] = '\0';
		int size = zmq_msg_send(&msg, publisher, 0);
		assert(len == size);
    }
    zmq_close (publisher);
    zmq_ctx_destroy (context);
    return 0;
}

//client--subscribe
#include <iostream>
#include <string>
#include <cstring>
#include <assert.h>
#include <zmq.h>

using namespace std;

int main (int argc, char *argv [])
{
    //  Socket to talk to server
    printf ("Collecting updates from weather server...\n");
    void *context = zmq_ctx_new ();
    void *subscriber = zmq_socket (context, ZMQ_SUB);
    int rc = zmq_connect (subscriber, "tcp://localhost:5556");
    assert (rc == 0);

    char filter1[] = "10001 ";
    char filter2[] = "20002 ";
    rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,filter1, strlen (filter1));
    assert (rc == 0);
    rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,filter2, strlen (filter2));
    assert (rc == 0);

    //  Process 100 updates
    int size;
    char buffer [256];
    zmq_msg_t msg;
    rc = zmq_msg_init(&msg);
    assert(0 == rc);
    
    for (int update_nbr = 0; update_nbr < 100; update_nbr++) {

        int size = zmq_msg_recv(&msg, subscriber, 0);
         if (size == -1){
            cout<< "receiver error , skip this message"<<endl;
            continue;
        }
		static_cast<char*>(zmq_msg_data(&msg))[size] = '\0';
		
        cout <<static_cast<char*>(zmq_msg_data(&msg)) <<endl;
    }

    zmq_close (subscriber);
    zmq_ctx_destroy (context);
    return 0;
}

ZeroMQ官方API：http://api.zeromq.org/4-2:_start

几个要点

1. 没有什么特别的要求我觉得还是使用zmq_send()和zmq_recv()这两个函数来进行数据的收发比较好，简单直接。
2. 初始化消息推荐使用zmq_msg_init_size(zmq_msg_t*, size_t len)，我在使用的时候，发现服务器这边如果使用zmq_init(zmq_msg_t)来初始化消息，zmq_msg_send()返回值是0,这意味着发出去的字节数是0,客户端也收不到收据，我搞不懂这是什么原因，换成zmq_msg_init_size()来初始化消息就能返回发送消息的字节数了。客户端这边我用zmq_msg_init()初始化消息能正常接收到来自服务器的消息。
3. 使用消息，推荐将消息的创建放在循环体内，看别人的博客提到消息发送成功后就会被释放，再次使用的时候就会出现问题。所以每次循环的时候都创建一个消息，然后初始化后使用。

参考文章：https://www.cnblogs.com/chenny7/p/6245236.html