如何从c#中的工作线程关闭ZMQQueueDevice
本文关键字:线程 ZMQQueueDevice 工作 | 更新日期: 2023-09-27 18:28:30
第一次使用ZMQ时,我正在尝试设置一个进程来处理许多getimage请求。当我调试时,会发生一些异常,我试图修复这些异常,并实现一种停止QueueDevice的方法,终止所有线程并正常退出。
- receiver.Connect(BackendBindAddress);throws在NetMQ.dll中发生类型为"NetMQ.InvalidException"的未处理异常,错误代码为NetMQ.zmq.ErrorCode.EINVAL。为什么此异常不停止线程的进一步执行
- 我曾尝试将QueueDevice设置为一个静态字段,并在关闭消息函数中使用QueueDevice.stop(),但随后线程开始抛出TerminatingExceptions并永不退出。那么我可以关闭所有线程和主线程吗
测试驾驶代码
[TestMethod]
public void ProgramStartupShutdownTest()
{
var mockClientWrapper = new Mock<IClient>(MockBehavior.Strict);
var target = new SocketListener(2, mockClientWrapper.Object);
var task = Task.Factory.StartNew(() => target.StartListening("tcp://localhost:81"));
using (var client = NetMQContext.Create())
{
var socket = client.CreateRequestSocket();
socket.Connect("tcp://localhost:81");
var m = new NetMQMessage(new ShutdownMessage().CreateMessageFrames());
socket.SendMessage(m);
}
var timedout = task.Wait(200);
Assert.IsTrue(timedout);
}
代码我正在使用
private const string BackendBindAddress = "inproc://workers";
public SocketListener(int numberOfWorkers, IClient client )
{
numberOfThreads = numberOfWorkers;
_client = client;
}
public void StartListening(string address)
{
StartZeroMQ(address, context =>
{
for (var i = 0; i <= numberOfThreads; i++)
{
var t = new Thread(WorkerRoutine);
t.Start(
new WorkerParamters
{
Context = context,
Client = _client
}
);
}
});
}
private void StartZeroMQ(string address, Action<NetMQContext> setupWorkers)
{
using (var context = NetMQContext.Create())
{
var queueDevice = new QueueDevice(context, address, BackendBindAddress, DeviceMode.Blocking);
setupWorkers(context);
queueDevice.Start();
}
}
struct WorkerParamters
{
public NetMQContext Context;
public IClient Client;
}
private static void WorkerRoutine(object startparameter)
{
var wp = (WorkerParamters) startparameter;
var client = wp.Client;
using (var receiver = wp.Context.CreateResponseSocket())
{
receiver.Connect(BackendBindAddress);
var running = true;
while (running)
{
var message = receiver.ReceiveMessage();
var letter = Message.ParseMessageFrame(message,
imageMessage => GetImage(imageMessage, client),
videoMessage => GetVideo(videoMessage, client),
shutdownMessage =>
{
running = false;
return true;
});
receiver.Send(letter.ToJson(), Encoding.Unicode);
}
}
}
为了解决这个问题,我向设备添加了一个Initialize方法,在启动工作程序之前调用它,在调用工作程序之后启动设备。
你可以从这里获取代码(你需要从存储库编译项目),我稍后会添加一个pull请求。
你也可以手工编写设备逻辑,它不应该很复杂。