RabbitMQ 和 C#

Question

RabbitMQ 有一种类似于 MSSMQ 的使用方法，可以从队列中弹出 1000 条消息，然后将数据插入数据库并从那里继续。

我似乎无法通过对通道的订阅，然后对订阅中的 BasicDeliveryEventArgs 执行 foreach 操作，并使用我想要在给定时间处理的最大消息数执行 If 语句来做到这一点。

提前致谢 然而，这仍然从队列中取出所有 22k 消息

using (IConnection connection = factory.CreateConnection())
{
    using (IModel channel = connection.CreateModel())
    {
        channel.QueueDeclare("****", true, false, false, null);

        var subscription = new Subscription(channel, "****", false);
        int maxMessages = 5;
        int i = 0;
        foreach (BasicDeliverEventArgs eventArgs in subscription)
        {
            if (++i == maxMessages)
            {
                Console.WriteLine("Took 5 messages");
                subscription.Ack(eventArgs);
                break;
            }
        }
    }
}

Answer 1

我假设您希望通过将消息组批处理成更大的事务来优化消息到数据库的加载，而不是消耗每条消息的事务成本。强制性警告是，这样做意味着大量消息可能会同时失败，即使只有其中一条消息导致问题，您也可以按照以下方法进行操作...

在通道上设置 QOS：

channel.BasicQos(0, 1000, false);

这将预取 1000 条消息并阻止进一步的流量，直到您确认某些内容。请注意，它不会以 1000 条为单位进行提取。相反，它会确保每次最多预提取 1000 条未确认的消息。模拟块传输非常简单，只需首先处理 1000 条消息，然后一次性 ACK 全部消息即可。

请参阅此处和这里比我的更权威的解释。

还有一点：您可能希望在消息可用时立即刷新队列，即使您尚未达到 1000 条消息的配额。您应该能够通过在 foreach 循环内调用 queue.BasicGet() 直到它耗尽，然后交付您拥有的任何内容（包括您拉出的消息）来完成此操作订阅）到数据库。警告：我自己还没有尝试过，所以我可能在胡说八道，但我认为它会起作用。此方法的优点在于它可以立即将消息推送到数据库中，而无需等待整批 1000 条消息。如果数据库因处理太多小事务而落后，则预取积压将在每个周期之间填满更多。