So veröffentlichen Sie einen Socket mit mehreren Threads in Nulomq ⇐ C#

[Guest]

Ich verwende NetMQ, Zeromqs C# -Bibliothek und ich habe sie wie folgt implementiert. Wissen Sie, wie man diese Situation löst und warum ich nicht mit mehreren Threads in einer Socket in Zeromq arbeiten kann. < /p>

Code: Select all

public class NetMqManager
{
private NetMQQueue _queue = new NetMQQueue();

private readonly Thread _subThread;

private readonly Thread _pubThread;

private readonly SubscriberSocket _subscriber;

private readonly PublisherSocket _publisher;

private readonly ZeroMqEndPoint _endPoint;

public NetMqManager(ZeroMqEndPoint endPoint)
{
_endPoint = endPoint;

_publisher = new PublisherSocket();
_publisher.Options.SendHighWatermark = 1500;
_publisher.SendReady += Publisher_SendReady;
_pubThread = new Thread(() =>
{
var poller = new NetMQPoller { _publisher };
poller.Run();
});

_subscriber = new SubscriberSocket();
_subscriber.Options.SendHighWatermark = 1500;
_subscriber.ReceiveReady += Subscriber_ReceiveReady;
_subThread = new Thread(() =>
{
var poller = new NetMQPoller { _subscriber };
poller.Run();
});
}

public async Task RunAsync()
{
await Task.Run(() =>
{
_publisher.Bind($"tcp://*:{_endPoint.PubPort}");
_subscriber.Bind($"tcp://*:{_endPoint.SubPort}");

_pubThread.Start();
_subThread.Start();
});
}

public async void PublishAsync(string topic, string payload)
{
await Task.Run(() => _queue.Enqueue((topic, payload)));
}

public async void SubscribeAsync(string topic)
{
_subscriber.Subscribe(topic);
}

private void Publisher_SendReady(object? sender, NetMQSocketEventArgs e)
{
var (topic, payload) = _queue.Dequeue();
_publisher.SendMoreFrame(topic).SendFrame(payload);
}

private void Subscriber_ReceiveReady(object? sender, NetMQSocketEventArgs e)
{
var topic = e.Socket.ReceiveFrameString();
var payload = e.Socket.ReceiveFrameString();

Console.WriteLine($"Topic: {topic}, Payload: {payload}");
}
}