C/C++编程：ZeroMQ示例学习

旧城等待， 2022-11-07 01:45 306阅读 0赞

> 获取例子：`git clone --depth=1 https://github.com/imatix/zguide.git`

# 获取版本 #

ZeroMQ确实有多个版本，并且经常出现问题，如果您遇到问题，它将在更高版本中得到修复。因此，准确地知道您实际链接的ZeroMQ版本是一个有用的技巧。

## C ##

#include <zmq.h>
    
    int main (void)
    {
        
        int major, minor, patch;
        zmq_version (&major, &minor, &patch);
        printf ("Current 0MQ version is %d.%d.%d\n", major, minor, patch);
        return 0;
    }

# 请求-答复模式 #

## （Hello World CS） ##

效果：客户端向服务器发送“ Hello”，服务器回复“ World”  
![在这里插入图片描述][20210310165308106.png]

### 服务端 ###

效果：在端口12345上打开ZeroMQ套接字，读取该端口上的请求，并对每个请求回复“World”

#include <zmq.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <assert.h>
    
    int main (void)
    {
        
        //  Socket to talk to clients
        void *context = zmq_ctx_new ();
        void *responder = zmq_socket (context, ZMQ_REP);
        int rc = zmq_bind (responder, "tcp://*:5555");
        assert (rc == 0);
    
        while (1) {
         
            char buffer [10];
            zmq_recv (responder, buffer, 10, 0);
            printf ("Received Hello\n");
            sleep (1);          //  Do some 'work'
            zmq_send (responder, "World", 5, 0);
        }
        return 0;
    }

### 客户端 ###

//  Hello World client
    #include <zmq.h>
    #include <string.h>
    #include <stdio.h>
    #include <unistd.h>
    
    int main (void)
    {
        
        printf ("Connecting to hello world server...\n");
        void *context = zmq_ctx_new ();
        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++) {
        
            char buffer [10];
            printf ("Sending Hello %d...\n", request_nbr);
            zmq_send (requester, "Hello", 5, 0);
            zmq_recv (requester, buffer, 10, 0);
            printf ("Received World %d\n", request_nbr);
        }
        zmq_close (requester);
        zmq_ctx_destroy (context);
        return 0;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]

## 文件传输 ##

1、服务器

#include <zmq.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <assert.h>
    #include <cstring>
    #include <malloc.h>
    #include <sstream>
    #include <vector>
    
    #define BUFSIZE 262144  //256k
    using namespace std;
    //ostringstream对象用来进行格式化的输出，常用于将各种类型转换为string类型
    //ostringstream只支持<<操作符
    template<typename T> string toString(const T& t){
        
        ostringstream oss;  //创建一个格式化输出流
        oss<<t;             //把值传递如流中
        return oss.str();
    }
    
    void SplitString(const std::string& s, std::vector<std::string>& v, const std::string& c)
    {
        
        std::string::size_type pos1, pos2;
        pos2 = s.find(c);
        pos1 = 0;
        while(std::string::npos != pos2)
        {
        
            v.push_back(s.substr(pos1, pos2-pos1));
    
            pos1 = pos2 + c.size();
            pos2 = s.find(c, pos1);
        }
        if(pos1 != s.length())
            v.push_back(s.substr(pos1));
    }
    
    
    int main (void)
    {
        
        //  Socket to talk to clients
        void *context = zmq_ctx_new ();
        void *socket = zmq_socket (context, ZMQ_REP);
        int rc = zmq_bind (socket, "tcp://*:5555");
        assert (rc == 0);
        char *buf = (char *)malloc(BUFSIZE);//申请一个缓冲区，存放接收到的文件内容
    
        memset(buf, 0, BUFSIZE);
    
        rc = zmq_recv (socket, buf, BUFSIZE, 0);
        if (rc == -1){
        
            printf("sorry, zmq_recv failed %s\n",  zmq_strerror(errno));
            return 0;
        }
        printf ("Received  %s\n", buf);
    
    
        FILE *fd;
        int total = 0;
        std::string rs ;
        std::vector<std::string> v;
        SplitString(buf, v, "_");
        if(v.size() != 2){
        
            memset(buf, 0, BUFSIZE);
            strcpy(buf, "name_total");
        }else{
        
            total = atoi(v[1].c_str());
            fd = fopen(v[0].c_str(), "wb");//以只写方式打开文件
            if (fd == NULL){
        
                memset(buf, 0, BUFSIZE);
                snprintf(buf, BUFSIZ, "open %s failed %s\n",  v[0].c_str(), strerror(errno));
                printf("open %s failed %s\n", v[0].c_str(), strerror(errno));
            }else{
        
                memset(buf, 0, BUFSIZE);
                strcpy(buf, "ok");
            }
        }
    
    
        zmq_send (socket, buf, strlen(buf), 0);
    
    
    
        while (true) {
        
            memset(buf, 0, BUFSIZE);
            rc = zmq_recv (socket, buf, BUFSIZE, 0);
            if (rc == -1){
        
                printf("sorry, zmq_recv failed %s\n",  zmq_strerror(errno));
                break;
            }else{
        
                printf("ok = %d\n", rc);
            }
    
            fwrite(buf, 1, rc, fd);//将从client端收到的内容写入文件
            total = total - rc;
            if(total == 0){
        
                printf("finished");
                fclose(fd);//关闭打开的文件
            }
    
    
            memset(buf, 0, BUFSIZE);
            snprintf(buf, BUFSIZ, "%d\n",  rc);
            zmq_send (socket, buf, strlen(buf), 0);
        }
    
        free(buf);
        zmq_close(socket);
        zmq_ctx_destroy(context);
        return 0;
    }

2、客户端

//  Hello World client
    #include <zmq.h>
    #include <string.h>
    #include <stdio.h>
    #include <unistd.h>
    #include "File.h"
    #define BUFSIZE 262144  //256k
    void getfilename(const char *filename, char *name)//从完整路径名中解析出文件名称，例如：/home/test/abc.txt,解析完成后为abc.txt
    {
        
        int len = strlen(filename);
        int i;
        for (i = (len - 1); i >= 0; i--)
        {
        
            if ((filename[i] == '\\') || (filename[i] == '/'))
            {
        
                break;
            }
        }
        strcpy(name, &filename[i + 1]);
        return;
    }
    
    int main (void)
    {
        
        printf ("Connecting to hello world server...\n");
        void *context = zmq_ctx_new ();
        void *socket = zmq_socket (context, ZMQ_REQ);
        zmq_connect (socket, "tcp://192.168.0.12:5555");
    
        const char * filename = "/home/oceanstar/CLionProjects/acl.tar.gz";
        FILE *file = fopen(filename, "rb");//以只读方式打开filename指定的文件
        if (file == NULL)//如果文件打开失败，函数返回
        {
        
            printf("open %s failed %s\n", filename, strerror(errno));
            return 0;
        }
    
        char *buf = (char *)malloc(BUFSIZE);//申请一个缓冲区，存放接收到的文件内容
    
    
        char name[256] ;
        getfilename(filename, name);
    
        rewind(file);
        fseek(file,0L,SEEK_END);
        long length=ftell(file);
        rewind(file);
    
    
        printf("%s has [%ld] characters...", name,  length);
    
    
        memset(buf, 0, BUFSIZE);
        snprintf(buf, BUFSIZ, "%s_%ld\n",  name, length);
    
    
        int rc;
        rc = zmq_send(socket, buf, strlen(buf), 0);
        if (rc == -1){
        
            printf("send filename failed %s\n",  zmq_strerror(errno));
        }
    
        rc = zmq_recv (socket, buf, BUFSIZE, 0);
        if (rc == -1){
        
            printf("sorry, zmq_recv ok failed %s\n",  zmq_strerror(errno));
            return 0;
        }
    
        if (strncmp(buf, "ok", 2) != 0){
        
            printf("sorry, server error %s\n",  zmq_strerror(errno));
            return 0;
        }
    
    
    
    
    
    
        while (true){
        
            memset(buf, 0, BUFSIZE);
            rc = fread(buf, 1, BUFSIZE, file);
            if (rc <= 0)
            {
        
                if (rc < 0)
                    printf("fread failed %s\n", strerror(errno));
                break;
            }
    
            printf("file = %d\t", rc);
            rc = zmq_send(socket, buf, rc, 0);
            if (rc == -1){
        
                printf("send failed %s\n",  zmq_strerror(errno));
            }else{
        
                printf("ok = %d\t\t", rc);
            }
    
            memset(buf, 0, BUFSIZE);
            rc = zmq_recv (socket, buf, BUFSIZE, 0);
            printf("%d, %s\n", rc, buf);
        }
    
    
        fclose(file);
    
        zmq_close (socket);
        zmq_ctx_destroy (context);
        return 0;
    }

# 发布-订阅模式(天气 CS) #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]  
效果：

*  服务器先启动：
    
     *  服务器会一直运行(while)，这个期间只要有客户端连上来了，服务器就会主动往客户端发送消息。当客户端接收完消息，客户端会停止
 *  客户端先启动：
    
     *  客户端会一直阻塞，直到接收到服务器发送的消息，才会停止

## C ##

### 服务器 ###

#include <zmq.h>
    
    #include <assert.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <time.h>
    
    //  Provide random number from 0..(num-1)
    #define randof(num)  (int) ((float) (num) * random () / (RAND_MAX + 1.0))
    
    //  Convert C string to 0MQ string and send to socket
    static int s_send (void *socket, char *string) {
        
        int size = zmq_send (socket, string, strlen (string), 0);
        return size;
    }
    
    int main (void)
    {
        
        //  Prepare our context and publisher
        void *context = zmq_ctx_new ();
        void *publisher = zmq_socket (context, ZMQ_PUB);
        int rc = zmq_bind (publisher, "tcp://*:5556");
        assert (rc == 0);
    
        //  Initialize random number generator
        srandom ((unsigned) time (NULL));
        while (1) {
        
            //  Get values that will fool the boss
            int zipcode, temperature, relhumidity;
            zipcode     = randof (100000);
            temperature = randof (215) - 80;
            relhumidity = randof (50) + 10;
    
            //  Send message to all subscribers
            char update [20];
            sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);
            s_send (publisher, update);
        }
        zmq_close (publisher);
        zmq_ctx_destroy (context);
        return 0;
    }

### 客户端 ###

#include <zmq.h>
    #include <assert.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    
    
    //  Receive 0MQ string from socket and convert into C string
    //  Caller must free returned string. Returns NULL if the context
    //  is being terminated.
    static char *s_recv (void *socket) {
        
        enum {
         cap = 256 };
        char buffer [cap];
        int size = zmq_recv (socket, buffer, cap - 1, 0);
        if (size == -1)
            return NULL;
        buffer[size < cap ? size : cap - 1] = '\0';
    
    #if (defined (WIN32))
        return strdup (buffer);
    #else
        return strndup (buffer, sizeof(buffer) - 1);
    #endif
    
        // remember that the strdup family of functions use malloc/alloc for space for the new string.  It must be manually
        // freed when you are done with it.  Failure to do so will allow a heap attack.
    }
    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);
    
        //  Subscribe to zipcode, default is NYC, 10001
        const char *filter = (argc > 1)? argv [1]: "10001 ";
        rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,
                             filter, strlen (filter));
        assert (rc == 0);
    
        //  Process 100 updates
        int update_nbr;
        long total_temp = 0;
        for (update_nbr = 0; update_nbr < 100; update_nbr++) {
        
            char *string = s_recv (subscriber);
    
            int zipcode, temperature, relhumidity;
            sscanf (string, "%d %d %d",
                    &zipcode, &temperature, &relhumidity);
            total_temp += temperature;
            free (string);
        }
        printf ("Average temperature for zipcode '%s' was %dF\n",
                filter, (int) (total_temp / update_nbr));
    
        zmq_close (subscriber);
        zmq_ctx_destroy (context);
        return 0;
    }

# 在字符串上的小注解 #

除字节大小之外，ZMQ对你发送的数据一无所知。这意味着你有责任安全的格式化数据，使得应用程序可以读取它

*  在C语言和某些其他语言中，字符串以空字节终止。我们可以发送一个像“ HELLO”这样的字符串，并带有一个额外的空字节：

zmq_send (requester, "Hello", 6, 0);

*  但是，如果您使用其他语言发送字符串，则该字符串可能不会包含该空字节。例如，当我们在Python中发送相同的字符串时，我们将执行以下操作：

socket.send ("Hello")

*  如果您从C程序中读取此内容，可能会得到一个错误的结果。也就是说，当您的客户端和服务器在字符串格式上不一致时，您将得到奇怪的结果。
 *  因此，让我们建立一个规则，即**ZeroMQ字符串是长度指定的，并且在发送时不带尾随null**。在最简单的情况下（我们将在示例中进行此操作），ZeroMQ字符串整齐地映射到ZeroMQ消息帧，它看起来像下图所示—一个长度和一个字节  
    ![在这里插入图片描述][20210310170846976.png]  
    在[zhelpers.h][]，封装了很多有用辅助函数，比如下面函数可以帮助接收ZeroMQ字符串并将其作为有效的C字符串传递给应用程序

#include <zmq.h>
    #include <memory.h>
    
    //从套接字接收ZeroMQ字符串并转换为C字符串
    //如果字符串长度较长，则将字符串截断为255个字符
    static char *s_recv (void *socket) {
        
        char buffer [256];
        int size = zmq_recv (socket, buffer, 255, 0);
        if (size == -1)
            return NULL;
        if (size > 255)
            size = 255;
        buffer [size] = '\0';
        /* 在*nix中使用strndup(buffer, sizeof(buffer)-1)*/
        return strdup (buffer);
    }

[20210310165308106.png]: /images/20221023/2bc7e09eb86b48e0af1a255e3a7499dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]: /images/20221023/87b499cf0f15480cbe6fd72e701f6cf5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]: /images/20221023/7381475797e14eda80b495b2b860625a.png
[20210310170846976.png]: /images/20221023/fc606d80cd61466aad562e5d3b964128.png
[zhelpers.h]: https://github.com/booksbyus/zguide/blob/master/examples/C/zhelpers.h