From 32753e7606842f76410cb752ee04c34f8fc73c61 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Tue, 21 Jul 2026 10:45:57 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AE=80=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Messaging/OrderBatchJobEventHandler.cs | 13 +--- .../Messaging/OrderBatchRabbitMqOptions.cs | 7 +- .../Orders/Batch/OrderBatchJobProcessor.cs | 70 +++---------------- src/Oms.Tests/OrderBatchJobProcessorTests.cs | 51 +++----------- .../appsettings.Development.json | 7 +- 5 files changed, 25 insertions(+), 123 deletions(-) diff --git a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs index 9228509..580e23c 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs @@ -8,7 +8,8 @@ using Oms.Application.Services.Orders.Batch; namespace Oms.Application.Messaging; /// -/// 订单批量集成事件处理器。消费通道/Ack/Nack/DLX/重试由 Fake.EventBus.RabbitMQ 负责。 +/// 订单批量集成事件处理器。 +/// 正常返回 Ack；抛异常由 Fake.EventBus.RabbitMQ 重试后 Nack 进 DLX。 /// public sealed class OrderBatchJobEventHandler( OrderBatchJobProcessor processor, @@ -29,15 +30,7 @@ public sealed class OrderBatchJobEventHandler( } using var unitOfWork = unitOfWorkManager.Begin(); - var message = @event.ToMessage(); - var result = await processor.ProcessAsync(message, cancellationToken); + await processor.ProcessAsync(@event.ToMessage(), cancellationToken); await unitOfWork.CompleteAsync(cancellationToken); - - // Fake EventBus：成功返回则 Ack；抛异常则框架内重试后 Nack(requeue=false) 进 DLX。 - if (result.Kind == OrderBatchJobProcessor.ProcessResultKind.Reject && result.Requeue) - { - throw new InvalidOperationException( - $"Order batch system failure, allow Fake EventBus retry. action={message.Action} jobItemId={message.JobItemId} messageId={message.MessageId}"); - } } } diff --git a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs index baaa7bb..7158cf3 100644 --- a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs +++ b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs @@ -1,7 +1,7 @@ namespace Oms.Application.Messaging; /// -/// 订单批量消息业务开关。连接/消费/QoS/DLX 由 Fake.EventBus.RabbitMQ（RabbitMqEventBus）配置。 +/// 订单批量消息业务开关。连接/消费/QoS/DLX/重试由 Fake.EventBus.RabbitMQ 配置。 /// public class OrderBatchRabbitMqOptions { @@ -11,9 +11,4 @@ public class OrderBatchRabbitMqOptions /// 是否启用订单批量消息发布与处理。 /// public bool Enabled { get; set; } - - /// - /// 系统失败时任务项最大重试次数（业务侧），与 Fake EventBus 进程内重试相互独立。 - /// - public int MaxRetryCount { get; set; } = 3; } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index cbd99e7..543d472 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -1,7 +1,5 @@ -using Fake.Application; using Fake.DependencyInjection; using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; using Newtonsoft.Json; using Oms.Application.Dtos.Orders; using Oms.Application.Messaging; @@ -13,15 +11,15 @@ namespace Oms.Application.Services.Orders.Batch; /// /// 消费 RabbitMQ 消息后更新任务状态并执行业务。 +/// 正常返回由 Fake EventBus Ack；抛异常则框架重试后进 DLX。 /// public sealed class OrderBatchJobProcessor( IOrderBatchJobRepository batchJobRepository, IOrderTaskRepository orderTaskRepository, IOrderBatchActionExecutor actionExecutor, - IOptions rabbitMqOptions, ILogger logger) : IScopedDependency { - public async Task ProcessAsync( + public async Task ProcessAsync( OrderBatchJobMessage message, CancellationToken cancellationToken = default) { @@ -32,18 +30,18 @@ public sealed class OrderBatchJobProcessor( if (item == null) { logger.LogWarning("Job item {JobItemId} not found, ack message {MessageId}", message.JobItemId, message.MessageId); - return ProcessResult.Ack; + return; } if (item.Status == OrderBatchJobItemStatusCodes.Success) { - return ProcessResult.Ack; + return; } if (item.Status == OrderBatchJobItemStatusCodes.Cancelled) { logger.LogInformation("Skip cancelled job item {JobItemId}", item.Id); - return ProcessResult.Ack; + return; } item.Status = OrderBatchJobItemStatusCodes.Running; @@ -56,7 +54,7 @@ public sealed class OrderBatchJobProcessor( if (message.JobItemId <= 0) { - // fire-and-forget（如轨迹同步）：失败也不 requeue，避免空转打爆 DB + // fire-and-forget（如轨迹同步）：失败只记日志，不抛、不进 DLX if (!result.Success) { logger.LogWarning( @@ -66,13 +64,13 @@ public sealed class OrderBatchJobProcessor( result.FailReason); } - return ProcessResult.Ack; + return; } item = await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken); if (item == null) { - return ProcessResult.Ack; + return; } item.FinishedAt = DateTime.Now; @@ -98,7 +96,7 @@ public sealed class OrderBatchJobProcessor( await batchJobRepository.UpdateItemAsync(item, cancellationToken); await RefreshJobSummaryAsync(item.JobId, cancellationToken); - return ProcessResult.Ack; + return; } if (result.IsBusinessFailure) @@ -112,39 +110,7 @@ public sealed class OrderBatchJobProcessor( } await RefreshJobSummaryAsync(item.JobId, cancellationToken); - return ProcessResult.Ack; } - - item.RetryCount++; - var maxRetryCount = Math.Max(1, rabbitMqOptions.Value.MaxRetryCount); - if (item.RetryCount >= maxRetryCount) - { - item.Status = OrderBatchJobItemStatusCodes.Failed; - item.FailReason = string.IsNullOrWhiteSpace(result.FailReason) - ? "系统重试次数已耗尽" - : result.FailReason; - item.FinishedAt = DateTime.Now; - item.UpdateTime = DateTime.Now; - await batchJobRepository.UpdateItemAsync(item, cancellationToken); - if (message.Action == OrderBatchActionCodes.ImportRow) - { - await RecordImportTaskFailAsync(message, item.FailReason, cancellationToken); - } - - await RefreshJobSummaryAsync(item.JobId, cancellationToken); - logger.LogWarning( - "Job item {JobItemId} retry exhausted after {RetryCount} attempts", - item.Id, - item.RetryCount); - return ProcessResult.Ack; - } - - item.Status = OrderBatchJobItemStatusCodes.Pending; - item.FailReason = null; - item.FinishedAt = null; - item.UpdateTime = DateTime.Now; - await batchJobRepository.UpdateItemAsync(item, cancellationToken); - return ProcessResult.Reject(isBusinessFailure: false); } private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken) @@ -251,22 +217,4 @@ public sealed class OrderBatchJobProcessor( } ], cancellationToken); } - - public enum ProcessResultKind - { - Ack, - Reject, - } - - public readonly struct ProcessResult - { - public ProcessResultKind Kind { get; init; } - - public bool Requeue { get; init; } - - public static ProcessResult Ack => new() { Kind = ProcessResultKind.Ack }; - - public static ProcessResult Reject(bool isBusinessFailure) => - new() { Kind = ProcessResultKind.Reject, Requeue = !isBusinessFailure }; - } } diff --git a/src/Oms.Tests/OrderBatchJobProcessorTests.cs b/src/Oms.Tests/OrderBatchJobProcessorTests.cs index 28e11aa..61586e1 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -1,6 +1,5 @@ using FluentAssertions; using Microsoft.Extensions.Logging.Abstractions; -using Microsoft.Extensions.Options; using Moq; using Oms.Application.Messaging; using Oms.Application.Services.Orders.Batch; @@ -23,12 +22,11 @@ public class OrderBatchJobProcessorTests _batchJobRepository.Object, _orderTaskRepository.Object, _actionExecutor.Object, - Options.Create(new OrderBatchRabbitMqOptions { MaxRetryCount = 3 }), NullLogger.Instance); } [Fact] - public async Task ProcessAsync_BusinessFailure_ShouldMarkFailedAndAck() + public async Task ProcessAsync_BusinessFailure_ShouldMarkFailed() { var item = CreateItem(id: 10, status: OrderBatchJobItemStatusCodes.Pending); var message = CreateMessage(item.Id); @@ -38,54 +36,30 @@ public class OrderBatchJobProcessorTests .ReturnsAsync(OrderBatchActionExecuteResult.BusinessFail("订单不存在")); SetupJobRefresh(item.JobId, [item]); - var result = await _processor.ProcessAsync(message); + await _processor.ProcessAsync(message); - result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack); item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed); item.FailReason.Should().Be("订单不存在"); } [Fact] - public async Task ProcessAsync_SystemFailure_WithRetriesRemaining_ShouldResetPendingAndReject() + public async Task ProcessAsync_SystemFailure_ShouldThrow() { - var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending, retryCount: 1); + var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending); var message = CreateMessage(item.Id); SetupItemLookup(item); _actionExecutor .Setup(x => x.ExecuteAsync(message, It.IsAny())) .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); - var result = await _processor.ProcessAsync(message); + var act = () => _processor.ProcessAsync(message); - result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Reject); - result.Requeue.Should().BeTrue(); - item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending); - item.RetryCount.Should().Be(2); - item.FailReason.Should().BeNull(); - item.FinishedAt.Should().BeNull(); + await act.Should().ThrowAsync() + .WithMessage("*下游超时*"); } [Fact] - public async Task ProcessAsync_SystemFailure_WhenRetryExhausted_ShouldMarkFailedAndAck() - { - var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending, retryCount: 2); - var message = CreateMessage(item.Id); - SetupItemLookup(item); - _actionExecutor - .Setup(x => x.ExecuteAsync(message, It.IsAny())) - .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); - SetupJobRefresh(item.JobId, [item]); - - var result = await _processor.ProcessAsync(message); - - result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack); - item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed); - item.RetryCount.Should().Be(3); - item.FailReason.Should().Be("下游超时"); - } - - [Fact] - public async Task ProcessAsync_CancelledItem_ShouldAckWithoutExecuting() + public async Task ProcessAsync_CancelledItem_ShouldSkipWithoutExecuting() { var item = CreateItem(id: 13, status: OrderBatchJobItemStatusCodes.Cancelled); var message = CreateMessage(item.Id); @@ -93,16 +67,15 @@ public class OrderBatchJobProcessorTests .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny())) .ReturnsAsync(item); - var result = await _processor.ProcessAsync(message); + await _processor.ProcessAsync(message); - result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack); _actionExecutor.Verify( x => x.ExecuteAsync(It.IsAny(), It.IsAny()), Times.Never); } [Fact] - public async Task ProcessAsync_FireAndForgetFailure_ShouldAckWithoutRequeue() + public async Task ProcessAsync_FireAndForgetFailure_ShouldNotThrow() { var message = new OrderBatchJobMessage { @@ -116,10 +89,8 @@ public class OrderBatchJobProcessorTests .Setup(x => x.ExecuteAsync(message, It.IsAny())) .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); - var result = await _processor.ProcessAsync(message); + await _processor.ProcessAsync(message); - result.Kind.Should().Be(OrderBatchJobProcessor.ProcessResultKind.Ack); - result.Requeue.Should().BeFalse(); _batchJobRepository.Verify( x => x.GetItemByIdAsync(It.IsAny(), It.IsAny()), Times.Never); diff --git a/src/SimpleAdmin.Api/appsettings.Development.json b/src/SimpleAdmin.Api/appsettings.Development.json index 6e8f80d..9680e9d 100644 --- a/src/SimpleAdmin.Api/appsettings.Development.json +++ b/src/SimpleAdmin.Api/appsettings.Development.json @@ -31,10 +31,6 @@ "RefreshExpiryInDays": "7", "ClockSkewMinutes": "5" }, - "FeiShu": { - "Webhook": "https://example.com/webhook", - "EnableFeiShuExceptionSubscribe": false - }, "FeiShuNotice": { "Webhook": "https://example.com/webhook", "EnableFeiShuExceptionSubscribe": false @@ -112,8 +108,7 @@ }, "OrderBatch": { "RabbitMq": { - "Enabled": false, - "MaxRetryCount": 3 + "Enabled": false } } } \ No newline at end of file