正在停止读取MSMQ的服务

本文关键字：MSMQ 服务 读取 | 更新日期: 2023-09-27 18:29:50

我是一名Java程序员，被要求对C#应用程序进行一些更改。我已经使用C#一周了，我终于到了看文档没有帮助的地步，而且我在谷歌上找不到解决方案。

在这种情况下，我有一个Windows服务，它处理到达MSMQ的消息。当接收到一条消息时，当前正在侦听的线程会将其拾取并启动以执行一个需要几秒钟时间的操作。

public void Start()
{
    this.listen = true;
    for (int i = 0; i < Constants.ThreadMaxCount; i++)
    {
        ThreadPool.QueueUserWorkItem(new WaitCallback(this.StartListening), i);
    }
    ...
private void StartListening(Object threadContext)
{
    int threadId = (int)threadContext;
    threads[threadId] = Thread.CurrentThread;
    PostRequest postReq;
    while(this.listen)
    {
        System.Threading.Monitor.Enter(locker);
        try
        {
            postReq = GettingAMessage();
        }
        finally
        {
            System.Threading.Monitor.Exit(locker);
        }
    }
    ...
}

GettingAMessage（）具有以下侦听消息的行：

Task<Message> ts = Task.Factory.FromAsync<Message>
    (queue.BeginReceive(), queue.EndReceive);
ts.Wait();

问题是，当调用Stop（）方法并且没有消息进入MSMQ时，线程都会坐在那里等待消息。我尝试过使用超时，但这种方法对我来说似乎并不优雅（在切换到任务工厂后，我不确定目前如何实现它们）。我的解决方案是将每个线程的引用添加到数组中，这样我就可以取消它们。以下内容在创建后由每个工作线程调用。

threads[threadId] = Thread.CurrentThread;

然后被中止

public void Stop()
{
    try
    {
        this.listen = false;
        foreach(Thread a in threads) {
            a.Abort();
        }
    }
    catch
    {...}
}

关于为什么不关闭线程，有什么建议吗？（或者更好的是，有人能告诉我应该在哪里寻找如何正确取消ts.Wet（）吗？）

正在停止读取MSMQ的服务

使用ManualResetEvent类实现适当的&优雅地停止你正在运行的线程。

此外，不要将ThreadPool用于长时间运行的线程，而是使用您自己创建的线程，否则，对于大量长时间运行任务，您可能会导致线程池饥饿，甚至可能导致死锁：

public class MsmqListener
{
    privatec ManualResetEvent _stopRequested = new ManualResetEvent(false);
    private List<Thread> _listenerThreads;
    private object _locker = new _locker();
    //-----------------------------------------------------------------------------------------------------
    public MsmqListener
    {
        CreateListenerThreads();
    }
    //-----------------------------------------------------------------------------------------------------
    public void Start()
    {
      StartListenerThreads();
    }
    //-----------------------------------------------------------------------------------------------------
    public void Stop()
    {
        try
        {
            _stopRequested.Set();
            foreach(Thread thread in _listenerThreads)
            {
                thread.Join(); // Wait for all threads to complete gracefully
            }
        }
        catch( Exception ex)
        {...}
    }
    //-----------------------------------------------------------------------------------------------------
    private void StartListening()
    {
            while( !_stopRequested.WaitOne(0) ) // Blocks the current thread for 0 ms until the current WaitHandle receives a signal
            {
                lock( _locker )
                {
                    postReq = GettingAMessage();
                }
            ...
    }
    //-----------------------------------------------------------------------------------------------------
    private void CreateListenerThreads()
    {
        _listenerThreads = new List<Thread>();
        for (int i = 0; i < Constants.ThreadMaxCount; i++)
        {
            listenerThread = new Thread(StartListening);
            listenerThreads.Add(listenerThread);
        }
    }
    //-----------------------------------------------------------------------------------------------------
    private void StartListenerThreads()
    {
        foreach(var thread in _listenerThreads)
        {
            thread.Start();
        }
    }
}

更新：为了支持停止多个等待线程，我将AutoResetEvent与ManualResetEvent的使用进行了更改（使用ManualResetEvent，一旦您发出信号，所有等待线程都将得到通知，并且可以自由地继续他们的作业-在您的情况下，停止消息池）。

使用易失性bool并不能提供所有的保证。它可能仍然读取过时的数据。最好使用底层操作系统同步机制，因为它提供了更强的保证。来源：stackoverflow.com/a/1953661/952310

相关文章：