问题标签 [jeromq]

我在多线程环境中使用 Jeromq，如下所示。下面是我的代码，其中的构造函数SocketManager首先连接到所有可用的套接字，然后我将它们放在方法中的liveSocketsByDatacentermap 中connectToZMQSockets。之后，我在同一个构造函数中启动一个后台线程，该线程每 30 秒运行一次，它调用updateLiveSockets方法来 ping 地图中已经存在的所有套接字，并根据这些套接字是否处于活动状态来liveSocketsByDatacenter更新地图。liveSocketsByDatacenter

多个读取器线程同时调用该方法以获取下一个实时可用的套接字，getNextSocket()然后我们使用该套接字在其上发送数据。所以我的问题是我们在多线程环境中正确使用 Jeromq 吗？因为我们刚刚在我们的生产环境中看到了一个异常，当我们试图将数据发送到那个活动套接字时，这个堆栈跟踪所以我不确定它是一个错误还是其他什么？

下面是我的代码：

以下是我如何从多个阅读器线程同时使用类getNextSocket()方法：SocketManager

我不认为这是正确的，我相信。似乎getNextSocket()可以返回一个0MQ socketto thread A。同时，计时器线程可以访问0MQ socket它以对其进行 ping 操作。在这种情况下thread A和定时器线程发生变异相同0MQ socket，这会导致问题。那么解决此问题的最佳和有效方法是什么？

注意： SocketHolder 是一个不可变的类

更新：

我只是注意到我的另一个盒子上发生了同样的问题，ArrayIndexOutOfBoundsException但这次是"YQueue"文件中的 71 行号。唯一一致的是256总是。所以肯定有一些与 256 相关的东西，我无法弄清楚这里的 256 是什么？

我们的应用程序使用 JeroMQ 0.4.3 在客户端应用程序和 API 之间发送消息。（使用 ZAuth 的曲线加密）。

我们从ROUTER代理（MDP 模式）中的套接字获取客户端消息。

在代理中接收消息时找到发件人 IP 的最佳方法是什么？

在 ZeroMQ 中，似乎可以使用ZMQ_SRCFD消息选项，getpeername()但我还没有找到将其转换为 JeroMQ 的方法。

查看 ZeroMQ 绑定。我发现jzmq 上一次发布几乎是三年前。此绑定是否已弃用？

另一方面，ZeroMQ 有一个名为 JeroMQ 的本地 Java 实现，它受到 ZeroMQ 社区的支持，在Github中有很多活动，它基于最新版本的 libzmq(4.1.7) 之一。

所有使用 ZeroMQ 的新 Java 开发都应该基于 JeroMQ 吗？（考虑到 jzmq 和 JeroMQ API 开始出现分歧）

JeroMQ 如何与其他语言的 ZeroMQ 绑定互操作？

我目前正在对一些通信中间件进行延迟基准测试，我发现 ZeroMQREQ和REP套接字速度非常慢。例如，我期望 ZeroMQ 会比 ZeroC Ice 快。为了完整起见，这里是迄今为止的测试结果：

1. Ice 每秒 7600 条消息
2. ZeroMQ 4300 - 每秒 4500 条消息

该消息是以数据结构作为参数的远程调用。数据结构包含基本类型的字段（int、float、double、boolean、string）。起初我怀疑使用 Protocol Buffers 序列化数据可能是 ZeroMQ 的瓶颈，但后来我使用空消息进行了测试，结果非常相似。即使 ZeroMQ 发送空消息，Ice 也更快。

尽管如此，我还是想给 ZeroMQ 一个公平的机会。REQ本着这种精神，我想知道我是否可以REP以任何方式提高速度。也许我应该使用其他套接字？唯一的限制是通信应该类似于 RPC。客户端在期望其消息的结果时不应做任何工作。

这是我的实现代码。我提供了 Java 版本的基准测试，但 Python（慢一点）和 C++（快一点）的结果非常相似：

客户

服务器

其余代码基本上用于计算平均延迟和每秒消息数。

也许 ZeroMQ 对于这种通信模式来说只是慢，它在多对多通信模式中表现出色......

注意：我在记录结果之前“加热”了两个中间件。尽管如此，ZeroMQ 总是比 Ice 慢。

EDIT1：我将升温迭代次数增加到 10000。新结果是：

1. Ice 每秒 8100 条消息（随着变暖而收敛到 8250）
2. ZeroMQ + 协议缓冲区收敛到每秒 4650 条消息

在我们最近的项目中，我们尝试保护 pub-sub jeromq 连接。

为此，我们希望主机首先通过网络对新主机进行身份验证，并且阻止未通过身份验证的每个主机订阅。另一个好处是在网络中发送的消息更少。

那么，是否可以创建发布白名单？

在我的代码中，我有一个线程等待 ZeroMQ 端点上的事件，它的主循环看起来类似于：

.recv()问题是如果我想退出线程（例如因为我想关闭应用程序），我不知道如何中断对方法的调用。

我试图中断线程，但这似乎不起作用。

我有一个简单的应用程序，可以将数据从一台机器传输到另一台机器。随着应用程序的运行，堆的大小正在缓慢增加。于是我dump了堆，分析了一下，发现zmq.poll.Poller消耗的内存最大。它们属于线程' iothread-2'：

堆截图在这里

我的应用程序演示是这样的：

当我检查Poller堆中的对象时，我发现有 400 万个HashMap$Node，并且 hashmap 节点的值是一个包含 10 个空对象数组列表的列表。

堆被命令转储：
jmap -dump:live,format=b,file=dump.hprof [pid]

jdk 是 1.8.0_131，操作系统是 CentOS 7.2.1511 和 jeromq 0.4.2

我用poller错了吗？非常感谢任何帮助的人！

是否可以使用 ZeroMQ（使用 JeroMQ Java 移植）实现带队列的无代理网络？

在我的网络中，所有对等点都是发布者和接收者（SUB/PUB模式），因此当对等点发送消息时，所有其他对等点都会收到消息。

问题是消息不可靠并且可能会丢失（例如连接问题）并且不再恢复。

我想实现一个队列，对等方可以检索他们未收到的消息。

我正在查看本指南（即使它是针对 Python 的），看来我应该实现XREP/XREQ模式：

但似乎这只能实现队列服务器。这是真的吗？

据我从文档中可以看出，ZeroMQ 套接字不应该从不同的线程中使用（例如读取/写入）。

这反过来又阻止了我在默认调度程序上运行的 Akka Actor 中使用 ZMQ 套接字（不能保证哪个线程将执行我的receive方法）。

使用PinnedDispatcher 是否允许我在 Actor 内安全地使用此类套接字，前提是我注意不要阻塞（至少不要阻塞过多的时间）？

这个调度器为每个使用它的actor指定一个唯一的线程；即每个actor都有自己的线程池，池中只有一个线程

通过：https ://doc.akka.io/docs/akka/2.5/dispatchers.html#types-of-dispatchers

我正在使用 JeroMQ 0.4.0 和 Akka 2.5。我意识到 Akka 曾经有一个 ZeroMQ 扩展，但似乎大部分都被放弃了。

我是 ZeroMQ 的新手，并且似乎在我的begin()方法中循环丢失了消息。

我想知道我是否错过了一个我没有排队消息或什么的东西？

当我在我的发布者上引发一个事件时，它会向我的订阅者发送两条消息，中间有一个小间隔，我似乎没有收到第二条被中继的消息。我错过了什么？