如何使 Zeromq PUB/SUB 删除旧消息而不是新消息（用于实时提要）？

说我有一个PUB服务器zmq_send()的实时消息SUB客户。如果客户很忙而无法zmq_recv()消息足够快，那么消息将在客户端（和/或服务器）中缓冲。

如果缓冲区变得太大（高水位线），则新消息将被丢弃。对于实时消息来说，这与人们想要的相反。应删除旧消息，为新消息腾出位置。

有什么方法可以做到这一点吗？

理想情况下我想要SUB客户端的接收队列为空或仅包含最新消息。当收到新消息时，它将替换旧消息。 （我猜这里的问题是客户端会阻止zmq_recv()当队列为空时，这样做会浪费时间。 ）

那么实时提要通常是如何实现的ZeroMQ?

我会在这里回答我自己的问题。设置 ZMQ_CONFLATE“仅保留最后一条消息”似乎很有希望，但它不适用于订阅过滤器。它只在队列中保留一条消息。如果您有多个过滤器，则其他过滤器类型的旧消息和新消息都会被丢弃。

同样，zeromq 指南建议简单地杀死慢速订阅者，但这似乎不是现实的解决方案。让具有不同阅读速度的订阅者订阅同一个快速发布者，应该是正常的用例。其中一些订阅者可能生活在慢速计算机上，其他人可能生活在快速计算机上，等等。ZeroMQ 应该能够以某种方式处理这个问题。

http://zguide.zeromq.org/page:all#Slow-Subscriber-Detection-Suicidal-Snail-Pattern http://zguide.zeromq.org/page:all#Slow-Subscriber-Detection-Suicidal-Snail-Pattern

我最终在客户端手动删除旧的排队消息。看起来效果很好。我通过这种方式向客户端订阅了不到 3 毫秒的消息（通过 tcp localhost）。即使在队列中有 5000 条 10 秒的旧消息，位于后面那几条实时消息前面的情况下，这种方法也有效。这对我来说已经足够好了。

我不禁认为这是图书馆应该提供的东西。它可能可以做得更好。

无论如何，这是客户端，旧消息丢失，代码：

bool Empty(zmq::socket_t& socket) {
    bool ret = true;
    zmq::pollitem_t poll_item = { socket, 0, ZMQ_POLLIN, 0 };
    zmq::poll(&poll_item, 1, 0); //0 = no wait
    if (poll_item.revents & ZMQ_POLLIN) {
        ret = false;
    }
    return ret;
}

std::vector<std::string> GetRealtimeSubscribedMessageVec(zmq::socket_t& socket_sub, int timeout_ms)
{
    std::vector<std::string> ret;

    struct MessageTmp {
        int id_ = 0;
        std::string data_;
        boost::posix_time::ptime timestamp_;
    };

    std::map<int, MessageTmp> msg_map;

    int read_msg_count = 0;
    int time_in_loop = 0;
    auto start_of_loop = boost::posix_time::microsec_clock::universal_time();
    do {
        read_msg_count++;

        //msg format sent by publisher is: filter, timestamp, data
        MessageTmp msg;
        msg.id_ = boost::lexical_cast<int>(s_recv(socket_sub));
        msg.timestamp_ = boost::posix_time::time_from_string(s_recv(socket_sub));
        msg.data_ = s_recv(socket_sub);

        msg_map[msg.id_] = msg;

        auto now = boost::posix_time::microsec_clock::universal_time();
        time_in_loop = (now - start_of_loop).total_milliseconds();
        if (time_in_loop > timeout_ms) {
            std::cerr << "Timeout reached. Publisher is probably sending messages quicker than we can drop them." << std::endl;
            break;
        }
    } while ((Empty(socket_sub) == false)); 

    if (read_msg_count > 1) {
        std::cout << "num of old queued up messages dropped: " << (read_msg_count - 1) << std::endl;
    }

    for (const auto &pair: msg_map) {
        const auto& msg_tmp = pair.second;

        auto now = boost::posix_time::microsec_clock::universal_time();
        auto message_age_ms = (now - msg_tmp.timestamp_).total_milliseconds();

        if (message_age_ms > timeout_ms) {
            std::cerr << "[SUB] Newest message too old. f:" << msg_tmp.id_ << ", age: " << message_age_ms << "ms, s:" << msg_tmp.data_.size() << std::endl;
        }
        else {
            std::cout << "[SUB] f:" << msg_tmp.id_ << ", age: " << message_age_ms << "ms, s:" << msg_tmp.data_.size() << std::endl;
            ret.push_back(msg_tmp.data_);
        }
    }

    return ret;
}