RabbitMQ 和 C#答案

【问题标题】：RabbitMQ and C#RabbitMQ 和 C#
【发布时间】：2011-11-18 06:25:52
【问题描述】：

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;
            }
        }
    }
}

【问题讨论】：

我不明白这个问题，特别是“做一个 if 语句”部分。你能澄清一下吗？
对我来说仍然没有意义。你想完成什么？
我试图一次从队列中获取 1000 条消息，而不是一次将所有消息都下来

标签： c# rabbitmq

【解决方案1】：

我假设您希望通过将一组消息分批成更大的事务而不是消耗每条消息的事务成本来优化将消息加载到数据库中。强制性警告表明，这样做意味着大量消息可能一起失败，即使其中只有一个会导致问题，这就是你应该如何去做...

在频道上设置QOS：

channel.BasicQos(0, 1000, false);

这将预先获取 1000 条消息并阻止进一步的流量，直到您确认某些内容。请注意，它不会以 1000 个块为单位进行获取。相反，它确保在任何时候最多预获取 1000 条 UNACK 消息。模拟块传输就像首先处理 1000 条消息，然后一次性 ACK 一样简单。

请参阅 here 和 here 以获得比我更权威的解释。

还有一点：您可能希望在消息可用时立即刷新队列，即使您尚未达到 1000 条消息的配额。您应该能够通过在foreach 循环中调用queue.BasicGet() 直到它干涸，然后将您拥有的任何内容（包括您从subscription 中提取的消息）传递到数据库来执行此操作。警告：我自己没有尝试过，所以我可能在说垃圾，但我认为它会起作用。这种方法的美妙之处在于它立即将消息推送到数据库中，而无需等待整批 1000 条消息。如果数据库因处理太多小事务而落后，预取积压将在每个周期之间简单地填满。

【讨论】：