问题标签 [netmq]

I am trying to implement a simple message passing between two applications using NetMQ (a slightly more elaborate description of what I am trying to achieve is below).
After a bit of trial and error I've found that I can't just send or receive messages right away after a Connect/Bind calls, since they are non blocking and actually return even if the connection hasn't been established yet.
For now I solved this with Thread.Sleep(), but this has a bad taste to it and definitely a no-go for a production system.

So the question is, how's one supposed to do it in NetMQ/ZeroMQ?

Client example:

Server example:

Description of what I am trying to achieve:

Client - Sends messages to a single server. The client should not block and should not discard messages when server is not available. The client can come offline/online at any time.

Server - Receives messages from a single client. The server blocks until a message is received. Server needs to do lengthy processing of the message and should not loose any other messages while processing. The server can come offline/online at any time.

目标：在一次迭代中向所有订阅者发送大量消息。我有 40k-100k 条消息。我已经开始使用 PUB/SUB 套接字类型。

问题：订阅者收到的消息数量低于发布者发送的消息数量。如果我在发送每条消息后添加一个 Thread.Sleep(1)，那么所有消息都会被传递，但是需要传递的消息数量很多，这意味着 40-100 秒的延迟。这是不可接受的。

下面的代码在 NetMQ (3.0.0) 中，它是 alpha 版本，但它只是作为示例，因为我已经使用 libzmq 3.2.4 (stable) 在 c 中实现了相同的代码。和simptoms是一样的。

发布者/服务器端：

订户/客户端：

我正在尝试实现 NetMQ Pub/Sub 模型，但订阅者没有收到任何消息。这里可能有什么问题？

两者都在不同的应用程序中。

我正在使用基于 ZeroMq 和 NetMQ 的通信（两个项目中的问题相同）

我有应用程序在防火墙后面运行，

在服务器端，它很容易定义为入站流量打开哪些端口，

但是在客户端，我似乎只能指定目标（又名服务器）地址和端口。

是否可以定义哪个端口将用于客户端应用程序上的出站流量。

例如（在 NetMQ 中）

在一个简单的通信场景中，我们有一个服务器和一个客户端

服务器正在机器 A (192.168.1.10) 上运行 - 并监听端口 5555

客户端在机器 B (192.168.1.9) 上运行 - 并且正在启动与服务器 (A) 的通信

如果我们查看客户端上的通信信息，我们会看到系统分配了端口 XXXXX（例如 51234）及其出站到 192.168.1.10:5555

大多数情况下 XXXXX 端口是系统分配的（空闲端口），但在某些极端情况下，XXXXX 端口需要是特定端口（由于极端的安全环境）

第一次使用 ZMQ，我正在尝试设置一个进程来处理许多 getimage 请求。当我调试时会发生一些异常，我试图修复并实现一种停止 QueueDevice 终止所有线程并正常退出的方法。

1. 接收者.连接（后端绑定地址）；在 NetMQ.dll 中发生类型为“NetMQ.InvalidException”的未处理异常，错误代码为 NetMQ.zmq.ErrorCode.EINVAL。为什么这个异常不会停止线程的进一步执行？
2. 我尝试将 QueueDevice 设为静态字段并在关闭消息函数中使用 queueDevice.stop() 但随后线程开始抛出 TerminatingExceptions 并且永不退出。那么我可以关闭所有线程和主线程吗？

测试驱动代码

我正在使用的代码

我需要从不同的代理向单个进程发送大量短消息。

来自 10 个代理的大约 10-15 2-3 Kb 消息/秒。

一些代理在同一台机器上运行，而另一些则在同一 VLAN 中的不同机器上运行。

要求是消息生成和主服务内开始处理它之间的最小延迟。峰值的最大值50ms是可以接受的（尽管不是理想的）。

我计划使用 NetMq 进行传输。

Q1： socket类型和协议的最佳配置是什么？

Q2：使用DEALER/ ROUTER，PUSH/PULL或PUSH/ROUTER设置更好吗？

Q3：我应该 IPC在同一台机器上为代理使用套接字还是TCP对两种类型的代理都足够快？

Q4：我需要在同一个上下文中创建TCP和IPC套接字还是创建一个单独的上下文？

我正在使用 Netmq 在我的前端和后端之间实现可靠、快速和高可用性的通信。

通信流程是： - 客户端连接（xmpp 客户端、websockets 或任何其他客户端类型）并向后端发送消息，此消息必须由特定类型的工作人员处理（xmpp 调度程序处理 xmpp 消息） - 调度程序连接（例如 xmpp 调度程序）并将消息发送到特定的前端客户端（xmpp 客户端）。

因此，通信可以由顶部（客户端）或底部（调度程序）发起。

为了实现上述目标，这里有一个架构模型：

由于我在http://zguide.zeromq.org/page:all上阅读的内容，这还不可靠，因为某个客户端接收到发送到前端（代理）的消息并且该消息被接受在知道是否有任何调度员可以处理我的工作之前。在这个阶段，我可以丢弃此消息或将其存储，以在将来有工作人员可用时进行处理。这不是一个好的解决方案，因为如果 Broker 长时间无人传送流量，大量消息将在 Broker 上累积。正确的行为是：代理不应该接受来自客户端的任何消息，或者客户端应该停止发送消息。

这是我想要实现的行为。为了实现这一点，我需要实施 Majordomo Pattern 对吗？可能是 Majordomo 管理界面吧？问题是我找不到 C# Majordomo 实现的任何好的示例，只有 C、C++ 和 Python 示例。在 Netmq 或任何其他 C# ZMQ 绑定上是否有任何可用的 Majordomo 模式实现？拥有它会很棒，因为仅使用 Netmq 来实现 Majordomo 非常复杂，而且我在这方面还很陌生。

我正在构建一个简单的 Azure 工作者角色来订阅 zeromq 发布者（在工作者中使用 NetMQ）并将消息存储到 Azure 表中。

当我在我的开发机器上运行工作程序时，一切正常。消息进来并被写入表中。

但是，当我将工作人员部署到 Azure 时，某些东西会停止工作。我可以附加到工作人员并看到它卡在阻塞的subscriberSocket.Receive() 调用上。从工作人员的角度来看，订阅套接字不会出现任何消​​息。

我已经对工作人员进行了 RDP，并使用 netstat 我可以看到从订阅套接字到远程发布者的传出连接已建立，因此它看起来不像是防火墙问题。

从日志/诊断信息来看，一切似乎都很正常，只是没有收到任何消息。

我错过了一些 Azure 配置吗？允许传出连接但阻止数据通过的东西？

工人代码的相关部分：

我已经开始探索 NetMQ 3.3.0.11 并遇到了使用 Poller 的问题。

我试图实现轮询器轮询大约 1 秒，然后停止并允许在它恢复轮询 1 秒之前完成其他操作，依此类推。

我有以下代码：

我面临的问题是，在轮询期间，即使准备好接收消息，也不会触发 Client.ReceiveReady 事件。并提出了一个InvalidOperationException陈述Poller is started。

知道我做错了什么吗？