RabbitMQ 响应在控制器中丢失

Question

大家晚上好，我有一个使用 .NET 编写的网络应用程序和一个移动应用程序。 我正在通过我的网络应用程序向 rabbitMQ 服务器发送一些值，这工作正常，我将它放入队列中，但是当移动应用程序接受请求时，我没有得到返回值。

这是我的控制器

    public async Task<ActionResult> GetCollect(int id)
    {
        int PartnerId = 0;
        bool SentRequest = false;

        try
        {
            SentRequest = await RuleRabbitMQ.SentRequestRule(id);
            if(SentRequest )
            {
                PartnerId = await RuleRabbitMQ.RequestAccepted();
            }
        }
        catch (Exception Ex)
        {

        }
    }

这是我的 RabbitMQ 类

 public class InteractionRabbitMQ
{

    public async Task<bool> SentRequestRule(int id)
    {
        bool ConnectionRabbitMQ = false;
        await Task.Run(() => 
        {
            try
            {
                ConnectionFactory connectionFactory = new ConnectionFactory()
                {
                   //credentials go here
                };

                IConnection connection = connectionFactory.CreateConnection();
                IModel channel = connection.CreateModel();

                channel.QueueDeclare("SolicitacaoSameDay", true, false, false, null);

                string rpcResponseQueue = channel.QueueDeclare().QueueName;

                string correlationId = Guid.NewGuid().ToString();

                IBasicProperties basicProperties = channel.CreateBasicProperties();
                basicProperties.ReplyTo = rpcResponseQueue;
                basicProperties.CorrelationId = correlationId;
                byte[] messageBytes = Encoding.UTF8.GetBytes(string.Concat(" ", id.ToString()));
                channel.BasicPublish("", "SolicitacaoSameDay", basicProperties, messageBytes);

                channel.Close();
                connection.Close();

                if (connection != null)
                {
                    ConnectionRabbitMQ = true;
                }
                else
                {
                    ConnectionRabbitMQ = false;
                }
            }
            catch (Exception Ex)
            {
                throw new ArgumentException($"Thre was a problem with  RabbitMQ server. " +
                    $"Pleaser, contact the support with Error: {Ex.ToString()}");
            }
        });
        return ConnectionRabbitMQ;
    }

    public async Task<int> RequestAccepted()
    {

        bool SearchingPartner= true;
        int PartnerId = 0;

        await Task.Run(() =>
        {
            try
            {
                var connectionFactory = new ConnectionFactory()
                {
                    // credentials
                };

                IConnection connection = connectionFactory.CreateConnection();
                IModel channel = connection.CreateModel();

                channel.BasicQos(0, 1, false);

                var eventingBasicConsumer = new EventingBasicConsumer(channel);


                eventingBasicConsumer.Received += (sender, basicDeliveryEventArgs) =>
                {
                    string Response = Encoding.UTF8.GetString(basicDeliveryEventArgs.Body, 0, basicDeliveryEventArgs.Body.Length);
                    channel.BasicAck(basicDeliveryEventArgs.DeliveryTag, false);
                    if(!string.IsNullOrWhiteSpace(Response))
                    {
                        int Id = Convert.ToInt32(Response);
                        PartnerId = Id > 0 ? Id : 0;
                        SearchingPartner = false;

                    }
                };


                channel.BasicConsume("SolicitacaoAceitaSameDay", false, eventingBasicConsumer);
            }
            catch (Exception Ex)
            {
               // error message
            }
        });
        return PartnerId ;
    }

Answer 1

我不确定这是否有效，无法构建基础架构来快速测试，但是 - 你的问题是 RequestAccepted 返回一个 Task 完成 之前 Received事件被 Rabbit 客户端库捕获。

同步这两个可能可以解决问题，但是请注意，这可能会使您的代码等待很长时间（甚至 - 永远不会得到）响应。

public Task<int> RequestAccepted()
{
    bool SearchingPartner= true;
    int PartnerId = 0;

    var connectionFactory = new ConnectionFactory()
    {
        // credentials
    };

    IConnection connection = connectionFactory.CreateConnection();
    IModel channel = connection.CreateModel();

    channel.BasicQos(0, 1, false);

    TaskCompletionSource<int> tcs = new TaskCompletionSource<int>();

    var eventingBasicConsumer = new EventingBasicConsumer(channel);
    eventingBasicConsumer.Received += (sender, basicDeliveryEventArgs) =>
    {
         string Response = Encoding.UTF8.GetString(basicDeliveryEventArgs.Body, 0, basicDeliveryEventArgs.Body.Length);
         channel.BasicAck(basicDeliveryEventArgs.DeliveryTag, false);
         if(!string.IsNullOrWhiteSpace(Response))
         {
             int Id = Convert.ToInt32(Response);
             PartnerId = Id > 0 ? Id : 0;
             SearchingPartner = false;

             tcs.SetResult( PartnerId );
         }
    };


    channel.BasicConsume("SolicitacaoAceitaSameDay", false, eventingBasicConsumer);

    return tcs.Task;
}

这种方法存在几个问题。

首先，没有错误处理。 那么，如果事件是由 RMQ 在消费者订阅它之前发送的呢？消费者将阻塞，因为它永远不会收到任何回复。

最后，我不认为 RMQ 消费者打算在对您的控制器的每个请求中创建，然后从不处置。虽然这可以在您手动创建几个请求的开发盒上工作，但它可能永远无法解决数十/数百并发用户访问您的网站并且多个 RMQ 消费者相互竞争的场景。

我认为除了将消费者从您的网络应用程序中完全分离出来，将其放入系统服务或 Hangfire 作业中并让它从缓存中获取对所有可能请求的响应之外，没有其他简单的方法可以解决此问题- 响应网络请求。

不过，这纯属推测，基于我对您尝试做的事情的理解。当然，我在这里可能是错的。

Answer 2

byte[] messageBytes = Encoding.UTF8.GetBytes(string.Concat(" ", idColeta.ToString()));

我认为“idColeta”是空白的。