如何提升IBM MQ .NET Core消费者应用的消息获取速率

问题背景 #

基于.NET Core 3.1 + IBMMQDotnetClient 9.2.5开发的MQ消费应用，当前单线程单条拉取消息的速率仅约8条/秒，峰值时段队列深度可达30000并触发阈值告警。现有核心代码如下：

MqClient核心代码 #

public string GetMessage()
{
    MQMessage retrievedMessage = new MQMessage()
    {
        Version = MQC.MQMD_VERSION_2
    };

    this.queue.Get(retrievedMessage, getMessageOptions);
    
    if (retrievedMessage.DataLength == 0)
    {
        return string.Empty;
    }

    var message = retrievedMessage.ReadString(retrievedMessage.DataLength);
    return message;
}

this.getMessageOptions = new MQGetMessageOptions
{
    Options = MQC.MQGMO_NO_WAIT
        | MQC.MQGMO_FAIL_IF_QUIESCING
        | MQC.MQGMO_COMPLETE_MSG,
    // it is adviced not to use MQC.MQGMO_CONVERT
    WaitInterval = 10000, 
    MatchOptions = MQC.MQMO_NONE
};

public void Connect()
{
    this.queueManager = new MQQueueManager(Configuration.ManagerName, Configuration.ToMqHashTable());
    // If Browse is ever needed: MQC.MQOO_BROWSE
    this.queue = queueManager.AccessQueue(this.Configuration.QueueName, MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_INQUIRE);
}

消费主逻辑代码 #

static void Main(string[] args)
{
    Console.WriteLine("SATS DataFeeder POC");
    dataBuffer = new BufferBlock<string>();
    var timer = new Stopwatch();
    timer.Start();
    int messageCount = 0;
    while (true)
    {
        try
        {
            if (!IsConnected)
                InitializeClient();

            var data = client.GetMessage();
            dataBuffer.Post(data);
            //count = client.CountMessagesInQueue();
            messageCount += 1;
            Console.WriteLine($"Queue count : {messageCount}");

            if (messageCount == 100)
            {
                timer.Stop();
                var totalSeconds = timer.Elapsed.Seconds;
                Console.WriteLine("Message Consumption Rate = Total Time taken to consume 100 messages  = " + 100 / totalSeconds + " messages/second");
            }
        }
        catch (MQException exc) when (exc.ReasonCode == (int)MQExceptionReasonCodes.MQRC_NO_MSG_AVAILABLE)
        {
            Console.WriteLine(exc);
        }
        catch (MQException exc) when (exc.ReasonCode == (int)MQExceptionReasonCodes.MQRC_HOST_NOT_AVAILABLE)
        {
            Console.WriteLine(exc);
        }
        catch (MQException exc)
        {
            Console.WriteLine(exc);
        }
        catch (Exception exc)
        {
            Console.WriteLine(exc);
        }
    }
}

优化方案 #

1. 启用批量拉取（核心优化） #

单条拉取会产生大量MQ往返请求，启用批量拉取可大幅减少交互次数。修改配置与拉取方法：

// 更新getMessageOptions，添加批量拉取配置
this.getMessageOptions = new MQGetMessageOptions
{
    Options = MQC.MQGMO_NO_WAIT
        | MQC.MQGMO_FAIL_IF_QUIESCING
        | MQC.MQGMO_COMPLETE_MSG
        | MQC.MQGMO_ALL_MSGS_AVAILABLE, // 等待批量消息就绪
    WaitInterval = 10000,
    MatchOptions = MQC.MQMO_NONE,
    MaxMsgCount = 100 // 单次拉取最大消息数，可根据消息大小调整
};

// 新增批量拉取方法
public List<string> GetMessagesBatch()
{
    var messages = new List<string>();
    MQMessage retrievedMessage;
    do
    {
        retrievedMessage = new MQMessage()
        {
            Version = MQC.MQMD_VERSION_2
        };
        try
        {
            this.queue.Get(retrievedMessage, getMessageOptions);
            if (retrievedMessage.DataLength > 0)
            {
                messages.Add(retrievedMessage.ReadString(retrievedMessage.DataLength));
            }
        }
        catch (MQException exc) when (exc.ReasonCode == (int)MQExceptionReasonCodes.MQRC_NO_MSG_AVAILABLE)
        {
            break; // 无更多消息，终止循环
        }
    } while (messages.Count < getMessageOptions.MaxMsgCount);
    return messages;
}

2. 多线程并行消费 #

单线程无法充分利用系统资源，根据CPU核心数启动多个消费任务：

static void Main(string[] args)
{
    Console.WriteLine("SATS DataFeeder POC");
    // 设置缓冲区上限，避免内存溢出
    dataBuffer = new BufferBlock<string>(new DataflowBlockOptions { BoundedCapacity = 1000 });

    // 启动多线程消费，数量建议为CPU核心数的2倍
    var consumerTasks = new List<Task>();
    int consumerCount = Environment.ProcessorCount * 2;
    for (int i = 0; i < consumerCount; i++)
    {
        consumerTasks.Add(Task.Run(async () =>
        {
            int localCount = 0;
            var timer = Stopwatch.StartNew();
            while (true)
            {
                try
                {
                    if (!IsConnected)
                        InitializeClient();

                    // 批量拉取消息
                    var batch = client.GetMessagesBatch();
                    foreach (var data in batch)
                    {
                        await dataBuffer.SendAsync(data);
                        localCount++;
                        
                        // 降低日志输出频率，减少性能开销
                        if (localCount % 100 == 0)
                        {
                            timer.Stop();
                            var rate = 100 / timer.Elapsed.TotalSeconds;
                            Console.WriteLine($"Consumer {Task.CurrentId} - Rate: {rate:F2} messages/second");
                            timer.Restart();
                        }
                    }
                }
                catch (MQException exc) when (exc.ReasonCode == (int)MQExceptionReasonCodes.MQRC_HOST_NOT_AVAILABLE)
                {
                    Console.WriteLine(exc);
                    await Task.Delay(5000); // 断开后延迟重连，避免频繁重试
                }
                catch (Exception exc)
                {
                    Console.WriteLine(exc);
                }
            }
        }));
    }

    Task.WaitAll(consumerTasks.ToArray());
}

3. 优化连接与队列权限 #

• 移除不必要的队列权限：若仅做消费，删除MQOO_OUTPUT权限，减少资源占用
• 优化连接检查逻辑：避免循环内重复检查，改为断开后自动重连

修改Connect方法：

public void Connect()
{
    this.queueManager = new MQQueueManager(Configuration.ManagerName, Configuration.ToMqHashTable());
    // 仅保留消费所需权限
    this.queue = queueManager.AccessQueue(this.Configuration.QueueName, MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_INQUIRE);
}

4. 减少同步开销 #

• 控制台输出优化：Console.WriteLine为同步操作，高频输出会严重拖慢性能，建议改用异步日志框架（如Serilog、NLog），或大幅降低输出频率
• 批量写入缓冲区：拉取到批量消息后统一处理，减少SendAsync调用次数

5. 调整MQ客户端网络配置 #

• 增大客户端缓冲区：提升网络传输效率
• 禁用Nagle算法：减少小数据包延迟，适合高频消息场景

在Configuration.ToMqHashTable()中添加以下配置：

hashTable.Add(MQC.MQBUFSIZE_PROPERTY, 1048576); // 设置1MB缓冲区
hashTable.Add(MQC.TCP_NODELAY_PROPERTY, true); // 禁用Nagle算法

内容的提问来源于stack exchange，提问作者dimakavi