Teaser:伙计们，这个问题不是关于如何实现重试政策的。这是关于正确完成 TPL 数据流 block 。

这个问题主要是我之前问题的延续 Retry policy within ITargetBlock .这个问题的答案是@svick 的智能解决方案，它利用了 TransformBlock(来源)和 TransformManyBlock(目标)。剩下的唯一问题就是如何以正确的方式完成这个 block :先等待所有重试完成，然后再完成目标 block 。这是我最终得到的结果(这只是一个片段，不要过多关注非线程安全的 retries 集):

var retries = new HashSet<RetryingMessage<TInput>>();

TransformManyBlock<RetryableMessage<TInput>, TOutput> target = null;
target = new TransformManyBlock<RetryableMessage<TInput>, TOutput>(
    async message =>
    {
        try
        {
            var result = new[] { await transform(message.Data) };
            retries.Remove(message);
            return result;
        }
        catch (Exception ex)
        {
            message.Exceptions.Add(ex);
            if (message.RetriesRemaining == 0)
            {
                if (failureHandler != null)
                    failureHandler(message.Exceptions);

                retries.Remove(message);
            }
            else
            {
                retries.Add(message);
                message.RetriesRemaining--;

                Task.Delay(retryDelay)
                    .ContinueWith(_ => target.Post(message));
            }
            return null;
        }
    }, dataflowBlockOptions);

source.LinkTo(target);

source.Completion.ContinueWith(async _ =>
{
    while (target.InputCount > 0 || retries.Any())
        await Task.Delay(100);

    target.Complete();
});

想法是执行某种轮询并验证是否还有消息等待处理并且没有消息需要重试。但在这个解决方案中，我不喜欢轮询的想法。

是的，我可以将添加/删除重试的逻辑封装到一个单独的类中，甚至例如当重试集变空时执行一些操作，但是如何处理 target.InputCount > 0 条件？当 block 没有未决消息时，不会调用这样的回调，因此似乎在循环中以小延迟验证 target.ItemCount 是唯一的选择。

有人知道实现此目标的更聪明的方法吗？

最佳答案

也许是 ManualResetEvent可以为您解决问题。

将公共(public)属性添加到 TransformManyBlock

private ManualResetEvent _signal  = new ManualResetEvent(false);
public ManualResetEvent Signal { get { return _signal; } }

给你:

var retries = new HashSet<RetryingMessage<TInput>>();

TransformManyBlock<RetryableMessage<TInput>, TOutput> target = null;
target = new TransformManyBlock<RetryableMessage<TInput>, TOutput>(
    async message =>
    {
        try
        {
            var result = new[] { await transform(message.Data) };
            retries.Remove(message);

            // Sets the state of the event to signaled, allowing one or more waiting threads to proceed
            if(!retries.Any()) Signal.Set(); 
            return result;
        }
        catch (Exception ex)
        {
            message.Exceptions.Add(ex);
            if (message.RetriesRemaining == 0)
            {
                if (failureHandler != null)
                    failureHandler(message.Exceptions);

                retries.Remove(message);

                // Sets the state of the event to signaled, allowing one or more waiting threads to proceed
                if(!retries.Any()) Signal.Set(); 
            }
            else
            {
                retries.Add(message);
                message.RetriesRemaining--;

                Task.Delay(retryDelay)
                    .ContinueWith(_ => target.Post(message));
            }
            return null;
        }
    }, dataflowBlockOptions);

source.LinkTo(target);

source.Completion.ContinueWith(async _ =>
{
    //Blocks the current thread until the current WaitHandle receives a signal.
    target.Signal.WaitOne();

    target.Complete();
});

我不确定您的 target.InputCount 设置在哪里。因此，在您更改 target.InputCount 的地方，您可以添加以下代码:

if(InputCount == 0)  Signal.Set();

关于c# - 实现可重试 block 的正确完成，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/17488205/