MSMQ异步读写方法

Question

我开始使用 MSMQ。我发现这种方法非常快，对于需要处理数百万个请求的解决方案，我编写了一个性能测试并看到：它仅在 43 秒内写入 1000000 个对象（使用我的电脑硬件配置）！我喜欢。所以，我需要读写数百万条消息，性能对我来说非常重要。我使用线程来提高速度和性能，我错了吗？这是正确的吗？这是我的代码：

public static void WriteToQueue(List<Human> humans)
    {
        new Thread(() =>
        {
            var queue = new Queue().Prepare(queueName);
            Logger.LogInfo("Started writing to queue");
            foreach (var human in humans)
            {
                queue.Add(human.JsonSerialize());
            }
        }).Start();
    }

private static void ReadFromQueue()
    {
        new Thread(() =>
        {
            var queue = new Queue().Prepare(queueName);
            var receivedCount = 0;
            while (receivedCount <= itemsCount)
            {
                var human = queue.Pop().JsonDeserialize<Human>();
                receivedCount++;
                if (receivedCount % 10000 == 0)
                {
                    Logger.Count(receivedCount);
                }
                if (receivedCount == itemsCount)
                {
                    watch.Stop();
                    Logger.LogInfo("Took {0} to write/read {1} times to/from queue".Fill(watch.ElapsedMilliseconds, itemsCount));
                }
            }
            Queue.Delete(queueName);
        }).Start();
    }

我用它们:

        var humans = Enumerable.Range(0, itemsCount).Select(i => new Human
        {
            FirstName = "first name" + i,
            LastName = "last name" + i
        }).ToList();

        watch.Start();

        WriteToQueue(humans);

        ReadFromQueue();

        Console.ReadLine();

好的，这是我的第一个解决方案（使用线程）。我发现异步编程是最好的方法（我错了吗？），我发现异步编程是更好的方法！因为此方法（读取和写入 msmq）是 IO 绑定的（等待 io 操作）！我提出了许多等待 io 的线程！我是异步编程的新手，任何人都可以帮助我并为我编写一些代码来创建这些方法（读取和写入）与异步编程一起使用吗？而且，最好的方法是什么？（parallel.foreach - 异步 TAP - 异步 APM - 异步 EAP -线程）

Answer 1

我找到了更好的方法，我喜欢分享它。线程很好，最好的方法取决于您的状态和您的业务，对我来说，我每天需要使用 msmq 处理大约 2 亿条消息！所以，我得到了这个结果：读取和写入 10000 条消息可以节省大约 300 毫秒！您可以节省大约 6 秒的时间来读取和写入 100 万条消息！读写1000万条消息可节省60秒左右！

class MsmqListener
{
   public void StartListening(string queuePath)
    {
        MessageQueue msQueue = new MessageQueue(@".\private$\" + queuePath);
        msQueue.ReceiveCompleted += QueueMessageReceived;
        msQueue.BeginReceive();
        // you can do averything here, its async
    }

    private void QueueMessageReceived(object source, ReceiveCompletedEventArgs args)
    {
        MessageQueue msQueue = (MessageQueue)source;

        //once a message is received, stop receiving
        Message msMessage = null;
        msMessage = msQueue.EndReceive(args.AsyncResult);

        //do something with the message
        //...

        //begin receiving again
        msQueue.BeginReceive();
    }
}

此方法（beginRecieve 和 endRecieve）是 msmq 的异步方法。

MSMQ异步读写方法

MSMQ read and write Methods with Asynchronous

c#

multithreading

asynchronous

msmq