在 .NET ActiveMQ 中异步发送答案

【问题标题】：Async sends in .NET ActiveMQ在 .NET ActiveMQ 中异步发送
【发布时间】：2012-08-03 17:12:50
【问题描述】：

我希望提高针对 ActiveMQ 编写的高吞吐量生产者的性能，根据this useAsyncSend 将：

强制使用异步发送，从而大幅提升性能；但是意味着 send() 方法将立即返回是否消息已发送或未发送，可能导致消息丢失。

但是我看不出它对我的简单测试用例有什么影响。

使用这个非常基本的应用程序：

const string QueueName = "....";
const string Uri = "....";

static readonly Stopwatch TotalRuntime = new Stopwatch();

static void Main(string[] args)
{
    TotalRuntime.Start();
    SendMessage();
    Console.ReadLine();
}

static void SendMessage()
{
    var session = CreateSession();
    var destination = session.GetQueue(QueueName);
    var producer = session.CreateProducer(destination);

    Console.WriteLine("Ready to send 700 messages");
    Console.ReadLine();

    var body = new byte[600*1024];

    Parallel.For(0, 700, i => SendMessage(producer, i, body, session));         
}

static void SendMessage(IMessageProducer producer, int i, byte[] body, ISession session)
{
     var message = session.CreateBytesMessage(body);

     var sw = new Stopwatch();
     sw.Start();
     producer.Send(message);
     sw.Stop();

     Console.WriteLine("Running for {0}ms: Sent message {1} blocked for {2}ms", 
            TotalRuntime.ElapsedMilliseconds, 
            i, 
            sw.ElapsedMilliseconds);
}       

static ISession CreateSession()
{
     var connectionFactory = new ConnectionFactory(Uri)
                                    {
                                        AsyncSend = true,
                                        CopyMessageOnSend = false
                                    };
     var connection = connectionFactory.CreateConnection();
     connection.Start();
     var session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
     return session;
}

我得到以下输出：

Ready to send 700 messages

Running for 2430ms: Sent message 696 blocked for 12ms
Running for 4275ms: Sent message 348 blocked for 1858ms
Running for 5106ms: Sent message 609 blocked for 2689ms
Running for 5924ms: Sent message 1 blocked for 2535ms
Running for 6749ms: Sent message 88 blocked for 1860ms
Running for 7537ms: Sent message 610 blocked for 2429ms
Running for 8340ms: Sent message 175 blocked for 2451ms
Running for 9163ms: Sent message 89 blocked for 2413ms
.....

这表明每条消息发送大约需要 800 毫秒，并且对 session.Send() 的调用会阻塞大约两秒半。即使文档说

"send() 方法将立即返回"

如果我将并行 for 更改为普通 for 循环或将 AsyncSend = true 更改为 AlwaysSyncSend = true，这些数字也基本相同，所以我不相信异步开关在所有工作......

谁能看到我在这里缺少什么以使发送异步？

经过进一步测试：

根据 ANTS 性能分析器，运行时的大部分时间都花在等待同步上。看来问题在于各种传输类通过监视器在内部阻塞。特别是，我似乎对MutexTransport 的 OneWay 方法很感兴趣，该方法一次只允许一个线程访问它。

看起来对 Send 的调用会阻塞，直到上一条消息完成，这解释了为什么我的输出显示第一条消息阻塞了 12 毫秒，而下一条消息阻塞了 1858 毫秒。我可以通过实现每条消息连接模式来进行多种传输，这种模式可以改善问题并使消息发送并行工作，但大大增加了发送单个消息的时间，并且消耗了如此多的资源，看起来不像正确的解决方案。

我已经用 1.5.6 重新测试了所有这些，没有发现任何差异。

【问题讨论】：

这里缺少一些关键信息，NMS.ActiveMQ 的版本是什么？您传递以连接到代理的实际 URI 是什么？代理是否启用了任何生产者流控制？
@Tim 好点，对不起。我运行的是 1.5.1 版本，URI 是一个内部服务器，但格式为“tcp://...”，而且我对代理的设置方式一无所知 - 我可以访问控制台网页，但这似乎并没有暴露该信息。进一步的调查使我相信问题出在我发送的消息的大小（600Kb）上，但我仍在调查它，因此我们将不胜感激。

标签： performance asynchronous activemq blocking throughput

【解决方案1】：

一如既往，最好的办法是更新到最新版本（撰写本文时为 1.5.6）。如果代理启用了生产者流控制并且您已达到队列大小限制，则发送可能会阻塞，尽管使用异步发送这不应该发生，除非您使用 producerWindowSize 集进行发送。获得帮助的一种好方法是创建一个测试用例并通过 Jira 问题将其提交到 NMS.ActiveMQ 站点，以便我们可以使用您的测试代码对其进行调查。自 1.5.1 以来已进行了许多修复，因此我建议尝试使用该新版本，因为它可能已经不是问题了。

【讨论】：