ZeroMQ发布订阅模式之多进程实现

ZeroMQ的发布订阅模式是单向的数据发布，服务器（即消息发布方）将更新的消息/事件推送到一组客户端（即订阅方）。

消息发布者创建ZMQ_PUB类型的socket并将消息发送到消息队列中。
消息订阅者创建ZMQ_SUB类型的socket，并使用zmq_setsockopt接口和ZMQ_SUBSCRIBE订阅事件。如果socket没有设置订阅，那么将收不到任何消息。订阅者可以设置多个订阅，多个订阅将会被累加起来，如果某个事件匹配任何订阅，那么订阅者都会接收到该事件。
订阅者可以通过ZMQ_UNSUBSCRIBE取消订阅某事件。
SUB套接字可以在循环中执行zmq_msg_recv来接收消息，也可以结合zmq_poll来接收消息，但不能向PUB套接字发送消息。
同样，PUB套接字可以多次执行zmq_msg_send，但不能执行zmq_msg_recv。
示例代码改编自教材，有一些函数冗余。
zmq_help.h：封装公共接口

#ifndef _ZMQ_HELPER_H_
#define _ZMQ_HELPER_H_

#include <vector>
#include "zmq.h"

class CZmqMsg
{
public:
    CZmqMsg()
    {
        (void)zmq_msg_init(&m_stMsg);
    }

    virtual ~CZmqMsg()
    {
        (void)zmq_msg_close(&m_stMsg);
    }

    void *GetMsgData()
    {
        return zmq_msg_data(&m_stMsg);
    }

    zmq_msg_t m_stMsg;
};

long ZQM_RecvMore(void * hSocket, std::vector<zmq_msg_t>& oRecvMsgList)
{
    int iMore = 0;
    size_t iMoreSize = sizeof(iMore);

    do
    {
        zmq_msg_t stMsg;
        (void)zmq_msg_init(&stMsg);

        long lRet = zmq_msg_recv(&stMsg, hSocket, 0);
        if (-1 == lRet)
        {
            (void)zmq_msg_close(&stMsg);
            return lRet;
        }

        oRecvMsgList.push_back(stMsg);

        lRet = zmq_getsockopt(hSocket, ZMQ_RCVMORE, &iMore, &iMoreSize);
        if (-1 == lRet)
        {
            /* ´Ë´¦Ê§°Ü²»ÐèÒª¹Ø±Õ msg_data£¬Í³Ò»Óɵ÷ÓÃÕß´¦Àí */
            return lRet;
        }
    } while (iMore);

    return 0;
}

void ZMQ_Recv_Data(void *requester, int bufLen, char *pcRspBuf)
{
    zmq_msg_t reply;
    zmq_msg_init(&reply);
    zmq_msg_recv(&reply, requester, 0);

    memcpy(pcRspBuf, zmq_msg_data(&reply), bufLen);
    zmq_msg_close(&reply);

    return;
}

#endif

client.cpp：订阅者，结合zmq_poll和zmq_msg_recv来接收消息

#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/prctl.h>
#include <map>

#include "zmq_helper.h"
#include "zmq.h"

typedef void *(*CLIENT_THR_FUNC_CB)(void *);

void client_req_rsp_mode(void *)
{
    prctl(PR_SET_NAME, (unsigned long)"request");

    void *context = zmq_ctx_new();

    // Socket to talk to server
    printf("Connecting to hello world server.\n");
    void *requester = zmq_socket (context, ZMQ_REQ);
    zmq_connect (requester, "tcp://localhost:5555");

    int request_nbr;
    for (request_nbr = 0; request_nbr != 10; request_nbr++)
    {
        struct timeval start_time;
        char szSendBuf[10] = {0};
        snprintf(szSendBuf, sizeof(szSendBuf), "Hello_%03d", request_nbr);

        zmq_msg_t request;
        zmq_msg_init_data(&request, szSendBuf, strlen(szSendBuf), NULL, NULL);

        zmq_msg_send(&request, requester, 0);
        zmq_msg_close(&request);
        gettimeofday(&start_time, NULL);
        printf("Sending Hello Times:%d tv_sec:%ld tv_usec:%ld.\n", request_nbr, start_time.tv_sec, start_time.tv_usec);

        /* recv response data */
        char szRecvBuf[10] = {0};
        ZMQ_Recv_Data(requester, sizeof(szRecvBuf), szRecvBuf);
        struct timeval end_time;
        gettimeofday(&end_time, NULL);
        printf("Times:%d Received replay:%s tv_sec:%ld tv_usec:%ld\n", request_nbr, szRecvBuf, end_time.tv_sec, end_time.tv_usec);

        sleep(1);
    }

    sleep(2);
    zmq_close(requester);
    zmq_term(context);

    return;
}

void *client_create_subscriber_socket(void *context, const char *filter)
{
    void *subscriber = zmq_socket(context, ZMQ_SUB);
    
    /* 高水位选项的设置要在bind/connect之前, 否则不生效 */
    int iSocketOpt = 0;
    (void)zmq_setsockopt(subscriber, ZMQ_RCVHWM, &iSocketOpt, sizeof(iSocketOpt));
    (void)zmq_setsockopt(subscriber, ZMQ_SNDHWM, &iSocketOpt, sizeof(iSocketOpt));
    
    zmq_connect(subscriber, "tcp://localhost:5556");

    (void)zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));
    
    return subscriber;
}

std::map<void *, void *> goSubscriberSocket;

void client_subscriber(void *)
{
    printf("Collecting updates from weather server....\n");
    prctl(PR_SET_NAME, (unsigned long)"subscriber");

    void *context = zmq_ctx_new();

    void *subscriber1 = client_create_subscriber_socket(context, "10001");
    goSubscriberSocket.insert(std::pair<void *, void *>(subscriber1, subscriber1));

    void *subscriber2 = client_create_subscriber_socket(context, "10002");
    goSubscriberSocket.insert(std::pair<void *, void *>(subscriber2, subscriber2));

    void *subscriber3 = client_create_subscriber_socket(context, "10003");
    goSubscriberSocket.insert(std::pair<void *, void *>(subscriber3, subscriber3));

    int update_num = 0;
    int total_temp = 0;
    unsigned int i = 0;
    zmq_pollitem_t astPollItems[3];

    while (true)
    {
        (void)memset(astPollItems, 0, sizeof(astPollItems));

        std::map<void *, void *>::iterator it = goSubscriberSocket.begin();
        for (i = 0; it != goSubscriberSocket.end(); ++it, i++)
        {
            astPollItems[i].socket = it->first;
            astPollItems[i].events = ZMQ_POLLIN;
        }

        long lRet = zmq_poll(astPollItems, goSubscriberSocket.size(), (long)10);
        if (lRet <= 0)
        {
            continue;
        }

        for (i = 0; i < goSubscriberSocket.size(); i++)
        {
            if (astPollItems[i].revents & ZMQ_POLLIN)
            {
                char update[20] = {0};
                ZMQ_Recv_Data(astPollItems[i].socket, sizeof(update), update);

                int zipcode = 0, temp = 0, relhumidity = 0;
                sscanf(update, "%d %d %d", &zipcode, &temp, &relhumidity);
                total_temp += temp;
                printf("subscriber Times:%d zipcode:%d temp:%d rel:%d\n", update_num, zipcode, temp, relhumidity);

                update_num++;
            }
        }

        if (update_num >= 100)
        {
            break;
        }
    }

    printf("Average temperature for zipcode:%s was %dF\n", "10001", (int)total_temp / update_num);

    zmq_close(subscriber1);
    zmq_close(subscriber2);
    zmq_close(subscriber3);
    zmq_ctx_destroy(context);

    return;
}

int main (void)
{
    pthread_t thridReqMode;
    //pthread_create(&thridReqMode, NULL, (CLIENT_THR_FUNC_CB)client_req_rsp_mode, NULL);

    pthread_t thridSublist;
    pthread_create(&thridSublist, NULL, (CLIENT_THR_FUNC_CB)client_subscriber, NULL);

    for ( ; ; )
    {
        sleep(1);
    }

    pthread_join(thridReqMode,NULL);
    pthread_join(thridSublist,NULL);

    return 0;
}

server.cpp：发布者

#include <stdio.h>
#include <string.h>
#include <iostream>
#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#include <stdlib.h>
#include <sys/prctl.h>

#include "zmq_helper.h"
#include "zmq.h"

typedef void *(*BP_THR_FUNC_CB)(void *);


void server_req_rsp_mode(void *)
{
    prctl(PR_SET_NAME, (unsigned long)"response");

    // Prepare our context and socket
    void *context = zmq_ctx_new();
    void *responder = zmq_socket(context, ZMQ_REP);
    zmq_bind(responder, "tcp://*:5555");

    while (true)
    {
        char szRecvBuf[10] = {0};
        ZMQ_Recv_Data(responder, sizeof(szRecvBuf), szRecvBuf);

        struct timeval start_time;
        gettimeofday(&start_time, NULL);
        printf("Server Received data:%s tv_sec:%ld tv_usec:%ld\n", szRecvBuf, start_time.tv_sec, start_time.tv_usec);

        sleep(1);

        // Send reply back to client
        zmq_msg_t reply;
        zmq_msg_init_data(&reply, "World", 6, NULL, NULL);

        zmq_msg_send(&reply, responder, 0);

        zmq_msg_close(&reply);
    }

    zmq_close(responder);
    zmq_ctx_destroy(context);

    return;
}

void server_publisher(void *)
{
    prctl(PR_SET_NAME, (unsigned long)"publisher");

    void *context = zmq_ctx_new();
    void *publisher = zmq_socket(context, ZMQ_PUB);

	/* 先设置高水位再bind */
    int iSocketOpt = 0;
    (void)zmq_setsockopt(publisher, ZMQ_RCVHWM, &iSocketOpt, sizeof(iSocketOpt));
    (void)zmq_setsockopt(publisher, ZMQ_SNDHWM, &iSocketOpt, sizeof(iSocketOpt));

    zmq_bind(publisher, "tcp://*:5556");
    zmq_bind(publisher, "ipc://weather.ipc");

    while (true)
    {
        int zipcode = 0, temperature = 0, relhumidity = 0;
        zipcode = rand() %1000 + 10000;
        temperature = rand() % 215 - 80;
        relhumidity = rand() % 50 + 10;

        char update[20] = {0};
        snprintf(update, sizeof(update), "%05d %04d %03d", zipcode, temperature, relhumidity);

        zmq_msg_t request;
        zmq_msg_init_data(&request, update, strlen(update), NULL, NULL);

        zmq_msg_send(&request, publisher, 0);
        zmq_msg_close(&request);
        usleep(1*1000);
    }

    zmq_close(publisher);
    zmq_ctx_destroy(context);

    return;
}

int main ()
{
    pthread_t thridReqMode;
    pthread_create(&thridReqMode, NULL, (BP_THR_FUNC_CB)server_req_rsp_mode, NULL);

    pthread_t thridPublist;
    pthread_create(&thridPublist, NULL, (BP_THR_FUNC_CB)server_publisher, NULL);

    for ( ; ; )
    {
        sleep(1);
    }

    pthread_join(thridReqMode,NULL);
    pthread_join(thridPublist,NULL);

    return 0;
}

编译命令：
g++ -g -Wall -fPIC -I/home/zhangjian/mq -L/home/zhangjian/mq -lzmq -lpthread client.cpp -o client
g++ -g -Wall -fPIC -fpermissive -I/home/zhangjian/mq -L/home/zhangjian/mq -lpthread -lzmq server.cpp -o server