想要在 C# .net 中使用安全线程快速读取 msmq 消息

Question

我想读取 MSMQ，其中以字节为单位的队列数据以及在 1 分钟内生成的队列数量约为 1500。因此，如果连续读取，队列 CPU 会持续 30%。一段时间后它停止了。我需要长达 4 小时的大量读取队列.. 所以我希望以不应该被阻塞的方式安全地读取线程。 实际上我不擅长线程所以你能帮我吗..

目前我正在以这种方式阅读

    bool ProcessStatus; //process
    Thread _UDPthreadConsme;

    private void btn_receive_Click(object sender, EventArgs e)
    {

    if (MessageQueue.Exists(@".\private$\myquelocal"))
    {

    ThreadStart _processrcs = new ThreadStart(receivemessages);
    _UDPthreadConsme = new Thread(_processrcs);
    ProcessStatus = true;
    _UDPthreadConsme.Start();
    }
    }


    private void receivemessages()
    {
    MessageBox.Show("Start");
    while (ProcessStatus)
    {
    try
    {

    // Connect to the a queue on the local computer.
    MessageQueue myQueue = new MessageQueue(@".\private$\myquelocal");


    System.Messaging.Message[] myMessagecount = myQueue.GetAllMessages();

    if (myMessagecount.Length <= 0)
    return;


    myQueue.Formatter = new BinaryMessageFormatter();

    // Receive and format the message.
    System.Messaging.Message myMessage = myQueue.Receive();
    byte[] buffer = (byte[])myMessage.Body;

// here i convert buffer to its related structure and then insert the values in database sqlserver.

}
} 

Answer 1

我会像这样重写代码

    private void receivemessages()
    {
        Console.WriteLine("Start");

        MessageQueue myQueue = new MessageQueue(@".\private$\myquelocal");

        while (ProcessStatus)
        {
            try
            {
                // Waits 100 millisec for a message to appear in queue
                System.Messaging.Message msg = myQueue.Receive(new TimeSpan(0, 0, 0, 0, 100));

                // Take care of message and insert data into database

            }
            catch (MessageQueueException)
            {
                // Ignore the timeout exception and continue processing the queue

            }
        }
    }

Answer 2

这是在控制台上运行并异步读取队列的类的示例。这是最安全、最快的方法。但请注意，根据您运行此程序的位置，如果您正在执行诸如使用消息正文更新文本框或类似操作之类的操作，则仍然需要某种锁定机制。

public sealed class ConsoleSurrogate {

    MessageQueue _mq = null;

    public override void Main(string[] args) {

        _mq = new MessageQueue(@".\private$\my_queue", QueueAccessMode.Receive);
        _mq.ReceiveCompleted += new ReceiveCompletedEventHandler(_mq_ReceiveCompleted);
        _mq.Formatter = new ActiveXMessageFormatter();
        MessagePropertyFilter filter = new MessagePropertyFilter();
        filter.Label = true;
        filter.Body = true;
        filter.AppSpecific = true;
        _mq.MessageReadPropertyFilter = filter;
        this.DoReceive();

        Console.ReadLine();
        _mq.Close();
    }

    void _mq_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e) {
        _mq.EndReceive(e.AsyncResult);
        Console.WriteLine(e.Message.Body);
        this.DoReceive();
    }

    private void DoReceive() {
        _mq.BeginReceive();
    }
}