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

Question

我阅读了ZeroMq guide 并偶然发现了以下内容：

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

及以后：

记住：除了创建它们的线程之外，不要使用或关闭套接字。

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

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

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

• 将eventhandler-invoking-thread同步到ZeroMQ-Socket创建的线程
• 使用线程安全的 ZeroMQ-Context 为事件处理程序中的线程创建一个新的 ZeroMQ-Socket / 获取现有的 ZeroMQ-Socket

似乎 0MQ-Guide 不鼓励第一个，我不认为为每个线程创建一个新的 ZeroMq-Socket 是高性能的/要走的路。

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

此外，该指南的作者在写作时是否考虑到了用于 .Net 的 ZeroMQ 绑定：

唯一的 远程理智的地方 线程之间的共享套接字在 需要做的语言绑定 像垃圾收集一样的魔法 插座。 ?

这里有一些示例代码来强调我的问题/疑问：

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. 在父线程中创建一个套接字并在线程创建时传递给附加线程。线程创建调用将执行完整的内存屏障。从那时起，在子线程中使用套接字only。 “使用”是指 recv、send、setsockopt、getsockopt 和 close。
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 网络层。