RabbitMQ.Client BasicPublish возвращается до его завершения

Question

У меня проблема с консольным приложением, которое запускается как запланированное задание и отправляет около 350 тыс. Сообщений в очередь кроликов, удаляет их и завершает работу.Моя проблема в том, что model.BasicPublish возвращается до того, как сообщение фактически полностью опубликовано, и когда вызывается model.Close, это приводит к истечению времени ожидания.Еще одна вещь, которую я заметил, заключается в том, что после того, как приложение закрывается и закрывается, пользовательский интерфейс управления кроликом сообщает, что обмен все еще принимает сообщения в течение 30-45 секунд.Я использую пакет слепка RabbitMQ.Client версии 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

пользовательский интерфейс управления кроликом сообщает, что обмен все еще принимает сообщения в течение 30-45 секунд

Это связано с буферизацией данных в сети.

Myпроблема в том, что model.BasicPublish возвращается до того, как сообщение фактически полностью опубликовано, и когда вызывается model.Close, это приводит к истечению времени ожидания

В этом случае следует включить подтверждение издателя (через ConfirmSelect и WaitForConfirms) и закрывайте модель только после подтверждения всех опубликованных сообщений.

---

_{ПРИМЕЧАНИЕ: команда RabbitMQ контролирует rabbitmq-usersсписок рассылки и только иногда отвечает на вопросы по StackOverflow.}