2. 程式人生 > >Linux下使用ZMQ實踐“請求-響應”非同步伺服器模型

Linux下使用ZMQ實踐“請求-響應”非同步伺服器模型

阿新 • • 發佈：2019-02-19

一、背景

    上一篇文章《Linux下使用ZMQ實踐“請求-響應”伺服器模型》中使用的是REP-REQ套件，該套件的特點是必須一個請求對應一個響應，如果在應用中不想使用同步處理的方式呢？ZMQ有沒有提供非同步處理的方法？答案是使用DEALER-ROUTER套件。

    另外如何在多執行緒中安全傳遞訊息的方法可以參見《Linux下使用ZMQ實踐“生產者-消費者”模型》ZMQ_PULL、ZMQ_PUSH的實踐。

二、相關知識

1、ZMQ_DEALER

       ZMQ_DEALER
           A socket of type ZMQ_DEALER is an advanced pattern used for extending request/reply sockets.
           Each message sent is round-robined among all connected peers, and each message received is
           fair-queued from all connected peers.

           When a ZMQ_DEALER socket enters the mute state due to having reached the high water mark for
           all peers, or if there are no peers at all, then any zmq_send(3) operations on the socket
           shall block until the mute state ends or at least one peer becomes available for sending;
           messages are not discarded.

           When a ZMQ_DEALER socket is connected to a ZMQ_REP socket each message sent must consist of
           an empty message part, the delimiter, followed by one or more body parts.

    DEALER是一種用於請求/答應模式的更高階的擴充套件Socket，它可以自由的收發訊息，沒有ZMQ_REP/ZMQ_REQ那樣的限制。對於每一個連線，接收訊息也是使用了公平佇列，傳送使用了迴圈佇列（RR）。

    ZMQ_DEALER受ZMQ_RCVHW和ZMQ_SNDHW兩個閥值影響（可通過zmq_setsockopt函式設定），一旦傳送或接收佇列達到閥值，Socket就會進入mute狀態，此時對DEALER的任何zmq_send操作都會阻塞，直到mute狀態結束。如果當前沒有有效的連線，zmq_send操作也會阻塞，直到有新的連線到來為止。DEALER發生阻塞並不會丟棄訊息。

    注意：如果ZMQ_DEALER連線到ZMQ_REP，每一個訊息包必須包含一個空幀，然後再緊跟著資料包體。

2、ZMQ_ROUTER

    ZMQ_ROUTER           
           A socket of type ZMQ_ROUTER is an advanced socket type used for extending request/reply
           sockets. When receiving messages a ZMQ_ROUTER socket shall prepend a message part containing
           the routing id of the originating peer to the message before passing it to the application.
           Messages received are fair-queued from among all connected peers. When sending messages a
           ZMQ_ROUTER socket shall remove the first part of the message and use it to determine the
           routing id of the peer the message shall be routed to. If the peer does not exist anymore,
           or has never existed, the message shall be silently discarded. However, if
           ZMQ_ROUTER_MANDATORY socket option is set to 1, the socket shall fail with EHOSTUNREACH in
           both cases.

    ZMQ_ROUTER是一種用於請求/答應模式的更高階的擴充套件Socket，它可以自由的收發訊息。當ZMQ_ROUTER接收到訊息時，會自動在訊息頂部加入來源地址識別符號，接收訊息使用了公平佇列。

    

    當傳送訊息時，ZMQ_ROUTER又會自動去掉這個識別符號，並且根據這個識別符號路由到相應的端點。

    如果此地址標識的端點不存在，預設會毫無徵兆的丟棄訊息，除非將ZMQ_ROUTER_MANDATORY 選項設定為1。

           When a ZMQ_ROUTER socket enters the mute state due to having reached the high water mark for
           all peers, then any messages sent to the socket shall be dropped until the mute state ends.
           Likewise, any messages routed to a peer for which the individual high water mark has been
           reached shall also be dropped. If, ZMQ_ROUTER_MANDATORY is set to 1, the socket shall block
           or return EAGAIN in both cases.
    當佇列達到閥值時，ZMQ_ROUTER Socket就會進入mute狀態，此時所有後續傳送到此Socket的訊息都會被丟棄，直到mute狀態結束。同樣的，如果對端的接收佇列達到了閥值，訊息也會被丟棄。但是如果設定了ZMQ_ROUTER_MANDATORY 選項，訊息不會丟棄，介面將等待發送完成後返回。

三、實踐

    程式碼基於REQ-REP版本進行修改，讀者可以通過ROUTER_DEALER巨集來看出與原來程式碼的區別；

    服務端主要是配合客戶端實現，程式碼修改不大，仍然保持著一次請求、一次響應的功能；

    需要注意的地方就是Router收到Dealer的訊息是包含標識頭部的（Dealer來源標識），傳送的時候需要使用sendmore把標識先發出去。

int main(int argc, char *argv[])
{
        void *ctx = zmq_ctx_new();
        server_master(ctx);
        zmq_ctx_destroy(ctx);
        exit(EXIT_SUCCESS);
}

static void *server_master(void *ctx)
{
        int ret = 0;
        char id[16] = {0};
        char request[1024];
        char respone[1024];

#ifdef DEALER_ROUTER
        void *server = zmq_socket(ctx, ZMQ_ROUTER);
#else
        void *server = zmq_socket(ctx, ZMQ_REP);
#endif

        s_set_id_ex(server, id, sizeof(id));
        zmq_bind(server, "ipc://server.ipc");
        zmq_pollitem_t items[] = {
                { server, 0, ZMQ_POLLIN, 0 },
        };

        LOGN("Server %s start\n", id);
        while (1) {
                ret = zmq_poll(items, 1 /* size */, 1000 /* ms */);
                assert(ret >= 0);

        if (items[0].revents & ZMQ_POLLIN) {
#ifdef DEALER_ROUTER
                        char peer[16] = {0};
                        s_recv(server, peer);
#endif
                        s_recv(server, request);
                        LOGN("Server %s recv: %s\n", id, request);

                        //TODO something handle
                        sleep(1);

#ifdef DEALER_ROUTER
                        s_sendmore(server, peer);
#endif
                        snprintf(respone, sizeof(respone), "%s-World", request);
                        s_send(server, respone);
                        LOGN("Server %s send: %s\n", id, respone);
                }
        }

        LOGN("Server %s Finish\n", id);
        zmq_close(server);
}

    客戶端的程式碼修改的比較多，主要是使用主執行緒推送請求，子執行緒中對請求進行響應處理，來達到非同步的目的（或者另一種實現就是在一個執行緒中zmq_poll統一處理）

int main(int argc, char *argv[])
{
        void *ctx = zmq_ctx_new();
        srandom(time(NULL));
#ifdef DEALER_ROUTER
        void *thread = zmq_threadstart(client_worker, ctx);
        client_master(ctx);
        zmq_threadclose(thread);
#else
        client_task(ctx);
#endif
        zmq_ctx_destroy(ctx);
        exit(EXIT_SUCCESS);
}
void client_master(void *ctx)
{
        int ix;
        int roll = randof(1000);
        char request[1024];
        void *pusher = zmq_socket(ctx, ZMQ_PUSH);

        zmq_connect(pusher, "inproc://client.inproc");

        for (ix = 0; ix < TEST_TIMES; ix++) {
                snprintf(request, sizeof(request), "Request-%03d-%03d", roll, ix);
                s_send(pusher, request);
        }

        zmq_close(pusher);
}

主執行緒通過 ZMQ_PUSH 與子執行緒的 ZMQ_PULL 進行對接，子執行緒再轉送給 ZMQ_DEALER 傳送出去；

由於子執行緒需要同時監聽兩個socket的收事件，所以使用了 zmq_poll 進行IO複用；

void client_worker(void *ctx)
{
        int ret = 0;
        int cnt = 0;
        char id[16] = {0};
        char request[1024];
        char respone[1024];
        void *puller = zmq_socket(ctx, ZMQ_PULL);
        void *dealer = zmq_socket(ctx, ZMQ_DEALER);
        
        s_set_id_ex(dealer, id, sizeof(id));
        zmq_connect(dealer, "ipc://server.ipc");
        zmq_bind(puller, "inproc://client.inproc");
        zmq_pollitem_t items[] = {
                { puller, 0, ZMQ_POLLIN, 0 },
                { dealer, 0, ZMQ_POLLIN, 0 }
        };
        LOGN("Client %s start\n", id);
        while (cnt < TEST_TIMES) {
                ret = zmq_poll(items, 2 /* size */, 1000 /* ms */);
                assert(ret >= 0);
        if (items[0].revents & ZMQ_POLLIN) {
                        s_recv(puller, request);
                        s_sendmore(puller, ""); // dealer
                        s_send(dealer, request);
                        LOGN("Client %s send: %s\n", id, request);
                }
        if (items[1].revents & ZMQ_POLLIN) {
                        s_recv(dealer, respone);
                        cnt++;
                        LOGN("Client %s recv: %s\n", id, respone);
                        //TODO something handle
                }
        }
        LOGN("Client %s finish\n", id);
        zmq_close(puller);
        zmq_close(dealer);
}

開啟兩個client（00FA與0064）、一個server，執行結果如下：

[ 1520703616.843 ]: Client 00FA start
[ 1520703616.844 ]: Client 00FA send: Request-251-000
[ 1520703616.844 ]: Client 00FA send: Request-251-001
[ 1520703616.844 ]: Client 00FA send: Request-251-002
[ 1520703616.844 ]: Client 00FA send: Request-251-003
[ 1520703616.844 ]: Client 00FA send: Request-251-004
[ 1520703616.844 ]: Client 00FA send: Request-251-005
[ 1520703616.844 ]: Client 00FA send: Request-251-006
[ 1520703616.844 ]: Client 00FA send: Request-251-007
[ 1520703616.844 ]: Client 00FA send: Request-251-008
[ 1520703616.844 ]: Client 00FA send: Request-251-009
[ 1520703622.523 ]: Client 00FA recv: Request-251-000-World
[ 1520703624.523 ]: Client 00FA recv: Request-251-001-World
[ 1520703626.525 ]: Client 00FA recv: Request-251-002-World
[ 1520703628.527 ]: Client 00FA recv: Request-251-003-World
[ 1520703630.528 ]: Client 00FA recv: Request-251-004-World
[ 1520703632.531 ]: Client 00FA recv: Request-251-005-World
[ 1520703634.534 ]: Client 00FA recv: Request-251-006-World
[ 1520703636.535 ]: Client 00FA recv: Request-251-007-World
[ 1520703638.544 ]: Client 00FA recv: Request-251-008-World
[ 1520703639.541 ]: Client 00FA recv: Request-251-009-World
[ 1520703639.541 ]: Client 00FA finish
[ 1520703617.217 ]: Client 0064 start
[ 1520703617.217 ]: Client 0064 send: Request-609-000
[ 1520703617.217 ]: Client 0064 send: Request-609-001
[ 1520703617.217 ]: Client 0064 send: Request-609-002
[ 1520703617.217 ]: Client 0064 send: Request-609-003
[ 1520703617.217 ]: Client 0064 send: Request-609-004
[ 1520703617.217 ]: Client 0064 send: Request-609-005
[ 1520703617.217 ]: Client 0064 send: Request-609-006
[ 1520703617.217 ]: Client 0064 send: Request-609-007
[ 1520703617.217 ]: Client 0064 send: Request-609-008
[ 1520703617.217 ]: Client 0064 send: Request-609-009
[ 1520703620.521 ]: Client 0064 recv: Request-609-000-World
[ 1520703621.521 ]: Client 0064 recv: Request-609-001-World
[ 1520703623.522 ]: Client 0064 recv: Request-609-002-World
[ 1520703625.524 ]: Client 0064 recv: Request-609-003-World
[ 1520703627.526 ]: Client 0064 recv: Request-609-004-World
[ 1520703629.527 ]: Client 0064 recv: Request-609-005-World
[ 1520703631.529 ]: Client 0064 recv: Request-609-006-World
[ 1520703633.531 ]: Client 0064 recv: Request-609-007-World
[ 1520703635.534 ]: Client 0064 recv: Request-609-008-World
[ 1520703637.537 ]: Client 0064 recv: Request-609-009-World
[ 1520703637.537 ]: Client 0064 finish
[ 1520703619.450 ]: Server 00C8 start
[ 1520703619.519 ]: Server 00C8 recv: Request-609-000
[ 1520703620.521 ]: Server 00C8 send: Request-609-000-World
[ 1520703620.521 ]: Server 00C8 recv: Request-609-001
[ 1520703621.521 ]: Server 00C8 send: Request-609-001-World
[ 1520703621.521 ]: Server 00C8 recv: Request-251-000
[ 1520703622.522 ]: Server 00C8 send: Request-251-000-World
[ 1520703622.522 ]: Server 00C8 recv: Request-609-002
[ 1520703623.522 ]: Server 00C8 send: Request-609-002-World
[ 1520703623.523 ]: Server 00C8 recv: Request-251-001
[ 1520703624.523 ]: Server 00C8 send: Request-251-001-World
[ 1520703624.524 ]: Server 00C8 recv: Request-609-003
[ 1520703625.524 ]: Server 00C8 send: Request-609-003-World
[ 1520703625.524 ]: Server 00C8 recv: Request-251-002
[ 1520703626.525 ]: Server 00C8 send: Request-251-002-World
[ 1520703626.525 ]: Server 00C8 recv: Request-609-004
[ 1520703627.526 ]: Server 00C8 send: Request-609-004-World
[ 1520703627.526 ]: Server 00C8 recv: Request-251-003
[ 1520703628.527 ]: Server 00C8 send: Request-251-003-World
[ 1520703628.527 ]: Server 00C8 recv: Request-609-005
[ 1520703629.527 ]: Server 00C8 send: Request-609-005-World
[ 1520703629.528 ]: Server 00C8 recv: Request-251-004
[ 1520703630.528 ]: Server 00C8 send: Request-251-004-World
[ 1520703630.529 ]: Server 00C8 recv: Request-609-006
[ 1520703631.529 ]: Server 00C8 send: Request-609-006-World
[ 1520703631.530 ]: Server 00C8 recv: Request-251-005
[ 1520703632.530 ]: Server 00C8 send: Request-251-005-World
[ 1520703632.531 ]: Server 00C8 recv: Request-609-007
[ 1520703633.531 ]: Server 00C8 send: Request-609-007-World
[ 1520703633.532 ]: Server 00C8 recv: Request-251-006
[ 1520703634.532 ]: Server 00C8 send: Request-251-006-World
[ 1520703634.533 ]: Server 00C8 recv: Request-609-008
[ 1520703635.533 ]: Server 00C8 send: Request-609-008-World
[ 1520703635.534 ]: Server 00C8 recv: Request-251-007
[ 1520703636.535 ]: Server 00C8 send: Request-251-007-World
[ 1520703636.536 ]: Server 00C8 recv: Request-609-009
[ 1520703637.537 ]: Server 00C8 send: Request-609-009-World
[ 1520703637.538 ]: Server 00C8 recv: Request-251-008
[ 1520703638.541 ]: Server 00C8 send: Request-251-008-World
[ 1520703638.541 ]: Server 00C8 recv: Request-251-009
[ 1520703639.541 ]: Server 00C8 send: Request-251-009-World

    從客戶端資訊來看，10個請求快速地就傳送出去了，然後再下來的20秒內，大約每2秒能夠獲取到一次響應；

    從服務端資訊上來看，對於兩個客戶端的處理基本是輪流地處理的；

四、總結

    應用可以通過 ROUTER-DEALER套件來實現非同步的“請求-響應" 處理，在訊息的處理過程中只要注意空幀的傳送、標識頭的處理就行；


參考文章：

[1] http://zguide.zeromq.org/page:all

    ZMQ_DEALER受ZMQ_RCVHW和ZMQ_SNDHW兩個閥值影響（可通過zmq_setsockopt函式設定），一旦傳送或接收佇列達到閥值，Socket就會進入mute狀態，此時對DEALER的任何zmq_send操作都會阻塞，直到mute狀態結束。如果當前沒有有效的連線，zmq_send操作也會阻塞，直到有新的連線到來為止。DEALER發生阻塞並不會丟棄訊息。

相關推薦

Linux下使用ZMQ實踐“請求-響應”非同步伺服器模型

一、背景    上一篇文章《Linux下使用ZMQ實踐“請求-響應”伺服器模型》中使用的是REP-REQ套件，該套件的特點是必須一個請求對應一個響應，如果在應用中不想使用同步處理的方式呢？ZMQ有沒有提供非同步處理的方法？答案是使用DEALER-ROUTER套件。    另外

linux下搭建我的世界spongeforge 伺服器 (海綿端)

以下我用的都是1.10.2版本 且以下用的伺服器連線管理軟體有WinSCP、Xshell 5   首先，去下載一個MC1.10.2的純淨服務端，這個不會很難，百度一下！ 比如我下的檔案就叫minecraft_server.1.10.2.jar 然後我們在自己linux系統下的root資料夾

在Linux下搭建我的世界(Minecraft)伺服器

最近薅了百度雲雙12的羊毛，1核2G一年150。突然想起以前大學整個宿舍通宵開黑挖泥土的歲月，所以剛好趁著這臺伺服器，打算自己搭建一個我的世界伺服器，重溫一下以前的感覺。 系統要求 google到的MC系統要求： 1.CPU：雙核 2.記憶體：20-40人2GB，30-60人3GB，60

Linux下兩種方式安裝apache伺服器踩過的坑

記得第一次找工作面試時，面試官問：你如何理解apache？對於一個只在tomcat上開發剛入行的程式設計師，我只記得apache是一個開源組織，下面掌管著各種高大上，聽都沒聽過的專案，面試完後我才知道他想問的是apache伺服器，而且後面肯定還有和apache伺服器相關的提問，就這樣被我

Linux下構建主、從域名伺服器

實驗背景：       小諾公司目前搭建了郵件伺服器、web伺服器，現在需要使用Linux搭建一臺DNS伺服器為上面的服務提供域名解析，考慮到DNS伺服器的安全性，還需要搭建一臺從DNS伺服器。

Linux使用curl檢視請求響應時間

1.curl 檢視web站點 curl -o /dev/test -s -w %{time_namelookup}::%{time_connect}::%{time_starttransfer}::%

Linux下使用Apache佈署視訊伺服器

1. 安裝apache2。 sudo apt-get install apache2 安裝地址為/etc/apache2 2. 更改預設埠。 將sites-enabled/000-default.conf和ports.conf的80埠改為希望的埠，如8090。 3. 配置視

linux下實現ssh自動登入遠端伺服器

前言：開發工作中，由於有時需要通過ssh登入遠端伺服器上去，每次都要輸入密碼，密碼很難記，感覺很麻煩，於是上網搜了通過 ssh自動登入遠端伺服器上去，從而免去了每次都要輸入密碼的困擾。 系統說明：本地機器：Redhat遠端機器：Redhat 步驟： A.本地機器需要做的修

linux下定時自動重啟tomcat伺服器

我們在Linux系統上的tomcat部署了專案之後，可能因為專案存在某些bug等原因會導致過段時間就掛掉，作為臨時解決方案，我們需要間隔一段時間之後就重啟一次伺服器，但是人為重啟又太low太費力，所以自動重啟就略顯高大上牛逼霸氣了~ 下面開始介紹步驟： 1.編輯一個文字檔

linux下rsync+inotify實現兩臺伺服器檔案實時同步

假設兩個伺服器： 192.168.0.1 源伺服器  有目錄 /opt/test/ 192.168.0.2 目標伺服器  有目錄 /opt/bak/test/ 實現的目的就是保持這兩個伺服器某個檔案目錄保持實時同步 實現方式： 通過rsync+inotify-too

Linux下配置完整安全的DHCP伺服器詳解

DHCP是動態主機配置協議。這個協議用於向計算機自動提供IP地址,子網掩碼和路由資訊。網路管理員通常會分配某個範圍的IP地址來分發給區域網上的客戶機。當裝置接入這個區域網時，它們會向 DHCP 伺服器請求一個 IP 地址。然後DHCP伺服器為每個請求的裝置分配一個地址，直到分

解決Linux下Svn檢出Windows SVN伺服器上專案SSL handshake failed: SSL error: Key usage violation in certificate ha

在Linux上檢出windows SVN伺服器上專案時出現了SSL handshake failed: SSL error: Key usage violation in certificate has been detected.的錯誤。 最後通過從網上檢索找到了一個

Linux下配置NTP 架設本地時間伺服器

Network Time Protocol (NTP) 也是RHCE新增的考試要求. 學習的時候也順便複習了一下如何設定Linux的時間,現在拿出來和大家分享 設定NTP伺服器不難但是NTP本身是一個很複雜的協議. 這裡只是簡要地介紹一下實踐方法 和上次一樣,下面的實驗都

Linux下搭建實現HttpRunnerManager的非同步執行、定時任務及任務監控

前言 在之前搭建的HttpRunnerManager介面測試平臺，我們還有一些功能沒有實現，比如非同步執行、定時任務、任務監控等，要完成非同步執行，需要搭建 RabbitMQ 等環境，今天我們就來實現這些功能。 需要在Linux上提前準備的環境（下面是本人搭建時的環境）： 1，HttpRunnerManag

linux下基於vc088x開發板分析CLK模型（時鐘管理）

Linux clk 模型   Linux clk模型採用面向物件的思想來設計實現的。 在porting層建立一個一個的clk節點物件，然後將所有的clk節點物件連成一個list。 當驅動層需要設定時鐘的時候，通過porting層與驅動層

LINUX 下配置郵件伺服器過程（實踐）

配置 sendmail.mc ： 開啟/etc/mail/sendmail.mc檔案， 找到： dnl TRUST_AUTH_MECH(`EXTERNAL DIGEST-MD5 CRAM-MD5 LOGIN PLAIN')dnl dnl define(`confAUTH_MECHANISMS', `EXTE

linux下用多執行緒實現socket伺服器和客戶端的非同步通訊

前面介紹了用select函式來實現socket的非同步收發資料，但是select函式也有一些缺陷，要使socket能持續地通訊，select必須不停地檢測，這樣程序就會一直阻塞在這裡，限制了功能的擴充套件，這裡我們用多執行緒的方式，另建立兩個執行緒用來發送/接收

Linux下的文件夾創建命令使用實踐

c程序 聲明 可見 strrchr 創建 二次 所在 begin false 【文章摘要】 本文以實際的C源程序為樣例，介紹了Linux下的文件夾創建命令(mkdir)的用法。為相關開發工作的開展提供了故意的參考。【關鍵詞】 C語言 Linux

Linux下UDP發送大量請求導致Operation not permitted的問題探討

通過 微服務 com 發現 又是 get 端口 www 模式 一、問題背景目前公司準備上微服務的架構模式，在對比大量的API網關之後，最後選中了Kong作為我們的API網關，在經過大量的研究之後，啊呸，踩坑之後，終於跑起來了，簡直是普天同慶，撒花祝賀。 但是在美滋滋的進行性

Http請求格式（在Linux下使用telnet親測，通過這篇我才明白）

語法 設置 rdquo 我們 令行 ati 訪問性 idl lib 命令行窗口中用telnet測試HTTP協議請求消息格式響應消息格式1. 命令行窗口中用telnet測試HTTP協議 HTTP消息是由普通ASCII文本組成。消息包括消息頭和數據體部分。消息頭以行為單位，每行