RabbitMQ.Client BasicPublish 在完成之前返回

Question

我遇到一个问题，我有一个控制台应用程序作为计划任务运行，并向兔子队列发送大约 350k 消息，处理队列并结束。我的问题是 model.BasicPublish 在消息实际完全发布之前返回，并且当调用 model.Close 时它最终会超时。我注意到的另一件事是，应用程序结束并关闭后，Rabbit 管理 UI 表示交易所仍在接收消息约 30-45 秒。我正在使用 RabbitMQ.Client nugget 包版本 5.1.0

代码非常简单，只需复制/粘贴（用户名/密码除外）：

        var connectionFactory = new ConnectionFactory();
        connectionFactory.HostName = "localhost";
        connectionFactory.Password = "pass";
        connectionFactory.UserName = "user";
        connectionFactory.VirtualHost = "test";

        var connection = connectionFactory.CreateConnection();
        var model = connection.CreateModel();

        model.QueuePurge("x");

        var fileLines = File.ReadAllLines(@"bigfile.txt");
        var accounts = fileLines.Select(x => x.Split(',').FirstOrDefault()).Where(x => !string.IsNullOrWhiteSpace(x));
        var watch = new Stopwatch();
        watch.Start();
        Console.WriteLine("Start: " + DateTime.Now);
        foreach (var account in accounts)
        {
            var bytes = Encoding.Default.GetBytes(account);
            model.BasicPublish("incoming", "", null, bytes);
        }
        Console.WriteLine("Done: " + DateTime.Now);
        Console.WriteLine("Time: " + watch.Elapsed);
        try
        {
            model.Close();
        }
        catch (Exception exception)
        {
            Console.WriteLine("Error: " + exception);
        }

我的兔子队列和交换配置：

{
    "rabbit_version": "3.7.7",
    "policies": [],
    "queues": [
        {
            "name": "x",
            "durable": false,
            "auto_delete": false,
            "arguments": {}
        }
    ],
    "exchanges": [
        {
            "name": "incoming",
            "type": "direct",
            "durable": false,
            "auto_delete": false,
            "internal": false,
            "arguments": {}
        }
    ],
    "bindings": [
        {
            "source": "incoming",
            "destination": "x",
            "destination_type": "queue",
            "routing_key": "",
            "arguments": {}
        }
    ]
}

foreach 块大约需要 14 秒。

有人知道如何让它正常工作吗？预先感谢。

Answer 1

Rabbit 管理 UI 表示交易所仍在接收消息 约30-45秒

这是由于网络数据缓冲造成的。

我的问题是 model.BasicPublish 在 消息实际上已完全发布，当 model.Close 为 称其最终超时

在这种情况下，您应该启用发布者确认（通过

ConfirmSelect

和

WaitForConfirms

），并且仅在确认所有已发布消息后才关闭模型。