0MQ：如何以线程安全的方式使用ZeroMQ？

Question

我阅读了 ZeroMq 指南，我偶然发现了以下内容：

您不得在之间共享 ØMQ 套接字 线程。 ØMQ套接字不是 线程安全。技术上是可以的 做到这一点，但它需要信号量， 锁或互斥锁。这将使您的 应用程序缓慢且脆弱。唯一的 一个远程理智的地方 线程之间共享套接字位于 需要做的语言绑定 像垃圾收集一样神奇 套接字。

以及稍后：

记住：不要使用或关闭套接字，除非在创建它们的线程中。

我还了解 ZeroMQ Context 是线程安全的。

如果一个类注册另一个类的事件，在 .Net 中，该事件可能会从与创建侦听器的线程不同的线程调用。

我认为只有两个选项可以通过 ZeroMQ-Sockets 从事件处理程序中分派某些内容：

• 将事件处理程序调用线程同步到在
• 创建新 ZeroMQ 中创建 ZeroMQ-Socket 的线程-Socket / 通过使用线程安全的 ZeroMQ-Context 获取事件处理程序中线程的现有 ZeroMQ-Socket

0MQ-Guide 阻止第一个，我不认为为每个线程创建一个新的 ZeroMq-Socket 是高性能/可行的方法。

我的问题：
从事件处理程序中通过 0MQ 发布消息的正确模式（本来的方式）是什么？

另外，该指南的作者在写下以下内容时是否考虑到了 .Net 的 ZeroMQ-Binding：

唯一的 一个远程理智的地方 线程之间共享套接字位于 需要做的语言绑定 像垃圾收集一样神奇 插座。 ？

这是一些示例代码来强调我的问题：

public class ExampleClass
{
    public event EventHandler<ByteEventArgs> SomethinIsCalledFromAnotherThread;
}

public class ByteEventArgs : EventArgs
{
    public byte[] BytesToSend;
}


public class Dispatcher
{
    ZMQ.Context ctx;

    public Dispatcher(ZMQ.Context mqcontext, ExampleClass exampleClassInstance)
    {
        this.ctx = mqcontext;
        exampleClassInstance.SomethinIsCalledFromAnotherThread += new EventHandler<ByteEventArgs>(exampleClass_SomethinIsCalledFromAnotherThread);
    }

    void exampleClass_SomethinIsCalledFromAnotherThread(object sender, ByteEventArgs e)
    {
        // this method might be called by a different thread. So I have to get a new socket etc?
        using (var socket = ctx.Socket(ZMQ.SocketType.PUSH))
        {
            // init socket etc..... and finally: 
            socket.Send(e.BytesToSend);
        }
        // isn't that too much overhead?
    }
}

Answer 1

您可以创建大量 0MQ 套接字，当然数量与线程数量一样多。如果您在一个线程中创建套接字，并在另一个线程中使用它，则必须在两个操作之间执行完整的内存屏障。任何其他情况都会导致 libzmq 中奇怪的随机失败，因为套接字对象不是线程安全的。

有一些传统模式，但我不知道它们如何专门映射到 .NET：

1. 在使用它们的线程中创建套接字，期间。在紧密绑定到一个进程的线程之间共享上下文，并在不紧密绑定的线程中创建单独的内容。在高级 C API (czmq) 中，这些线程称为附加线程和分离线程。
2. 在父线程中创建套接字，并在线程创建时将套接字传递给附加线程。线程创建调用将执行完整的内存屏障。从那时起，仅在子线程中使用套接字。 “使用”意味着接收、发送、setsockopt、getsockopt 和关闭。
3. 在一个线程中创建一个套接字，然后在另一个线程中使用，在每次使用之间执行您自己的完整内存屏障。这是非常微妙的，如果您不知道什么是“完整内存屏障”，则不应该这样做。

Answer 2

在.net Framework v4及更高版本中，您可以使用并发收集来解决这个问题。即生产者-消费者模式。多个线程（处理程序）可以将数据发送到线程安全队列，只有单个线程使用队列中的数据并使用套接字发送数据。

这个想法是这样的：

sendQueue = new BlockingCollection<MyStuff>(new ConcurrentQueue<MyStuff>());
// concurrent queue can accept from multiple threads/handlers safely
MyHandler += (MyStuff stuffToSend) => sendQueue.Add(stuffToSend);

// start single-threaded data send loop
Task.Factory.StartNew(() => {
    using(var socket = context.Socket()) {
        MyStuff stuffToSend;
        // this enumerable will be blocking until CompleteAdding is called
        foreach(var stuff in sendQueue.GetConsumingEnumerable())
            socket.Send(stuff.Serialize());
    }
});

// break out of the send loop when done
OnMyAppExit += sendQueue.CompleteAdding;

Answer 3

不要忘记查看 inproc 传输。使用 inproc:// 套接字进行线程间通信并让一个线程打开套接字与其他进程/服务器通信可能会很有用。

每个线程仍然需要至少一个套接字，但 inproc 根本不涉及 IP 网络层。