NetMQ多线程的最新示例

Question

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

每个客户端只需要发送一个请求(其中包含其Id，因为它是经销商套接字(，并从服务器获得其专用响应。服务器应该有一个线程池(IOCP风格(，它将使用消息并发回响应。

我在NetMQ/Samples github中找到了多线程示例，这似乎是我想要的，但它使用QueueDevice将客户端与工作线程连接起来，并且这个类似乎在某个时候被删除了：

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

然后是netmq.readthedocs.io上的路由器经销商示例，它使用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新手，我只想获得一个合适的"；多线程Hello World"；用于C#工作。

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

有人能给我指一个关于这个场景的最新示例的资源吗？或者帮助我了解使用新的NetMQ API的正确方法是什么？

或者这个库已被弃用，我应该使用clrzmq4.NET包装器？根据我所看到的，NetMQ在代码库中使用了Span<T>和ref结构，因此它似乎应该具有性能。

Answer 1

我已经修复了netmq.readthedocs.io上路由器经销商示例的几个问题，现在代码可以工作了。我把它发布在这里，也许它会很有用(也许@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} clientsrn");
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
}

Answer 2

我最近完成了一个相当好的pub子模型示例，它支持多个订阅者(许多任务(。由于您似乎正在使用C#，该代码还演示了如何使用CancellationToken在请求时退出所有任务；如果您正在编写NET核心服务工作者，这将有所帮助。我不是NetMQ方面的专家，但这是我学习少量内容的机会。

paultechguy/NetMQPubSubExample