54

Teaser：伙计们，这个问题不是关于如何实施重试策略的。这是关于正确完成 TPL 数据流块。

这个问题主要是我之前的问题Retry policy in ITargetBlock的延续。这个问题的答案是@svick 的智能解决方案，它利用TransformBlock（源）和TransformManyBlock（目标）。剩下的唯一问题是以正确的方式完成这个块：等待所有重试首先完成，然后完成目标块。这是我最终得到的（这只是一个片段，不要过多关注非线程安全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条件？当块没有待处理的消息时，不会调用这样的回调，因此似乎target.ItemCount在具有小延迟的循环中进行验证是唯一的选择。

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

4

2 回答 2

2

Maybe a ManualResetEvent can do the trick for you.

Add a public property to TransformManyBlock

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

And here you go:

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();
});

I am not sure where your target.InputCount is set. So at the place you change target.InputCount you can add following code:

if(InputCount == 0)  Signal.Set();
于 2013-10-07T07:31:53.670 回答
1

结合 hwcverwe 答案和 JamieSee 评论可能是理想的解决方案。

首先，您需要创建多个事件：

var signal  = new ManualResetEvent(false);
var completedEvent = new ManualResetEvent(false);

然后，您必须创建一个观察者并订阅TransformManyBlock，以便在相关事件发生时通知您：

var observer = new RetryingBlockObserver<TOutput>(completedEvent);
var observable = target.AsObservable();
observable.Subscribe(observer);

observable 可以很简单：

private class RetryingBlockObserver<T> : IObserver<T> {
        private ManualResetEvent completedEvent;

        public RetryingBlockObserver(ManualResetEvent completedEvent) {                
            this.completedEvent = completedEvent;
        }

        public void OnCompleted() {
            completedEvent.Set();
        }

        public void OnError(Exception error) {
            //TODO
        }

        public void OnNext(T value) {
            //TODO
        }
    }

您可以等待信号或完成（所有源项目的耗尽），或两者兼而有之

 source.Completion.ContinueWith(async _ => {

            WaitHandle.WaitAll(completedEvent, signal);
            // Or WaitHandle.WaitAny, depending on your needs!

            target.Complete();
        });

您可以检查 WaitAll 的结果值以了解设置了哪个事件，并做出相应的反应。您还可以将其他事件添加到代码中，将它们传递给观察者，以便它可以在需要时设置它们。例如，当出现错误时，您可以区分自己的行为并做出不同的响应

于 2014-09-03T12:28:49.660 回答