zeromq集群

ZeroMQ 是一个很有个性的项目，它原来是定位为“史上最快消息队列”，所以名字里面有“MQ”两个字母，但是后来逐渐演变发展，慢慢淡化了消息队列的身影，改称为消息内核，或者消息层了。从网络通信的角度看，它处于会话层之上，应用层之下，有了它，你甚至不需要自己写一行的socket函数调用就能完成复杂的网络通信工作。

先看看这段局域网内的简单通信：

receiver.cpp

//包含zmq的头文件 
#include <zmq.h>
#include "stdio.h"

int main(int argc, char * argv[])
{
    void * pCtx = NULL;
    void * pSock = NULL;
    const char * pAddr = "tcp://*:7766";

    //创建context，zmq的socket 需要在context上进行创建 
    if((pCtx = zmq_ctx_new()) == NULL)
    {
        return 0;
    }
    //创建zmq socket ，socket目前有6中属性 ，这里使用dealer方式
    //具体使用方式请参考zmq官方文档（zmq手册） 
    if((pSock = zmq_socket(pCtx, ZMQ_DEALER)) == NULL)
    {
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    int iRcvTimeout = 5000;// millsecond
    //设置zmq的接收超时时间为5秒 
    if(zmq_setsockopt(pSock, ZMQ_RCVTIMEO, &iRcvTimeout, sizeof(iRcvTimeout)) < 0)
    {
        zmq_close(pSock);
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    //绑定地址 tcp://*:7766 
    //也就是使用tcp协议进行通信，使用网络端口 7766
    if(zmq_bind(pSock, pAddr) < 0)
    {
        zmq_close(pSock);
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    printf("bind at : %s
", pAddr);
    while(1)
    {
        char szMsg[1024] = {0};
        printf("waitting...
");
        errno = 0;
        //循环等待接收到来的消息，当超过5秒没有接到消息时，
        //zmq_recv函数返回错误信息 ，并使用zmq_strerror函数进行错误定位 
        if(zmq_recv(pSock, szMsg, sizeof(szMsg), 0) < 0)
        {
            printf("error = %s
", zmq_strerror(errno));
            continue;
        }
        printf("received message : %s
", szMsg);
    }

    return 0;
}

 sender.cpp

//包含zmq的头文件 
#include <zmq.h>
#include "stdio.h"

int main(int argc, char * argv[])
{
    void * pCtx = NULL;
    void * pSock = NULL;
    //使用tcp协议进行通信，需要连接的目标机器IP地址为192.168.1.2
    //通信使用的网络端口 为7766 
    const char * pAddr = "tcp://192.168.1.2:7766";

    //创建context 
    if((pCtx = zmq_ctx_new()) == NULL)
    {
        return 0;
    }
    //创建socket 
    if((pSock = zmq_socket(pCtx, ZMQ_DEALER)) == NULL)
    {
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    int iSndTimeout = 5000;// millsecond
    //设置接收超时 
    if(zmq_setsockopt(pSock, ZMQ_RCVTIMEO, &iSndTimeout, sizeof(iSndTimeout)) < 0)
    {
        zmq_close(pSock);
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    //连接目标IP192.168.1.2，端口7766 
    if(zmq_connect(pSock, pAddr) < 0)
    {
        zmq_close(pSock);
        zmq_ctx_destroy(pCtx);
        return 0;
    }
    //循环发送消息 
    while(1)
    {
        static int i = 0;
        char szMsg[1024] = {0};
        snprintf(szMsg, sizeof(szMsg), "hello world : %3d", i++);
        printf("Enter to send...
");
        if(zmq_send(pSock, szMsg, sizeof(szMsg), 0) < 0)
        {
            fprintf(stderr, "send message faild
");
            continue;
        }
        printf("send message : [%s] succeed
", szMsg);
        getchar();
    }

    return 0;
}

从这个实例可以看出，因为zmq支持的是对等网络，所以zmq并没有accept和listen。

一般步骤是

zmq_ctx_new()->zmq_socket()->zmq_setsockopt()->zmq_bind()->zmq_recv()

或者

zmq_ctx_new()->zmq_socket()->zmq_setsockopt()->zmq_connect()->zmq_send()

这里有个点要注要下：

ZeroMQ是无代理的，其有高速数据传输能力，对IPC和socket有良好的封装特性。而且bind程序和 connect程序无论谁先启动，其实都不影响整个系统的正常运行。

zeromq模式

zeromq 有多种模式，常用的有三种:请求应答模式、订阅发布模式、push pull模式 。

1. 请求应答模式（req 和 rep）

消息双向的，有来有往，req端请求的消息，rep端必须答复给req端

2. 订阅发布模式 （sub 和 pub）

消息单向的，有去无回的。可按照发布端可发布制定主题的消息，订阅端可订阅喜欢的主题，订阅端只会收到自己已经订阅的主题。发布端发布一条消息，可被多个订阅端同时收到。

3. push pull模式

消息单向的，也是有去无回的。push的任何一个消息，始终只会有一个pull端收到消息.

后续的代理模式和路由模式等都是在三种基本模式上面的扩展或变异。

阻塞 和 非阻塞

以上三种基本模式都支持阻塞模式和非阻塞模式。req 和 rep的阻塞模式是这样的（其实跟原生的socket实现也非常像）。大家用过socket的，客户端要是先启动的话，会连接失败，或者是短时间内有超时问题。

参考：

zeromq实例

zeromq实现服务器间高性能通信