NetMQ多线程的最新示例

Question

我正在尝试创建一个单服务器/多客户端NetMQ测试程序(我认为这应该使用Router-Dealer模式)，它将允许服务器使用独立线程上的工作者处理唯一的客户端请求(并分别回复每个客户端)，但我找不到一个适用于当前nuget版本(撰写本文时的版本为4.0.1.5)的工作示例。

每个客户端只需要发送一个请求(包含它的Id，因为它是一个交易商套接字)，并从服务器获得它的专用响应。服务器应该有一个线程池(IOCP样式)，它将消费消息和发送回响应。

我在NetMQ/Samples github中找到了Multithreaded example，这似乎就是我要找的，但它使用QueueDevice来连接客户端和工作者，而且这个类似乎在某个时候被删除了：

var queue = new QueueDevice(
     "tcp://localhost:5555", 
     "tcp://localhost:5556", 
     DeviceMode.Threaded);

然后是netmq.readthedocs.io的Router-dealer example，它使用NetMQPoller，但不编译，因为它调用了不存在的RouterSocket.ReceiveMessage()和NetMQSocket.Receive(out bool hasmore)方法，这两个方法已被删除：

public static void Main(string[] args)
{
    // NOTES
    // 1. Use ThreadLocal<DealerSocket> where each thread has
    //    its own client DealerSocket to talk to server
    // 2. Each thread can send using it own socket
    // 3. Each thread socket is added to poller

    const int delay = 3000; // millis
    var clientSocketPerThread = new ThreadLocal<DealerSocket>();
    using (var server = new RouterSocket("@tcp://127.0.0.1:5556"))
    using (var poller = new NetMQPoller())
    {
        // Start some threads, each with its own DealerSocket
        // to talk to the server socket. Creates lots of sockets,
        // but no nasty race conditions no shared state, each
        // thread has its own socket, happy days.

        for (int i = 0; i < 3; i++)
        {
            Task.Factory.StartNew(state =>
            {
                DealerSocket client = null;
                if (!clientSocketPerThread.IsValueCreated)
                {
                    client = new DealerSocket();
                    client.Options.Identity =
                        Encoding.Unicode.GetBytes(state.ToString());
                    client.Connect("tcp://127.0.0.1:5556");
                    client.ReceiveReady += Client_ReceiveReady;
                    clientSocketPerThread.Value = client;
                    poller.Add(client);
                }
                else
                {
                    client = clientSocketPerThread.Value;
                }
                while (true)
                {
                    var messageToServer = new NetMQMessage();
                    messageToServer.AppendEmptyFrame();
                    messageToServer.Append(state.ToString());
                    PrintFrames("Client Sending", messageToServer);
                    client.SendMultipartMessage(messageToServer);
                    Thread.Sleep(delay);
                }
            }, string.Format("client {0}", i), TaskCreationOptions.LongRunning);
        }

        // start the poller
        poller.RunAsync();

        // server loop
        while (true)
        {
            var clientMessage = server.ReceiveMessage();
            PrintFrames("Server receiving", clientMessage);
            if (clientMessage.FrameCount == 3)
            {
                var clientAddress = clientMessage[0];
                var clientOriginalMessage = clientMessage[2].ConvertToString();
                string response = string.Format("{0} back from server {1}",
                    clientOriginalMessage, DateTime.Now.ToLongTimeString());
                var messageToClient = new NetMQMessage();
                messageToClient.Append(clientAddress);
                messageToClient.AppendEmptyFrame();
                messageToClient.Append(response);
                server.SendMultipartMessage(messageToClient);
            }
        }
    }
}

static void PrintFrames(string operationType, NetMQMessage message)
{
    for (int i = 0; i < message.FrameCount; i++)
    {
        Console.WriteLine("{0} Socket : Frame[{1}] = {2}", operationType, i,
            message[i].ConvertToString());
    }
}

static void Client_ReceiveReady(object sender, NetMQSocketEventArgs e)
{
    bool hasmore = false;
    e.Socket.Receive(out hasmore);
    if (hasmore)
    {
        string result = e.Socket.ReceiveFrameString(out hasmore);
        Console.WriteLine("REPLY {0}", result);
    }
}

也许这些问题很容易修复，但我是一个NetMQ新手，我只想为C#工作获得一个适当的“多线程Hello World”。

NetMQ API与官方0MQ文档中的稍有不同(它更习惯于使用C#，并使用新的低分配.NET构造)，因此我很难理解如何正确移植原始的C示例。

有没有人可以给我推荐一个包含此场景的最新示例的参考资料，或者帮助我了解使用新的NetMQ应用程序接口的正确方法？

或者这个库被弃用了，我应该使用clrzmq4 .NET包装器？据我所知，NetMQ在代码库中使用了Span<T>和ref结构，所以它看起来应该是高性能的。

Answer 1

我已经在netmq.readthedocs.io上用router-dealer example修复了这几个问题，现在代码可以正常工作了。我在这里发布了它，也许它会有用(也许@somdoron或者其他人可以查看它并更新文档)。

这个示例并没有完全实现我想要的(拥有一个公平地获得工作的工作者池)，我相信我必须实现单独的代理线程。

using NetMQ;
using NetMQ.Sockets;

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

internal static class Program
{
    public static void Main(string[] args)
    {
        InitLogger();

        const int NumClients = 5;
        Log($"App started with {NumClients} clients\r\n");

        using (var server = new RouterSocket("@tcp://127.0.0.1:5556"))
        using (var poller = new NetMQPoller())
        {
            // Start some threads, each with its own DealerSocket
            // to talk to the server socket. Creates lots of sockets,
            // but no nasty race conditions no shared state, each
            // thread has its own socket, happy days.

            for (int i = 0; i < NumClients; i++)
            {
                var clientNo = i + 1;
                Task.Factory.StartNew(async () =>
                {
                    var rnd = new Random(31 + clientNo * 57);
                    var clientId = $"C{clientNo}";
                    DealerSocket client = new DealerSocket();

                    client.Options.Identity = Encoding.ASCII.GetBytes(clientId);
                    client.Connect("tcp://127.0.0.1:5556");
                    client.ReceiveReady += (sender, e) => 
                    {
                        var msg = e.Socket.ReceiveMultipartMessage(3);
                        var clientid = Encoding.ASCII.GetString(e.Socket.Options.Identity);
                        Log($"Client '{clientid}' received <- server", msg);
                    };
                    poller.Add(client);

                    while (true)
                    {
                        var messageToServer = new NetMQMessage();
                        messageToServer.Append(NetMQFrame.Empty);
                        messageToServer.Append($"Some data ({clientId}|{rnd.Next():X8})", Encoding.ASCII);

                        Log($"Client {clientId} sending -> server: ", messageToServer);
                        client.SendMultipartMessage(messageToServer);

                        await Task.Delay(3000);
                    }
                }, TaskCreationOptions.LongRunning);
            }

            // start the poller
            poller.RunAsync();

            // server loop

            while (true)
            {
                var clientMessage = server.ReceiveMultipartMessage();

                if (clientMessage.FrameCount < 1)
                {
                    Log("Server received invalid frame count");
                    continue;
                }

                var clientid = clientMessage[0];
                Log($"Server received <- {clientid.ConvertToString()}: ", clientMessage);
                
                var msg = clientMessage.Last().ConvertToString(Encoding.ASCII);
                string response = $"Replying to '{msg}'";

                {
                    var messageToClient = new NetMQMessage();
                    messageToClient.Append(clientid);
                    messageToClient.Append(NetMQFrame.Empty);
                    messageToClient.Append(response, Encoding.ASCII);
                    Log($"Server sending -> {clientid.ConvertToString()}: {response}");
                    server.SendMultipartMessage(messageToClient);
                }

            }
        }
    }

    #region Poor man's console logging 

    static BlockingCollection<string> _logQueue;

    // log using a blocking collection
    private static void InitLogger()
    {
        _logQueue = new BlockingCollection<string>();
        Task.Factory.StartNew(() =>
        {
            foreach (var msg in _logQueue.GetConsumingEnumerable())
            {
                Console.WriteLine(msg);
            }
        });
    }
    
    // thread id, timestamp, message
    static void Log(string msg)
    {
        var thid = Thread.CurrentThread.ManagedThreadId;
        var time = GetTimestamp().ToString("HH:mm:ss.fff");
        _logQueue.Add($"[T{thid:00}] {time}: {msg}");
    }

    // log all frames in a message
    static void Log(string operationType, NetMQMessage message)
    {
        var frames = string.Join(", ", message.Select((m, i) => $"F[{i}]={{{m.ConvertToString(Encoding.ASCII)}}}"));
        Log($"{operationType} {message.FrameCount} frames: " + frames);
    }

    // if possible, use win10 high precision timestamps
    static DateTime GetTimestamp()
    {
        if (Environment.OSVersion.Platform == PlatformID.Win32NT && Environment.OSVersion.Version.Major >= 10) // win 10
        {
            long fileTime;
            WinApi.GetSystemTimePreciseAsFileTime(out fileTime);
            return DateTime.FromFileTimeUtc(fileTime);
        }

        return DateTime.UtcNow;
    }

    static class WinApi
    {
        [DllImport("Kernel32.dll", CallingConvention = CallingConvention.Winapi)]
        internal static extern void GetSystemTimePreciseAsFileTime(out long filetime);
    }

    #endregion
}