c# - 多线程MSMQ读取

Question

我创建了一个从 MSMQueue 读取消息的 Windows 服务，我需要并行执行此操作(两个线程应同时读取消息)。我怎样才能做到这一点？这是我的代码(与书上的差不多):

public partial class MyNewService : ServiceBase
    {
        System.Messaging.MessageQueue mq;
        System.Messaging.Message mes;

        public MyNewService()
        {
            InitializeComponent();

            if (MessageQueue.Exists("MyServer\\MyQueue"))
                mq = new System.Messaging.MessageQueue("MyServer\\MyQueue");

            mq.ReceiveCompleted += new ReceiveCompletedEventHandler(MyReceiveCompleted);
            mq.BeginReceive();

        }

        private static void MyReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
        {
           try
           {
                MessageQueue mq = (MessageQueue)source;
                Message m = mq.EndReceive(asyncResult.AsyncResult);

                // TODO: Process the m message here

                // Restart the asynchronous receive operation.
                mq.BeginReceive();
            }
            catch(MessageQueueException)
            {
             // Handle sources of MessageQueueException.
            }

            return; 
         }

}

这是我的主要功能:

static class Program
  {
    static void Main()
    {
      ServiceBase[] ServicesToRun;
      ServicesToRun = new ServiceBase[] 
            { 
                new MyNewService() 
            };
      ServiceBase.Run(ServicesToRun);
    }
  }

Answer 1

是否有理由不能只在多个线程上进行处理而不是在多个线程上出队？

这是一个非常的基本实现 - 它使用 ThreadPool 对项目进行排队，但是您随后依赖于 ThreadPool 的队列处理线程数和工作项数。根据许多其他因素，这可能不适合您的情况。

此外，请注意关于 SetMaxThreads 的备注部分 here .

private static void MyReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
{
   try
   {
       MessageQueue mq = (MessageQueue)source;
       Message m = mq.EndReceive(asyncResult.AsyncResult);

       // TODO: Process each message on a separate thread
       // This will immediately queue all items on the threadpool,
       // so there may be more threads spawned than you really want
       // Change how many items are allowed to process concurrently using ThreadPool.SetMaxThreads()
       System.Threading.ThreadPool.QueueUserWorkItem(new WaitCallback(doWork), m);


       // Restart the asynchronous receive operation.
       mq.BeginReceive();
   }
   catch(MessageQueueException)
   {
       // Handle sources of MessageQueueException.
   }

   return; 
}

private static void doWork(object message)
{
    // TODO: Actual implementation here.
}