ZMQ接收端重复读取消息：如何在接收后清空队列？

你遇到的无限重复读取同一消息的问题，本质是ZMQ REQ-REP模式的强制协议规则导致的：

• REQ套接字发送消息后，必须等待REP套接字的应答；如果未收到应答，ZMQ会自动重发消息，直到获取应答为止。
• 你的接收端只调用zmq_recv()读取消息，从未发送应答，发送端的REQ套接字就会不断重发"Hello"，导致接收端无限重复收到同一内容。

方案1：遵循REQ-REP协议，添加应答 #

修改接收端，在收到消息后发送一个应答，完成REQ-REP的完整交互流程，发送端收到应答后就不会再重发消息。

修改后的C接收端代码 #

#include <zmq.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>

int main (void)
{
    void *context = zmq_ctx_new();
    void *replier = zmq_socket(context, ZMQ_REP);
    int rc = zmq_bind(replier, "ipc://zmq-ffi-1");
    assert(rc == 0);

    char buffer[10];
    while(1) {
        zmq_recv(replier, buffer, 10, 0);
        printf("Received: %s\n", buffer);
        
        // 发送应答，满足REQ-REP协议要求
        zmq_send(replier, "OK", 2, 0);
    }

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

可选修改Perl发送端（接收应答） #

#!/usr/bin/perl

use strict;
use warnings;

use ZMQ::FFI qw(ZMQ_REQ);

my $endpoint = "ipc://zmq-ffi-1";
my $ctx      = ZMQ::FFI->new();

my $p = $ctx->socket(ZMQ_REQ);
$p->connect($endpoint);

$p->send('Hello');
# 接收应答，完成完整交互
my $reply = $p->recv();
print "Received reply: $reply\n";

sleep(5);

方案2：改用无应答的PUSH-PULL模式 #

如果你的业务不需要请求-应答机制，直接使用PUSH-PULL单向模式更合适：发送端发送一次消息即完成，接收端不会收到重复消息（除非发送端再次发送）。

修改后的Perl发送端代码 #

#!/usr/bin/perl

use strict;
use warnings;

use ZMQ::FFI qw(ZMQ_PUSH);

my $endpoint = "ipc://zmq-ffi-1";
my $ctx      = ZMQ::FFI->new();

my $p = $ctx->socket(ZMQ_PUSH);
$p->connect($endpoint);

$p->send('Hello');
sleep(5);

修改后的C接收端代码 #

#include <zmq.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>

int main (void)
{
    void *context = zmq_ctx_new();
    void *puller = zmq_socket(context, ZMQ_PULL);
    int rc = zmq_bind(puller, "ipc://zmq-ffi-1");
    assert(rc == 0);

    char buffer[10];
    // 仅读取一次消息，之后可选择退出或继续监听新消息
    zmq_recv(puller, buffer, 10, 0);
    printf("Received: %s\n", buffer);
    
    // 如果需要持续监听新消息，保留while循环即可，不会重复收到旧消息
    // while(1) {
    //     zmq_recv(puller, buffer, 10, 0);
    //     printf("Received: %s\n", buffer);
    // }

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

注意点 #

• PUSH-PULL模式下，如果接收端启动晚于发送端，可能会错过消息（默认无缓存）；REQ-REP模式则会缓存消息直到收到应答。根据你的业务场景选择合适的模式。

内容的提问来源于stack exchange，提问作者Guy