在ITargetBlock <TInput>中重试策略

Question

我需要在工作流程中引入重试策略.假设有3个块以这种方式连接:

var executionOptions = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 3 };
var buffer = new BufferBlock<int>();
var processing = new TransformBlock<int, int>(..., executionOptions);
var send = new ActionBlock<int>(...);

buffer.LinkTo(processing);
processing.LinkTo(send);

因此,有一个缓冲区累积数据,然后将其发送到转换块,该转换块一次处理不超过3个项目,然后将结果发送到操作块.

在处理过程中可能会出现变换块瞬态错误,如果错误是瞬态错误,我想重试该块.

我知道块通常不可重试(传递到块中的委托可以重试).其中一个选项是包装传递给支持重试的委托.

我也知道有一个非常好的库TransientFaultHandling.Core可以为瞬态故障提供重试机制.这是一个很棒的图书馆,但不是我的情况.如果我将传递给转换块的委托包装到RetryPolicy.ExecuteAsync方法中,则转换块内的消息将被锁定,并且在重试完成或失败之前,转换块将无法接收新消息.想象一下,如果所有3条消息都输入到重试中(假设下次重试尝试将在2分钟内完成)并且失败,则变换块将被卡住,直到至少有一条消息离开变换块.

我看到的唯一的解决办法是延长TranformBlock(实际上,ITargetBlock将有足够太),和做手工重试(比如从这里):

do
 {
    try { return await transform(input); }
    catch
    { 
        if( numRetries <= 0 ) throw;
        else Task.Delay(timeout).ContinueWith(t => processing.Post(message));
    }
 } while( numRetries-- > 0 );

ig将消息再次放入变换块中,但是在这种情况下,重试上下文(剩余的重试次数等)也应该传递给该块.听起来太复杂了......

有没有人看到更简单的方法来实现工作流程块的重试策略？

Answer 1

我认为你几乎必须这样做,你必须跟踪一条消息的剩余重试次数,你必须以某种方式安排重试的尝试.

但是你可以通过将其封装在一个单独的方法中来使其更好.就像是:

// it's a private class, so public fields are okay
private class RetryingMessage<T>
{
    public T Data;
    public int RetriesRemaining;
    public readonly List<Exception> Exceptions = new List<Exception>();
}

public static IPropagatorBlock<TInput, TOutput>
    CreateRetryingBlock<TInput, TOutput>(
    Func<TInput, Task<TOutput>> transform, int numberOfRetries,
    TimeSpan retryDelay, Action<IEnumerable<Exception>> failureHandler)
{
    var source = new TransformBlock<TInput, RetryingMessage<TInput>>(
        input => new RetryingMessage<TInput>
        { Data = input, RetriesRemaining = numberOfRetries });

    // TransformManyBlock, so that we can propagate zero results on failure
    TransformManyBlock<RetryingMessage<TInput>, TOutput> target = null;
    target = new TransformManyBlock<RetryingMessage<TInput>, TOutput>(
        async message =>
        {
            try
            {
                return new[] { await transform(message.Data) };
            }
            catch (Exception ex)
            {
                message.Exceptions.Add(ex);
                if (message.RetriesRemaining == 0)
                {
                    failureHandler(message.Exceptions);
                }
                else
                {
                    message.RetriesRemaining--;
                    Task.Delay(retryDelay)
                        .ContinueWith(_ => target.Post(message));
                }
                return null;
            }
        });

    source.LinkTo(
        target, new DataflowLinkOptions { PropagateCompletion = true });

    return DataflowBlock.Encapsulate(source, target);
}

我添加了代码来跟踪异常,因为我认为不应该忽略失败,它们应该至少记录下来.

此外,此代码在完成时不能很好地工作:如果有重试等待它们的延迟而你Complete()是块,它将立即完成并且重试将丢失.如果这对您来说是一个问题,您将必须跟踪未完成的reties并在完成target时source完成并且没有重试等待.