netMQ 4.0 多线程答案

【问题标题】：netMQ 4.0 multithreadingnetMQ 4.0 多线程
【发布时间】：2016-12-06 01:45:15
【问题描述】：

我对基于netMQ 4.0 的多线程服务器有一些问题。我尝试使用http://zguide.zeromq.org/cs:mtserver，但netMQ 4.0 上没有上下文。

我试过了：

for (var i = 0; i < workerCount; ++i)
{
    new Thread(() => Worker(connStr.Value)).Start();
}

//...
private void Worker(string connStr)
{
    using (var socket = new DealerSocket(connStr))
    {
        while (true)
        {
            var msg = socket.ReceiveMultipartMessage();
            //...
        }
    }
}

但我得到错误：

NetMQ.TerminatingException: CheckContextTerminated

是的，它被终止了。

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

【问题讨论】：

标签： c# .net multithreading zeromq netmq

【解决方案1】：

如果您使用的是.NET 4.0 或更高版本，则Thread 创建方法已过时，不应以这种方式使用 - 如果您的workerCount 足够高并且您没有提供任何调度程序逻辑上，您的性能可能会显着下降而不是受益。

您可以使用TPL 代替您的方法：

您可以轻松地将工作线程替换为 LongRunning tasks。
您可能应该为您的工作人员引入CancellationToken 以正确阻止他们。

所以你的代码可能是这样的：

/// field in your class
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

using (var clients = new RouterSocket(connStr.Value))
using (var workers = new DealerSocket())
{
    workers.Bind("inproc://workers");
    for (var i = 0; i < workerCount; ++i)
    {
        Task.Factory.StartNew(Worker
            , cancellationTokenSource.Token
            , TaskCreationOptions.LongRunning
            , TaskScheduler.Default);
    }
    var prx = new Proxy(clients, workers);
    prx.Start();
}

private void Worker()
{
    using (var socket = new ResponseSocket())
    {
        socket.Connect("inproc://workers");
        while (!cancellationTokenSource.Token.IsCancellationRequested)
        {
             //...
        }
        // Cancel the task and exit
        cancellationTokenSource.Token.ThrowIfCancellationRequested();
    }
}

为了简化它，您可以将 CancellationToken 作为参数传递给您的 Worker 方法。

【讨论】：

【解决方案2】：

正确的解决方案：

using (var clients = new RouterSocket(connStr.Value))
using (var workers = new DealerSocket())
    {
        workers.Bind("inproc://workers");
            for (var i = 0; i < workerCount; i++)
            {
                new Thread(Worker).Start();
            }
            var prx = new Proxy(clients, workers);
            prx.Start();
            }

private void Worker()
    {
        using (var socket = new ResponseSocket())
        {
            socket.Connect("inproc://workers");
            while (true)
            {
                 //...
            }
        }
    }

【讨论】：