gaoxing/UNIVPLMDataIntegration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加RabbitMQ消息处理失败，重新回炉

Browse Source
This commit is contained in:
eithday 2025-08-27 09:21:25 +08:00
parent 0ecf99ba1a
commit 844275262f
1 changed files with 24 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
Admin.NET/Admin.NET.Core/RabbitMQ/RabbitMqConsumer.cs
View File

@ -37,15 +37,34 @@ public class RabbitMqConsumer : BackgroundService
consumer.ReceivedAsync += async (sender, ea) =>
{
byte[] body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
try
{
byte[] body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
await _handler.HandleMessageAsync(message);
await _handler.HandleMessageAsync(message);
Console.WriteLine($"[x] Received from {_handler.QueueName}: {message}");
await channel.BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false);
Console.WriteLine($"[x] Received from {_handler.QueueName}: {message}");
await channel.BasicAckAsync(deliveryTag: ea.DeliveryTag, multiple: false);
}
catch (Exception ex)
{
_logger.LogError(ex, "消息处理失败");
await channel.BasicNackAsync(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);
}
};
await channel.BasicConsumeAsync(queue: _handler.QueueName, autoAck: false, consumer: consumer);
// 注册取消回调
stoppingToken.Register(async () =>
{
_logger.LogInformation("Stopping RabbitMQ consumer");
if (channel.IsOpen)
{
await channel.CloseAsync();
}
_connection.Dispose();
});
}
}