将 ZeroMQ 与 Boost::ASIO 一起使用

Question

我有一个使用 ZeroMQ 进行消息传递的 C++ 应用程序。但它还必须为基于 AJAX / Comet 的 Web 服务提供 SGCI 连接。

为此，我需要一个普通的 TCP 套接字。我可以通过普通的 Posix 套接字来做到这一点，但为了保持跨平台的便携性并让我的生活更轻松（我希望......）我正在考虑使用 Boost::ASIO。

但是现在我遇到了 ZMQ 想要使用它自己的 zmq_poll() 和 ASIO 的冲突 io_service.run()...

有没有办法让 ASIO 与 0MQ zmq_poll() 一起工作？

或者是否有其他推荐的方法来实现这样的设置？

注意：我可以通过使用多个线程来解决这个问题 - 但它只是一个小的单核/CPU 机器，它会以非常低的 SCGI 流量运行该程序，所以多线程会浪费资源......

Answer 1

阅读文档here和here后，特别是这一段

ZMQ_FD：检索与套接字关联的文件描述符 ZMQ_FD 选项应检索与 指定的套接字。返回的文件描述符可用于 将套接字集成到现有的事件循环中； ØMQ 库 应在边缘触发的套接字上发出任何未决事件的信号 通过使文件描述符准备好读取来实现时尚。

我认为您可以对每个zmq_pollitem_t 使用null_buffers，并将事件循环推迟到io_service，完全绕过zmq_poll()。然而，上述文档中似乎有一些警告，特别是

从返回的文件描述符中读取的能力不 必然表明可以从中读取消息，或 可以写入底层套接字；应用程序必须检索 随后检索 ZMQ_EVENTS 的实际事件状态 选项。

因此，当您的一个 zmq 套接字的处理程序被触发时，您必须在处理我认为的事件之前做更多的工作。未编译的伪代码如下

const int fd = getZmqDescriptorSomehow();
boost::asio::posix::stream_descriptor socket( _io_service, fd );
socket->async_read_some(
    boost::asio::null_buffers(),
    [=](const boost::system::error_code& error)
    {
       if (!error) {
           // handle data ready to be read
       }
     }
);

请注意，您不必在这里使用 lambda，boost::bind 到成员函数就足够了。

Answer 2

最后我发现有两种可能的解决方案：

• Sam Miller's 我们使用 ASIO 事件循环的地方
• ZeroMQ 的事件循环通过通过acceptor 和socket 的.native() 方法获取ASIO 文件描述符并将它们插入到zmq_pollitem_t 的数组中

我已接受 Sam Miller 的回答，因为这对我来说是 SCGI 情况下的最佳解决方案，即不断创建和结束新连接。因此，处理每一个变化的zmq_pollitem_t 数组是一件很麻烦的事，可以通过使用 ASIO 事件循环来避免。

Answer 3

获得 ZeroMQ 的套接字是战斗中最小的部分。 ZeroMQ 基于一个在 TCP 上分层的protocol，所以如果你走这条路，你将不得不在自定义的 Boost.Asio io_service 中重新实现 ZeroMQ。在使用 Boost.Asio 创建异步 ENet 服务时，我遇到了同样的问题，首先尝试使用 Boost.Asio UDP 服务从 ENet 客户端捕获流量。 ENet 是一个在 UDP 上分层的类似 TCP 的协议，所以我当时所做的只是捕获处于几乎无用状态的数据包。

Boost.Asio 是基于模板的，内置的 io_service 使用模板基本封装了系统套接字库来创建 TCP 和 UDP 服务。我的最终解决方案是创建一个自定义 io_service 来包装 ENet 库而不是系统套接字库，从而允许它使用 ENet 的传输功能，而不必使用内置的 UDP 传输重新实现它们。

ZeroMQ 也可以这样做，但 ZeroMQ 本身已经是一个非常高性能的网络库，它已经提供了异步 I/O。我认为您可以通过使用 ZeroMQ 现有的 API 接收消息并将消息传递到 io_service 线程池来创建一个可行的解决方案。这样消息/任务仍将使用 Boost.Asio 的反应器模式异步处理，而无需重新编写任何内容。 ZeroMQ 将提供异步 I/O，Boost.Asio 将提供异步任务处理程序/workers。

现有的 io_service 仍然可以耦合到现有的 TCP 套接字，允许线程池处理 TCP（在您的情况下为 HTTP）和 ZeroMQ。在这样的设置中，ZeroMQ 任务处理程序完全有可能访问 TCP 服务会话对象，从而允许您将 ZeroMQ 消息/任务的结果发送回 TCP 客户端。

以下只是为了说明这个概念。

// Create a pool of threads to run all of the io_services.
std::vector<boost::shared_ptr<boost::thread> > threads;
for(std::size_t i = 0; i < thread_pool_size_; ++i) {
    boost::shared_ptr<boost::thread> thread(new boost::thread(boost::bind(&boost::asio::io_service::run, &io_service_)));
    threads.push_back(thread);
}

while (1) {
    char buffer [10];
    zmq_recv (responder_, buffer, 10, 0);
    io_service_.post(boost::bind(&server::handle_zeromq_message, buffer, this));
}

Answer 4

在这个问题发生 2 年后，有人发布了一个正是这样做的项目。项目在这里：https://github.com/zeromq/azmq。讨论设计的博客文章在这里：https://rodgert.github.io/2014/12/24/boost-asio-and-zeromq-pt1/。

这是从自述文件中复制的示例代码：

#include <azmq/socket.hpp>
#include <boost/asio.hpp>
#include <array>

namespace asio = boost::asio;

int main(int argc, char** argv) {
    asio::io_service ios;
    azmq::sub_socket subscriber(ios);
    subscriber.connect("tcp://192.168.55.112:5556");
    subscriber.connect("tcp://192.168.55.201:7721");
    subscriber.set_option(azmq::socket::subscribe("NASDAQ"));

    azmq::pub_socket publisher(ios);
    publisher.bind("ipc://nasdaq-feed");

    std::array<char, 256> buf;
    for (;;) {
        auto size = subscriber.receive(asio::buffer(buf));
        publisher.send(asio::buffer(buf));
    }
    return 0;
}

看起来不错。如果您尝试，请在 cmets 中告诉我它是否在 2019 年仍然有效[我可能会在几个月后尝试，然后更新此答案]（回购已过时，最后一次提交是一年前）

Answer 5

解决方案是轮询您的 io_service 而不是 run()。

查看this solution 了解一些poll() 信息。

使用 poll 而不是 run 将允许您轮询 zmq 的连接，而不会出现任何阻塞问题。