RabbitMQ听众没有被解雇

Question

我正在使用ASP.NETCORE 6.0项目，

我正在使用RabbitMQ进行购物车实施。 IE：付款成功后，应进行预订。

首先，我正在创建队列：

var factory = new ConnectionFactory
                    {
                        Uri = new Uri(_config.GetValue<string>("AmpqUrl")),
                    };

                    try
                    {
                        using var connection = factory.CreateConnection();
                        using var channel = connection.CreateModel();

                        channel.QueueDeclare(queue: "confirmed_payments", durable: true, exclusive: false, autoDelete: false, arguments: null);

                        var data = new
                        {
                            transactionId,
                            paymentConfirmedAt = DateTime.UtcNow,
                        };

                        var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(data));

                        channel.BasicPublish(exchange: "", routingKey: "confirmed_payments", basicProperties: null, body: body);

                    }
                    catch (RabbitMQ.Client.Exceptions.BrokerUnreachableException ex)
                    {
                               Console.WriteLine("ex.ToString()");

                    }

它的听众（另一个项目）：

 public Task ListenPaymentConfimations(CancellationToken cancellationToken)
    {
        var factory = new ConnectionFactory
        {
            Uri = new Uri(_configuration.GetValue<string>("AmpqUrl")),
        };

        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();

        var confimedPaymentsConsumer = new EventingBasicConsumer(_channel);

        confimedPaymentsConsumer.Received += async (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var response = Encoding.UTF8.GetString(body);

            var data = JsonConvert.DeserializeAnonymousType(response,
                new { transactionId = "", paymentConfirmedAt = "" }
            );

            var date = DateTime.Parse(data.paymentConfirmedAt).ToUniversalTime();

            using var scope = _serviceProvider.CreateScope();

            var dbService =
                scope.ServiceProvider.GetRequiredService<ITechneDbService>();

            var isPaymentConfimed = await dbService.UpdateCartPaymentConfirmedAt(data.transactionId, date);

            _logger.LogInformation("Transaction - {0}", data.transactionId);
            _logger.LogInformation("Transaction - {0}", date);
            _logger.LogInformation("Payment Confirmed - {0}", isPaymentConfimed);

            if (isPaymentConfimed)
            {
               // handle booking
            }
        };

        _channel.BasicConsume(queue: "confirmed_payments",
                     autoAck: true,
                     consumer: confimedPaymentsConsumer);

        return Task.CompletedTask;
    }

当我尝试进行酒店预订时，有时会击中听众，我可以成功预订。 但是有时候听众没有被打击（创建队列时也没有例外）。

我不知道为什么会发生这种情况。

任何人都可以找到这个问题， 请帮我。

Answer 1

更改连接的构造函数，以表明应异步派遣消费者。

var factory = new ConnectionFactory
{
    Uri = new Uri(_configuration.GetValue<string>("AmpqUrl")),
    DispatchConsumersAsync = true // <---- this
};

您可能还需要更改为使用asynceventingBasicConsumer而不是EventingBasicConsumer。