c# - ZeroMQ (NetMQ) TCP 传输可以在同一进程中的发布者和订阅者之间使用吗？

Question

我认为这个问题确实说明了一切。

关于一些背景知识，我有一个订阅者，我正在尝试为其编写一些测试。为此，我在测试中启动了一个发布者，指定了 tcp://localhost:[port] 作为地址。发送消息时，订阅者不会收到它。下面是一些示例代码来演示:

   string address   =    "tcp://localhost:1026";
// string address   = "inproc://localhost1026";

   var    pubSocket =  new PublisherSocket();
   pubSocket.Bind(address);

   var subSocket = new SubscriberSocket();
   subSocket.Connect(address);
   subSocket.SubscribeToAnyTopic();

   pubSocket.SendFrame("Hello world!", false);

   Console.WriteLine(subSocket.ReceiveFrameString()); /* <-- tcp transport 
                                                         waits here forever
                                                         */
   subSocket.Dispose();
   pubSocket.Dispose();

如果我将协议(protocol)更改为 inproc:// 那么一切都很好。但是，我不想在我的测试中这样做，因为我还想测试一个监视器套接字，这不会为 inproc:// 连接引发事件(据我所知)。

请注意，我在 C# 代码中使用 NetMQ(在 .NET Framework 4.6.2 下运行)。

Answer 1

Can ZeroMQ (NetMQ) TCP transport be used between publisher and subscriber in the same process ?

当然，
没有任何限制可以阻止人们这样做，但是。 . .

您的观察与隐藏处理的延迟有关，它发生在主引擎内......在 ZeroMQ Context() 实例内。

事情不会在零时间发生。

好吧，在 Pieter HINTJENS 的“Code Connected，第 1 卷”关于 ZeroMQ Zen-of-Zero 之后，您可能会选择推迟这一篇。这两者都有道理，很有道理，不仅在这里 )。

因此，虽然 inproc:// 传输类具有(几乎)-零资源，但它是一个纯私有(private)的-"进程内”内存映射抽象(当然，除了一些 sub-[ns]-“设备”，如信号量/锁)，ZeroMQ 基础设施启动并且以“立即”方式运行，tcp:// 没有这种舒适感，因为它必须首先生成与 O 的所有传输类特定契约(Contract)/S，使用设备驱动程序，实例化传输类特定的 ISO/OSI-{ L0, L1, L3, L3+ }- 处理策略，将相应的数据泵代码实例化到 Context() 的 RTO 状态，分配和映射内存缓冲区来满足这些目的，因此在 PUB-side 进入之前，还有很多工作要做RTO 状态，它(在 ~ API 4.+ 的较新版本下)也有责任接收 和 处理订阅服务遥测，因为它承担了每个 的集中责任strong>SUB-client TOPIC-filterlist处理。

这就是 为什么 它导致挂起 .recv( ..., ZMQ_BLOCK ) 埋在 的 NetMQ 包装抽象中subSocket.ReceiveFrameString().

要测试它，只需进行修改后的测试:

  // --------------------------------------------------- // DEMO PSEUDO-CODE
     string rF = "";

     while True:
         
           pubSocket.SendFrame( "Hello world!", false ); // keep sending ...
                                                         // also may count++
                                                         // so as to "show" how
                                                         // many loops it took
           rF = subSocket.ReceiveFrameString(   false ); // non-blocking mode ~
                                                         // .recv(  ZMQ_NOBLOCK )
                                                         // or may use Poll()
                                                         // to just sniff for
                                                         // a message presence
           if  rF == "":
               continue; // ---^ LOOP NEXT, AS DID .recv() NOTHING YET
           break; // ----------v BREAK      AS DID .recv() MESSAGE
// ----------------------------------------------------------------------
   Console.WriteLine( rF );

人们可能会在这里进行更多的努力和实验，以了解 .connect() 相关开销的关键作用以及不要错过 < strong>SUB-设置订阅的端信号遥测 + 接收它的需要 + 在 PUB 上重新处理它>-side，在任何消息第一次被发送到预期之前，否则只是“永远”-等待，SUB

---

足够“胖”的 .sleep( someGuestimateTIME ) 已由 @HesamFaridmehr 提出，在 SUB-side 有两个 .connect()-ed 加上它是 .setsockopt( ZMQ_SUBSCRIBE ) (必须首先交付并在 PUB 端以适当的方式进行处理(以正确配置 TOPIC-filter list-processor 处理器))all在第一个 PUB.send() 将通过使其“间接”阻塞并停止代码执行流来掩盖根本原因之前就足够了, 而不是让解决方案足够智能 - 例如，使用非阻塞形式的 Poll() - 对于专业人士 distributed-system设计最佳实践，其中人们确实只能服从汇编黑客钟爱的第一行当前宏 #ASSUME NOTHING;。