RAbbitMQ：如何使用 EventBasicConsumer 在队列为空时停止消费、通道和连接

Question

我们有一些队列，错误队列数量相同。

我的老板让我写一个窗口服务来消费错误队列，并将数据传输到数据库。 Windows 服务应该能够根据数据库表中设置的配置管理应该使用哪些队列，这意味着例如，如果我不想再使用特定队列， 我必须能够在不停止窗口服务的情况下禁用 DB 上特定队列的配置。

我计划在 Quartz 的帮助下编写一个计划的 Windows 服务作业，每个计划的作业都应该读取数据库配置，打开单个连接， 为所有要使用的队列启动任务，每个任务使用一个通道来使用队列。我希望通过这个解决方案来解决，因为在任务结束时通道将被关闭， 连接将被关闭，错误队列将被消耗，并且在下一个作业调度中，要读取的队列的数量和名称可能不同。 此外，在每个计划中使用每个队列的所有消息都可以节省打开/关闭连接/通道，如果我在每个计划中使用一条消息，这可能会很麻烦， 所以计划的时间应该足以在下一个计划之前消耗队列中的所有错误消息。 此外，调度队列将使我有机会配置队列以在 DB 中使用，而无需停止 Windows 服务。

现在我写了一点代码来测试消费单队列的解决方案

var consumer = new QueueingBasicConsumer(channel);
channel.BasicConsume(queueParName, autoAck: false, consumer: consumer);
bool queueEmpty = false;
while (!queueEmpty)
{
    try
        BasicDeliverEventArgs result;
        bool bRead = consumer.Queue.Dequeue(timeOutQueueEmpty, out result);

        if (bRead)
        {
            var msgBody = Encoding.UTF8.GetString(result.Body);
            // TO DB ...
        }
        else
        {
            queueEmpty = true;
        }
    }
    catch (EndOfStreamException ex)
    {
        // ...
    }
}

问题是 QueueingBasicConsumer 已过时，并且在许多地方都被写入以防止 EventBasicConsumer

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);

    // TO DB...
};
channel.BasicConsume(queue: queueParName, autoAck: true, consumer: consumer);

但是对于 EventBasicConsumer，我无法理解如何在队列为空时停止消费以关闭连接和通道

编辑

正如评论中所要求的那样，进行一些更深入的解释。 把所有的代码都放在这里会很困难，因为我们使用了很多我们公司的编译库，所以代码不是完全可以理解的。 无论如何，简化：

1. 我有一个 Host 项目，其中使用 Quartz Package 来安排每 N 次的 Job。

...

var schedule = SimpleScheduleBuilder.Create();
schedule.WithIntervalInSeconds(ConfigurationManager.AppSettings["..."]);
schedule.RepeatForever();
s.ScheduleQuartzJob(q =>
    q.WithJob(() =>
        JobBuilder.Create<RabbitErrorDequeuerJob>().Build())
            .AddTrigger(() =>
                TriggerBuilder.Create()
                    .WithSchedule(schedule)
                    .Build())
);

...

拥有 Ninject IOC，并在模块文件中注入连接

// *** Ninject disposes every Disposable object that has another scope other than InTransientScope
Bind<IConnection>().ToMethod(x =>
{
IConnectionFactory cnf = new ConnectionFactory();
cnf.Uri = new Uri(ConfigurationManager.AppSettings["..."]);
return cnf.CreateConnection();
}).InCallScope();

我有一个 Job 项目，由 Quartz 每 N 次安排一次，其中我有 Execute 方法（Quartz.IJob 接口）

 public void Execute(IJobExecutionContext context)
    {
        try
        {
            List<RabbitQueueConfiguration> lst = //...LIST OF QUEUTE TO DEQUE FROM DATABASE
            foreach (RabbitQueueConfiguration queue in lst)
            {
                Task t = Task.Factory.StartNew(() =>
                {
                    DequeuSingleQueue(queue);
                });
            }
        }
        catch (Exception ex)
        {
            _log.FatalFormat("Error ", ex.Message);
            throw;
        }
    }

在DequeuSingleQueue(queue)中有出列的核心

Answer 1

RabbitMQ 团队会监控 the rabbitmq-users mailing list，但有时只会在 StackOverflow 上回答问题。

---

您应该使用BasicGet 一次处理一条消息。