c++ - ZeroMQ C++中的多个发布者这是一个不错的选择吗？

Question

我是ZeroMQ的新手。我想创建多个发布者，每个发布者都在其中发布特定的数据，例如:

发布者1:发布图像数据

发布者2:发布音频数据

发布者3:发布文本数据

基本上，我的要求是从多个发布者发布数据，并在另一侧使用多个接收者进行接收。

请参见下面的示例代码:

data_publisher.cpp

//  Prepare our context and all publishers
zmq::context_t context(1);
zmq::socket_t publisher1(context, ZMQ_PUB);
zmq::socket_t publisher2(context, ZMQ_PUB);
zmq::socket_t publisher3(context, ZMQ_PUB);
zmq::socket_t publisher4(context, ZMQ_PUB);

publisher1.bind("tcp://*:5556");
publisher2.bind("tcp://*:5557");
publisher3.bind("tcp://*:5558");
publisher4.bind("tcp://*:5559");

//  Initialize random number generator
srandom((unsigned)time(NULL));
while (1) {
    // sample data
    int zipcode1 = within(100000);
    int zipcode2 = within(100000);
    int zipcode3 = within(100000);
    int zipcode4 = within(100000);

    int temperature1 = within(215) - 80;
    int temperature2 = within(215) - 80;
    int temperature3 = within(215) - 80;
    int temperature4 = within(215) - 80;

    int relhumidity1 = within(50) + 10;
    int relhumidity2 = within(50) + 10;
    int relhumidity3 = within(50) + 10;
    int relhumidity4 = within(50) + 10;

    zmq::message_t message1(20);
    zmq::message_t message2(20);
    zmq::message_t message3(20);
    zmq::message_t message4(20);

    snprintf((char*)message1.data(), 20, "%05d %d %d", zipcode1, temperature1, relhumidity1);
    snprintf((char*)message2.data(), 20, "%05d %d %d", zipcode2, temperature2, relhumidity2);
    snprintf((char*)message3.data(), 20, "%05d %d %d", zipcode3, temperature3, relhumidity3);
    snprintf((char*)message4.data(), 20, "%05d %d %d", zipcode4, temperature4, relhumidity4);

    publisher1.send(message1);
    publisher2.send(message2);
    publisher3.send(message3);
    publisher4.send(message4);
}

data_receiver.cpp

zmq::context_t context(1);

//  Socket to talk to server
zmq::socket_t subscriber1(context, ZMQ_SUB);
zmq::socket_t subscriber2(context, ZMQ_SUB);
zmq::socket_t subscriber3(context, ZMQ_SUB);
zmq::socket_t subscriber4(context, ZMQ_SUB);

subscriber1.connect("tcp://localhost:5556");
subscriber2.connect("tcp://localhost:5557");
subscriber3.connect("tcp://localhost:5558");
subscriber4.connect("tcp://localhost:5559");

const char* filter = (argc > 1) ? argv[1] : "10001 ";
subscriber1.setsockopt(ZMQ_SUBSCRIBE, filter, strlen(filter));
subscriber2.setsockopt(ZMQ_SUBSCRIBE, filter, strlen(filter));
subscriber3.setsockopt(ZMQ_SUBSCRIBE, filter, strlen(filter));
subscriber4.setsockopt(ZMQ_SUBSCRIBE, filter, strlen(filter));

//  Process 100 updates
int update_nbr;
long total_temp1 = 0;
long total_temp2 = 0;
long total_temp3 = 0;
long total_temp4 = 0;

for (update_nbr = 0; update_nbr < 100; update_nbr++)
{
    zmq::message_t update1;
    zmq::message_t update2;
    zmq::message_t update3;
    zmq::message_t update4;
    int zipcode1, temperature1, relhumidity1;
    int zipcode2, temperature2, relhumidity2;
    int zipcode3, temperature3, relhumidity3;
    int zipcode4, temperature4, relhumidity4;

    subscriber1.recv(&update1);
    subscriber2.recv(&update2);
    subscriber3.recv(&update3);
    subscriber4.recv(&update4);

    std::istringstream iss1(static_cast<char*>(update1.data()));
    std::istringstream iss2(static_cast<char*>(update2.data()));
    std::istringstream iss3(static_cast<char*>(update3.data()));
    std::istringstream iss4(static_cast<char*>(update4.data()));

    iss1 >> zipcode1 >> temperature1 >> relhumidity1;
    iss2 >> zipcode2 >> temperature2 >> relhumidity2;
    iss3 >> zipcode3 >> temperature3 >> relhumidity3;
    iss4 >> zipcode4 >> temperature4 >> relhumidity4;

    total_temp1 += temperature1;
    total_temp2 += temperature2;
    total_temp3 += temperature3;
    total_temp4 += temperature4;
}

std::cout << "Average temperature for zipcode '" << filter << "' was "
          << (int)(total_temp1 / update_nbr) << "F" << std::endl;
std::cout << "Average temperature for zipcode '" << filter << "' was "
          << (int)(total_temp2 / update_nbr) << "F" << std::endl;
std::cout << "Average temperature for zipcode '" << filter << "' was "
          << (int)(total_temp3 / update_nbr) << "F" << std::endl;
std::cout << "Average temperature for zipcode '" << filter << "' was "
          << (int)(total_temp4 / update_nbr) << "F" << std::endl;

请注意，以上代码是获取建议/建议的示例代码。

我想知道它是否是一个不错的选择，如上面的示例代码所示？

Answer 1

仍在等待任何定量的事实，但让我们开始:

ZeroMQ是使用智能启用工具的概念，而ZeroMQ核心元素 Context -engine隐藏了底层系统编程。

这就是说，高级工具正式化为可扩展的正式通信模式原型(prototype)，提供了某种模仿人类的行为-PUB发布者确实是“发布”， SUB 订阅者可以“订阅”， REQ 请求者可以“请求” ”， REP 复制者确实可以“回复”，等等。

一旦被授予一些基本规则，这些具有行为的AccessPoint可以将 .bind()/.connect() 引入某种分布式行为基础架构中。这样的规则之一就是不打扰实际的运输类，因为它们都是功能丰富的技术，当前跨越了{ inproc:// | ipc:// | tcp:// | pgm:// | epgm:// | vmci:// }，低级细节的情况， Context() -instance将处理所有这些问题
对您的高级行为透明。只需忘记这一点。另一个规则是，您可以确保发送的每条消息都没有错误，或者完全没有错误-在这方面没有妥协，从来没有传递过折磨的垃圾来欺骗或破坏收件人的AccessPoint后处理。

不了解这一点，ZeroMQ并没有最大限度地为我们提供这种豪华工具所设计的舒适性和功能。

回到你的困境:

上面已经说了几句话，但是您的主要体系结构还不清楚，仍然可以在这里为您提供帮助。

ZeroMQ抽象的分布式行为套接字工具主要是纯 [SERIAL] 调度设备。这意味着，与套接字相关联的接收访问点{ .bind() | .connect() }中的任何一个都不能期望对消息的纯半随机流进行任意重新排序。

这意味着，在任何情况下，无论是“只是”- [CONCURRENT] 过程计划，还是在极端情况下，在技术上精心策划了真正的- [PARALLEL] 过程计划，单个“pure”- [SERIAL]交付通道将不允许{ [CONCURRENT] | [PARALLEL] } -system继续提供这种流程调度模式，并且会将事件/处理流程斩断为“纯”- [SERIAL] 消息序列。

A)这个确实是引入多个独立操作的ZeroMQ分布式行为Socket实例的原因和必须的。

另一方面，对全局分布式系统的行为一无所知，尚无法确定，进入多个独立操作的Socket实例是否不仅浪费时间和资源，而且不合理地交付由于极端错误或完全缺少初始工程决策，导致端到端系统行为性能低于平均水平或无法接受。

性能？

永远不要猜这个 Realm 。而是从首先定量声明的需求开始，基于该需求，技术上合理的设计将能够进行并定义资源映射和性能调整直至平台限制所需的所有步骤。

在过去的二十年中，ZeroMQ拥有出色的性能，具备出色的性能，并且设计和工程团队在完善可扩展性和性能范围方面做了大量工作，同时保持延迟几乎无法通过临时方式实现程序员。确实，ZeroMQ地下室中隐藏着大量的系统编程。

“数据是巨大的”-好的，定义了的大小-传递大小为的 1E+9 消息比传递1e + 3消息的大小为 1 [B] 的消息具有其他性能上的调整。

“尽可能快”-好的，为给定大小和消息的预期节奏的定义的快速，10 / s〜 1.000.000 [B] ，1000 / s〜

当然，在某些情况下，这种混合需求可能会超出现代计算设备功能范围的范围。必须在进行任何编程之前就对它进行最好的审查，因为否则，您可能会花很多精力在永远不会实现的事情上进行编程工作，因此最好有一个积极的证据证明解决方案体系结构在一个操作系统中是可行和可行的。可接受的资源和成本范围。

因此，如果您的项目需要某些东西，则首先定义并定量指定实际内容，然后接下来，解决方案体系结构可以开始对其进行分类并提供决策，哪些工具和哪些工具配置可以匹配所定义的目标级别功能和性能目标。

盖屋顶开始建造房屋永远不会回答以下问题:如何布置地下室墙，这将是足够但尚未过分设计的厚度的铁混凝土铠装，它将携带未知数量的高强度混凝土。高架建筑地板。已经建好的屋顶很容易出现，但与系统和严格的设计与工程实践无关。