c++ - ZeroMQ:使用 EPGM 传输

Question

我正在尝试在我的简单发布者-订阅者程序中使用 epgm 传输，但我无法这样做。据我了解，我无法在 bind 和 connect 语句中提供正确的地址字符串。

发布者和订阅者可以在相同或不同的机器上运行。

以下是使用 tcp 传输并正常工作所需的代码。它使用 cppzmq:https://github.com/zeromq/cppzmq .

发布商代码:

#include <zmq.hpp>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

int main () {
  zmq::context_t context (1);
  zmq::socket_t publisher (context, ZMQ_PUB);
  publisher.bind("tcp://10.1.1.8:5000");

  int i = 0;
  while (1) {
    int topic = 101;

    zmq::message_t message(50);
    snprintf ((char *) message.data(), 50, "%03d %10d %10d", topic, i, i);
    //fprintf(stderr, "message: %s\n", (char *) message.data());

    publisher.send(message);
    ++i;
  }
  return 0;
}

订阅者代码:

#include <zmq.hpp>
#include <iostream>
#include <sstream>
#include <unistd.h>
#include <cassert>

int main (int argc, char *argv[]) {
  zmq::context_t context (1);

  zmq::socket_t subscriber (context, ZMQ_SUB);
  subscriber.connect("tcp://10.1.1.8:5000");

  const char *filter = "101 ";
  subscriber.setsockopt(ZMQ_SUBSCRIBE, filter, strlen (filter));

  zmq::message_t tp;

  int maxx = 0;
  for (int i = 0; i < 1000; ++i) {
    zmq::message_t update;
    int topic, a, b;
    if(subscriber.krecv(&update, ZMQ_DONTWAIT)) {
      //fprintf(stderr, "size of data received: %zd\n", sizeof(update.data()));
      std::istringstream iss(static_cast<char*>(update.data()));
      iss >> topic >> a >> b;
      assert(a == b);
    }
    else {
      --i;
    }

    maxx = a > maxx ? a : maxx;
  }

  fprintf(stderr, "maxx = %d\n", maxx);
  return 0;
}

krecv 订阅者使用的方法:

inline bool krecv (message_t *msg_, int flags_ = 0) {
  int nbytes = zmq_msg_recv (&(msg_->msg), ptr, flags_);
  if (nbytes >= 0)
    return true;
  if (zmq_errno () == EAGAIN)
    return false;
  return false;
}

我尝试将发布者中的 bind 语句更改为以下内容:

1. publisher.bind("epgm://10.1.1.8:5000");
2. publisher.bind("epgm://224.1.1.1:5000");
3. publisher.bind("epgm://eth0;224.1.1.1:5000");
4. publisher.bind("epgm://10.1.1.8;224.1.1.1:5000");
5. publisher.bind("epgm://localhost:5000");

对于所有 5 种情况，程序崩溃并显示 Assertion failed: false (src/pgm_socket.cpp:165)。对于第 5 种情况 (epgm://localhost:5000)，我还收到以下警告以及崩溃:

Warn: Interface lo reports as a loopback device.
Warn: Interface lo reports as a non-multicast capable device.

我该如何解决这个问题？我猜测发布者和订阅者的地址更改将相同？

我正在使用 libpgm 5.2.122 和 zeromq-4.1.3。

注意机器有以下接口(interface):

1. eth0(以太网)——inet 地址:10.1.1.8
2. ib0 (InfiniBand) -- inet 地址:10.1.3.8
3. lo(本地环回)-- inet addr:127.0.0.1

Answer 1

在您的绑定(bind)中尝试一个 239.0.0.0/8 IP:

publisher.bind("epgm://;239.0.0.1:5000");

Wikipedia :

The 239.0.0.0/8 range is assigned by RFC 2365 for private use within an organization. From the RFC, packets destined to administratively scoped IPv4 multicast addresses do not cross administratively defined organizational boundaries, and administratively scoped IPv4 multicast addresses are locally assigned and do not have to be globally unique.