0

在没有非常详细地了解我们试图解决的问题的情况下，我需要让 NServiceBus 做 5 件事中的 1 件事，但我目前只是想让第一个工作。也就是说，给定来自 Web API 调用的回复，我们希望延迟重试、立即重试、放弃、取消或重新开始。延迟重试看起来最好使用自定义可恢复性来完成，所以我遵循了这个：自定义可恢复性策略并提出了这个

public static class UpdateEndpointConfiguration
{
    public static void ConfigureEndpointForUpdateVocxoSurveyApi(this EndpointConfiguration configuration)
    {
        var recoverabilitySettings = configuration.Recoverability();
        recoverabilitySettings.CustomPolicy(SetCustomPolicy);
    }

    private static RecoverabilityAction SetCustomPolicy(RecoverabilityConfig config, ErrorContext context)
    {
        var action = DefaultRecoverabilityPolicy.Invoke(config, context);
        if (context.Exception is DelayedRetryException delayedRetryException)
        {
            return RecoverabilityAction.DelayedRetry(TimeSpan.FromSeconds(delayedRetryException.DelayRetryTimeoutSeconds));
        }
        return action;
    }
}

然后作为测试，我发了一条简单的消息，这样我就不必强迫 web api 做一些愚蠢的事情：

public class ForceDelayRetry : ICommand
{
    public int DelayInSeconds { get; set; }
}

然后“处理”

public class TestRequestHandler : IHandleMessages<ForceDelayRetry>
{
    private static readonly ILog Log = LogManager.GetLogger(typeof(TestRequestHandler));

    public async Task Handle(ForceDelayRetry message, IMessageHandlerContext context)
    {
        Log.Info($"Start processing {nameof(ForceDelayRetry)}");
        var handleUpdateRequestFailure = IoC.Get<HandleUpdateRequestFailure>();
        await handleUpdateRequestFailure.HandleFailedRequest(new UpdateRequestFailed
        {
            DelayRetryTimeoutSeconds = message.DelayInSeconds,
            Message = $"For testing purposes I am forcing a delayed retry of {message.DelayInSeconds} second(s)",
            RecoveryAction = RecoveryAction.DelayRetry
        }, context, 12345);
        Log.Info($"Finished processing {nameof(ForceDelayRetry)}");
    }
}

我启动了服务，在大约 1.5 分钟的时间内，两条测试消息被处理了大约 5,400 次。日志消息与此类似（为简洁起见，省略了堆栈跟踪）

20180601 15:28:47 :INFO  [14] TestRequestHandler Start processing ForceDelayRetry
20180601 15:28:47 :WARN  [22] NServiceBus.RecoverabilityExecutor Delayed Retry will reschedule message '690f317e-5be0-4511-88b9-a8f2013ac219' after a delay of 00:00:01 because of an exception:
20180601 15:28:47 :INFO  [14] TestRequestHandler Start processing ForceDelayRetry
20180601 15:28:47 :WARN  [14] NServiceBus.RecoverabilityExecutor Delayed Retry will reschedule message '7443e553-b558-486d-b7e9-a8f2014088d5' after a delay of 00:00:01 because of an exception:
20180601 15:28:47 :INFO  [4] TestRequestHandler Start processing ForceDelayRetry
20180601 15:28:47 :WARN  [14] NServiceBus.RecoverabilityExecutor Delayed Retry will reschedule message '690f317e-5be0-4511-88b9-a8f2013ac219' after a delay of 00:00:01 because of an exception:
20180601 15:28:47 :INFO  [14] TestRequestHandler Start processing ForceDelayRetry
20180601 15:28:47 :WARN  [14] NServiceBus.RecoverabilityExecutor Delayed Retry will reschedule message '7443e553-b558-486d-b7e9-a8f2014088d5' after a delay of 00:00:01 because of an exception:

所以要么我做错了什么，要么有一个错误，但我不知道是哪个。任何人都可以看到问题是什么？

编辑

这是方法handleUpdateRequestFailure.HandleFailedRequest

    public async Task HandleFailedRequest(UpdateRequestFailed failure, IMessageHandlerContext context, long messageSurveyId)
    {
        switch (failure.RecoveryAction)
        {
            case RecoveryAction.DelayRetry:
                Log.InfoFormat("Recovery action is {0} because {1}. Retrying in {2} seconds", failure.RecoveryAction, failure.Message, failure.DelayRetryTimeoutSeconds);
                await context.Send(_auditLogEntryCreator.Create(_logger.MessageIsBeingDelayRetried, messageSurveyId));
                throw new DelayedRetryException(failure.DelayRetryTimeoutSeconds);
            case RecoveryAction.EndPipelineRequest:
            case RecoveryAction.RestartPipelineRequest:
            case RecoveryAction.RetryImmediate:
            case RecoveryAction.RouteToErrorQueue:
                break;
        }
    }

正如评论指出的那样，我也会对我发现的消息进行无限重试，但这是它的更新逻辑

    private static RecoverabilityAction SetCustomPolicy(RecoverabilityConfig config, ErrorContext context)
    {
        var action = DefaultRecoverabilityPolicy.Invoke(config, context);
        if (context.Exception is DelayedRetryException delayedRetryException)
        {
            if (config.Delayed.MaxNumberOfRetries > context.DelayedDeliveriesPerformed)
                return RecoverabilityAction.DelayedRetry(TimeSpan.FromSeconds(delayedRetryException.DelayRetryTimeoutSeconds));
        }
        return action;
    }
4

1 回答 1

0

也就是说，给定来自 Web API 调用的回复，我们希望延迟重试、立即重试、放弃、取消或重新开始。延迟重试看起来最好使用自定义可恢复性来完成

除了 NServiceBus 已经提供的功能之外，我不确定我是否理解您想要实现的目标？让立即重试和延迟重试做它最擅长的事情：进行实际重试。

如果您想要更多功能，请使用 saga。让 saga 编排流程并让单独的处理程序执行对外部服务的实际调用。然后，saga 可以根据这个处理程序的回复，决定它是否应该停止、继续、采用替代路径等。

如果您想进一步讨论这个问题，我建议您通过 support@particular.net 与我们联系，我们可以安排电话会议并向您展示我们将如何做到这一点。

于 2018-06-05T09:08:46.987 回答