问题标签 [netmq]

我正在尝试编写一个典型的股票交易程序，它从 netmq 接收股票代码/订单/交易，将流转换为 IObservable，并在 WPF 前端显示它们。我尝试将 async/await 与 NetMQ 阻塞 ReceiveString 一起使用（假设我期待一些字符串输入），以便 ReceiveString 循环不会阻塞主（UI）线程。由于我还是 C# 的新手，因此我在这篇文章中接受了 Dave Sexton 的回答：( https://social.msdn.microsoft.com/Forums/en-US/b0cf96b0-d23e-4461-9d2b-ca989be678dc/where -is-iasyncenumerable-in-the-lastest-release?forum=rx）并尝试编写一些这样的示例：

它无法使用“无法等待字符串”进行编译。虽然我知道它可能会期待一个任务，但我不太清楚如何完成整个事情。

再次包装：我想要实现的只是使用简单的阻塞 api 从 netmq 获取 IObservable 的ticker/orders/trades 流，但没有真正阻塞主线程。

我能用它做些什么吗？非常感谢。

我们有一个充当服务器的 Java 应用程序。客户端应用程序（用 C# 编写）使用 ZeroMQ 与它通信。我们（大部分）遵循懒惰的海盗模式。

服务器有一个路由器套接字，实现如下（使用 JeroMQ）：

客户端连接并发送如下消息：

当多个客户端同时发送消息时，我们会遇到消息丢失的情况。对于单个客户端，似乎没有任何问题。

我们是否以正确的方式实施多客户端单服务器设置？

更新：显示此行为的示例客户端和服务器：

服务器：

客户：

如果我运行服务器和单个客户端，我可以看到所有 100 条消息都到达。如果我同时运行 5 个客户端，我只会收到大约 200 -> 300 条消息到达，而不是全部 500 条。顺便说一句，关闭客户端中的套接字似乎会以某种方式停止服务器上的路由器套接字简短地接收消息，尽管这只是一个理论。

我一直在阅读有关 ZeroMQ 的更具体的 NetMQ 以及我看到的几乎每个 Pub/Sub 示例都用于绑定发布者套接字，然后订阅者套接字连接到另一个。

所以我想知道是否可以做相反的事情，我的意思是绑定订阅者套接字，然后发布者连接到它。

这可能吗 ？（我没有在文档中找到任何明确的内容）使用这种连接策略有什么缺点？

任何帮助都会很有用。

我有两个应用程序，一个 C++ 服务器和一个 C# WPF UI。C++ 代码通过 ZeroMQ 消息传递 [PUB/SUB] 服务接受请求（来自任何地方/任何人）。我使用我的 C# 代码进行回溯测试并创建“回溯测试”并执行它们。这些回溯测试可以由许多“单元测试”组成，每个测试都从 C++ 服务器发送/接收数千条消息。

目前，单个回溯测试运行良好，可以发送 N 个单元测试，每个测试包含数千个请求和捕获。我的问题是建筑；当我发送另一个回测（在第一个测试之后）时，由于轮询线程没有被取消和处置，我遇到了第二次完成事件订阅的问题。这会导致错误的输出。这似乎是一个微不足道的问题（也许对你们中的一些人来说），但是在我当前的配置下取消这个轮询任务是很麻烦的。一些代码...

我的消息代理类很简单，看起来像

回测“引擎”用于执行每个回测，首先构造一个Dictionary包含每个Test（单元测试）和消息以分派给每个测试的 C++ 应用程序。

DispatchTests方法，来了

最后的PING消息，它告诉 C++ 我们已经完成了。然后我们强制等待，以便在从 C++ 代码接收到所有返回之前不会调度下一个 [unit] 测试 - 我们使用ManualResetEvent.

当 C++ 收到 PING 消息时，它会直接返回消息。我们通过处理接收到的消息OnMessageRecieved，PING 告诉我们设置，ManualResetEvent.Set()以便我们可以继续单元测试；“下一位”...

我的问题是，broker.Dispose()在 finally 上面永远不会被击中。我很欣赏在后台线程上执行的 finally 块不能保证被执行。

上面划掉的文字是因为我弄乱了代码；在孩子完成之前，我正在停止父线程。但是，仍然存在问题...

现在broker.Dispose()被正确调用，并且broker.Dispose()被调用，在这个方法中我尝试取消轮询线程并Task正确处理以避免任何多个订阅。

要取消线程，我使用该CancelPolling()方法

但在StartPolling()方法中

ThrowIfCancellationRequested()永远不会被调用并且线程永远不会被取消，因此永远不会被正确处理。轮询线程被该subSocket.Receive()方法阻塞。

现在，我不清楚如何实现我想要的，我需要在用于轮询消息之外的线程上调用broker.Dispose()/以及如何强制取消。PollerCancel()线程中止不是我想不惜一切代价进入的。

本质上，我想broker在执行下一个回溯测试之前正确处理，如何正确处理这个问题，拆分轮询并在单独的应用程序域中运行它？

我已经尝试过，在处理程序中进行OnMessageRecived处理，但这显然是在与轮询器相同的线程上执行的，并且不是这样做的方法，没有调用其他线程，它会阻塞。

实现我想要的最好方法是什么，我可以遵循这种情况的模式吗？

谢谢你的时间。

我是 ZeroMQ（以及一般网络）的新手，并且有一个关于在多个客户端连接到单个服务器的设置中使用 ZeroMQ 的问题。我的情况如下：

--1服务器 --
多个客户端

--客户端向服务器发送消息：我已经想出了如何做这部分。
--服务器向特定客户端发送消息：这是我遇到问题的部分。当某些事件在服务器上得到处理时，服务器将需要向特定客户端发送消息——而不是所有客户端。换句话说，服务器需要能够选择将给定消息发送到哪个客户端。

现在，这是我的服务器代码：

我有一种感觉，问题在于我的设计是错误的，并且该ResponseSocket类型并不意味着以我想要的方式使用它。由于我是新手，因此非常感谢任何建议！

我正在尝试决定使用哪种技术将我的实时市场数据广播到我的交易策略中。场景如下，

我有多个交易策略订阅不同的代码，当收到某个代码的市场数据时，我会将这个代码的市场数据广播到策略。监听交易策略将根据收到的市场数据采取行动。

我有两种类型的实现，

1) 是使用类似Messenger提供的类的事件总线mvvmlight，

2) 另一种方法是使用消息队列系统，例如ZeroMQ( NetMQ)

注意市场数据api和交易策略都在同一台机器上，所以不涉及网络。

哪个更适合我的情况？

我想从客户端向服务器发送数据。只有最后一条消息对服务器很重要。如果服务器在失败后启动，我只希望服务器从客户端获取最后一条消息。

当服务器关闭时，我希望客户端继续处理和发送消息，或者至少将它们放入队列中（一条消息的长度）。

我尝试为此使用 NetMQ/ZeroMQ。怎么做到呢？

谢谢！

我正在尝试使用 netmq（zeroMQ 的端口）。这是我发现的问题。这是一个代码：

如果我们运行此代码并在控制台中观察，我们会看到一些消息丢失了。喜欢：

正如我们所见，哪里没有来自 的消息4211d77a-ad9f-40f1-9382-121156325128，bd735e75-2692-4abe-b8b1-fbddbe21e546还有另一个消息。是多线程的问题吗？还是我做错了什么？谢谢。

有人知道 NetMQ 是否与 ZMTP 1.0 兼容吗？我找不到有关支持哪些版本的任何信息。