ideship/itfx
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：mq消费未ack

Browse Source
This commit is contained in:
xiaolipro 2025-07-29 14:58:46 +08:00
parent 974ad920dd
commit 34e99f0a7e
1 changed files with 20 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/InterfaceForward.Application/Rabbit/RabbitClient.cs
View File

@ -29,17 +29,31 @@ public class RabbitClient(
} }
var consumer = new AsyncEventingBasicConsumer(channel); var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += async (model, ea) => consumer.Received += (model, ea) =>
{
_ = Task.Run(async () =>
{ {
await using var scope = serviceScopeFactory.CreateAsyncScope(); try
foreach (var handler in scope.ServiceProvider.GetServices<IRabbitHandler>())
{ {
if (handler.Enable(consumeOptions)) await using var scope = serviceScopeFactory.CreateAsyncScope();
foreach (var handler in scope.ServiceProvider.GetServices<IRabbitHandler>())
{ {
await handler.Handle(ea); if (handler.Enable(consumeOptions))
{
await handler.Handle(ea);
}
} }
channel.BasicAck(ea.DeliveryTag, false);
} }
}; catch (Exception ex)
{
logger.LogError($"rabbit on queue({consumeOptions.Queue}) received error: {ex}");
channel.BasicNack(ea.DeliveryTag, false, consumeOptions.FailedRequeue);
}
});
return Task.CompletedTask;
};
channel.CallbackException += (sender, args) => channel.CallbackException += (sender, args) =>
{ {