关于.net core：RabbitMQ Consumer : AlreadyClosedException

RabbitMQ Consumer : AlreadyClosedException

我有一个简单的 RabbitMQ 发布者和消费者代码，如下所示。

首先，我在 My_Tasks 队列中创建了 10 条不同的消息。当我尝试逐个获取这些消息并且 autoAck 标志为 false 时，我可以读取第一条消息，但无法将确认发送到 RabbitMQ 服务器。我收到下面写的错误；

出版商；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

var qName ="My_Tasks";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(qName, durable: true, false, false, null);

var body = Encoding.UTF8.GetBytes(message);

var prop = channel.CreateBasicProperties();
prop.Persistent = true;

channel.BasicPublish("", routingKey: qName, prop, body);
}
}

消费者；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

var qName ="My_Tasks";
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(qName, durable: true, false, false, null);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume(qName, autoAck: false, consumer);

consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
channel.BasicAck(ea.DeliveryTag, multiple: false);
};
}
}

RabbitMQ.Client.Exceptions.AlreadyClosedException://’已经关闭：AMQP操作被中断：AMQP关闭原因，由Application发起，code=200，text=//’Goodbye//’，classId=0，methodId=0 //’

在 RabbitMQ.Client.Impl.SessionBase.Transmit(命令 cmd)
在 RabbitMQ.Client.Impl.ModelBase.ModelSend(MethodBase 方法，ContentHeaderBase 标头，ReadOnlyMemory 1 正文)
在 RabbitMQ.Client.Framing.Impl.Model.BasicAck(UInt64 deliveryTag，布尔倍数)
在 RabbitMQ.Client.Impl.RecoveryAwareModel.BasicAck(UInt64 deliveryTag，布尔倍数)
在 RabbitMQ.Client.Impl.AutorecoveringModel.BasicAck(UInt64 deliveryTag，布尔倍数)
在 D:////////Projects//// 的 QueueExample.Consumer.Program.<>c__DisplayClass0_0.b__0(Object model, BasicDeliverEventArgs ea)
abbitMQTutorial////////QueueExample////////QueueExample.Consumer////////Program.cs：第 36 行
在 RabbitMQ.Client.Events.EventingBasicConsumer.HandleBasicDeliver(字符串 consumerTag，UInt64 deliveryTag，布尔重新交付，字符串交换，字符串 routingKey，IBasicProperties 属性，ReadOnlyMemory`1 正文)
在 RabbitMQ.Client.Impl.ConcurrentConsumerDispatcher.<>c__DisplayClass10_0.b__0()

感谢您的帮助

我找到了错误的答案。我的示例应用程序是一个控制台应用程序，因此当消费者收到消息时，连接已经关闭。当写入 Connsole.ReadLine() 以等待关闭连接的响应时，消费者可以读取所有消息。

那是临时解决方案，您可以从这个angular找到自己的解决方案。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

var consumer = new EventingBasicConsumer(channel);
channel.BasicConsume(qName, autoAck: false, consumer);

consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
channel.BasicAck(ea.DeliveryTag, multiple: false);
};

//Solution is here
Console.WriteLine("Press Any Key to Continue..");
Console.ReadLine();
}
}

这应该是评论，但我缺少声誉。

您的解决方案已经奏效了吗？你在哪里定义你与经纪人的联系？您是否在 docker-compose 配置中使用代理？在这种情况下，您的代理必须连接到

1	rabbitmq://host.docker.internal

当我开始在 .Net Core 中使用 RabbitMQ 时，我遇到了同样的问题。
由于我的应用程序(微服务模式)处于早期阶段，我切换到处理所有事情的 MassTransit 框架。我强烈推荐它，因为它”正常工作”，您可以完全专注于您想要的系统功能。

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/tech/269861.html

关于.net core：RabbitMQ Consumer : AlreadyClosedException

RabbitMQ Consumer : AlreadyClosedException

相关推荐

发表回复