问题标签 [netmq]

我有两个简单的组件，它们应该使用 REQ/REP ZeroMQ 模式相互通信。服务器（REP 套接字）是使用 pyzmq 在 Python 中实现的：

使用 NetMQ 库用 C# 编写的客户端（REQ 套接字）：

通过与使用 Python 实现的 REQ 套接字通信，Python 服务器实现已经过测试并且工作正常。但是 C# REQ 套接字在循环的第一次迭代中抛出以下错误，并且没有任何消息到达服务器：

NetMQ.dll 中出现“NetMQ.FiniteStateMachineException”类型的未处理异常附加信息：Req.XSend - 无法发送另一个请求

堆栈跟踪：

这是我使用 ZMQ 的第一步，C# 代码取自库文档。是什么让代码抛出这个错误？

我在用：

• pyzmq 14.7
• 网络MQ 3.3.3.4
• .NET 4.6

====================== 解决方案======================

正如@somdoron 在他的回答中所解释的那样，根本原因是两个套接字都需要完成发送/接收的完整周期，然后才能被重用。事实上，在 python 中实现的 REP 套接字也没有改变它的状态，所以错误出现在 python 和 C# 代码中。这是固定代码：

REP 插座

请求套接字

我在NetMQ v4中有一个经销商 <--> 路由器设置，我可以在任何方向异步发送和接收消息，没有问题。

我现在想将其形式化为服务器（路由器）侦听任何传入消息的抽象，但它还需要按需向任何连接的客户端（经销商）广播消息。

我试图避免使用 Pub <--> Sub 套接字，因为我需要订阅者也向服务器发送消息。与我试图实现的最接近的模式是 WebSocket 客户端-服务器通信。

侦听客户端消息的第一部分是通过以下方式完成的：

现在鉴于套接字不是线程安全的，我一直在寻找一种方法来向所有客户端广播消息（根据需要来自不同的线程）。

我可能可以TryReceiveMultipartMessage在循环中使用该方法，而不是设置超时，之后我可以检查队列中是否有任何广播消息，然后遍历每个发送此类消息的客户端。就像是：

这在某种程度上感觉不对，主要是因为：

• 超时的什么值是一个好的值？1 秒、100 毫秒等；
• 这是最有效/性能最好的解决方案吗，因为该程序将用于连接 10 万多个客户端，每个客户端每秒发送数千条消息。

任何关于什么是最好的方法的指针都非常感谢。

我正在尝试创建一个 rpc 程序来与位于不同网络上的主机通信，并选择此处提供的 NetMQ 的 Router-Dealer 配置：http: //netmq.readthedocs.io/en/latest/router-dealer/#router-dealer

但问题是路由器在将消息路由到后端时总是选择随机经销商。我使用的代码：

客户代码：

经销商代码：

是否可以在 frontend.ReceiveReady 上选择特定的后端？谢谢！

我正在尝试使用 NetMQ (3.3.3.4) 并创建一个 pub-sub 模式。

我希望主机/服务器监听一个端口（9000）上的所有传入数据，并将数据转发给另一个端口（9001）上的所有客户端/订阅者。

然后，客户端将在 9000 上发送数据，并在 9001 上接收（由任何人）发送的所有消息。

按照文档，我创建了类似下面的代码，但我无法让它工作。主要是，我相信，因为ReceiveReady从不调用！

我认为它应该如何工作：

• client.Publish应该导致第一行host.SubscriberSocket_ReceiveReady解除阻塞并将数据传递到另一个套接字
• 当数据被传递时，它应该出现在Task客户端的无限运行中

结果：

• // This line is never reached永远无法达到断点
• 任何地方都没有例外。
• 切换主机上的端口，使 publish = 9000 和 subscribe = 9001 无效
• Windows 防火墙已关闭，因此不应有任何阻塞
• 如果我将地址放入PublisherSocket构造函数中，或者我_publisherSocket.Bind(address)在 Host 或_publisherSocket.Connect(address)Client 中使用，这没有区别

我究竟做错了什么？

主持人

客户

测试仪

代码在这里：

您可以在以下位置阅读完整的主题： http: //netmq.readthedocs.io/en/latest/introduction/

我不明白的行是server.SendFrame("World");

服务器如何知道要发送此消息的客户端？

我需要在本地网络中创建具有以下功能的客户端-服务器应用程序：

1. 单服务器（数据访问等......）
2. 多客户端（普通3-5个，最多20个）
3. 每个客户端必须在服务器上授权（需要检查它有什么权限）
4. 客户端向服务器发送请求（通常小于 1 Kb，最大 3-5 Kb）并获得响应（普通 30-100 Kb，有时可能是大量数据，最大 1-2 Mb）
5. 在来自客户端的一些查询之后，服务器通知所有客户端并向它们发送新的和更新的数据（因此服务器必须知道连接了多少客户端）
6. 如果网络连接中断，客户端必须重新连接

我认为带有 Protobuf 的 NetMQ 足以满足我的目的。我查看文档，发现最适合我的任务的模式是http://zguide.zeromq.org/page:all#Service-Oriented-Reliable-Queuing-Majordomo-Pattern当每个客户端都是工作人员和服务器工作人员时同时是客户。

我认为这个解决方案也很复杂？是否有更简单的方法来解决此类问题（更简单的模式或基于 WCF 的方法）？

我对基于netMQ 4.0. 我尝试使用http://zguide.zeromq.org/cs:mtserver，但没有上下文netMQ 4.0。

我试过了：

但我得到错误：

NetMQ.TerminatingException: CheckContextTerminated

是的，它被终止了。

如何创建上下文netMQ 4.0或如何使用创建多线程服务器netMQ 4.0？

我想知道是否有办法在新订阅发布者套接字时引发事件

例如：

当然，OnSubscription实际上并不存在，但我想知道是否有任何解决方法。

我需要我的服务器知道所有订阅。

我虽然有两种方法来实现它：

1. 在服务器中创建一个额外的路由器套接字，所有订阅都将发送到发布者套接字和路由器套接字（取消订阅也是如此）。这将允许我从路由器轮询订阅。

2. 根本不使用发布者/订阅者，使用路由器/经销商创建所有发布订阅机制。

你会建议我做什么？

我在https://netmq.readthedocs.io/上采用了一个简单的接收/请求套接字示例，并希望使其在无限循环中与 parametrizedThread 一起工作。该代码在几个循环中运行良好，之后它会抛出

无法立即完成非阻塞套接字操作

对于我得到的上述内容，应该在第一个循环之后立即发生，而不是随机发生。这里有什么问题？听起来必须清除某些东西才能再次获得干净的连接（不确定）。

如果您使用多个线程通过套接字发送数据，建议使用 NetMQQueue 来同步这些请求。

我已将队列附加到轮询器，并且正在接收queue.ReceiveReady事件。但在这种情况下，我可以访问底层的 ConcurrentQueue 实例。但是我不确定如何处理这个队列，我应该只取出一个项目还是应该尝试清空它？

现在到我的第二个问题。我检查了队列实现，发现它只是在底层对套接字的发送方法周围使用了一个锁。那么我为什么要使用 NetMQQueue，如果我也可以简单地在我的发送方法周围使用锁，从而减少额外对套接字的开销？

我知道，按照文档，我应该为每个线程添加一个额外的 inproc-Router 套接字和一个 DEALER，以同步传出调用。但与自己处理同步相比，我有点害怕开销太大。我希望我的应用程序中有多达 20 个通信线程，双向异步通信。