TInput>在ITargetBlock&LT重试的政策;

本文介绍了TInput>在ITargetBlock&LT重试的政策;的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我需要引入一个重试的政策，工作流程。比方说，有一些连接以这样的方式3块：

I need to introduce a retry policy to the workflow. Let's say there are 3 blocks that are connected in such a way:

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个项目在同一时间，然后将结果发送到操作块。

So there is a buffer which accumulates data, then send it to the transform block that processes not more that 3 items at one time, and then the result send to the action block.

潜在在加工过程中变换块瞬态错误是可能的，我想再试，如果该块错误是短暂的几次。

Potentially during processing the transform block transient errors are possible, and I want retry the block if the error is transient for several times.

我知道块一般不重试（与会代表传递到块可以作出重试）。 ，而且其中的一个选项是包装传递给支持重试委托

I know that blocks generally are not retryable (delegates that passed into the blocks could be made retryable). And one of the options is to wrap the delegate passed to support retrying.

我也知道，有一个非常好的图书馆的 TransientFaultHandling.Core 提供重试机制，瞬态故障。这是我的情况下，一个优秀的库，但并非如此。如果我包传递到变换块到 RetryPolicy.ExecuteAsync 法的委托，在的消息的变换块内将被锁定，并直到重试完成或失败，变换块将无法接收新的消息。可以想象，如果所有3消息被输入到重试（比方说，下一个重试尝试将在2分钟）和失败，变换块会卡住，直到至少一个消息离开变换块

I also know that there is a very good library TransientFaultHandling.Core that provides the retry mechanisms to transient faults. This is an excellent library but not in my case. If I wrap the delegate that is passed to the transform block into the RetryPolicy.ExecuteAsync method, the message inside the transform block will be locked, and until retry either completes or fails, the transform block won't be able to receive a new message. Imagine, if all the 3 messages are entered into the retrying (let's say, the next retry attempt will be in 2 minutes) and fail, the transform block will be stuck until at least one message leave the transform block.

我看到的唯一的解决办法是延长 TranformBlock （实际上， ITargetBlock 将有足够太），以及手动重试（比如从这里）：

The only solution I see is to extend the TranformBlock (actually, ITargetBlock will be enough too), and do the retry manually (like from here):

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

i.g。用一个延迟把该消息的变换块内再次，但在这种情况下，重试上下文（左重试次数等），也应该被传递到该块。听起来太复杂了...

i.g. to put the message inside the transform block again with a delay, but in this case the retry context (number of retries left, etc.) also should be passed into this block. Sounds too complex...

有谁看到一个更简单的方法来实现重试政策，工作流程块？

Does anyone see a simpler approach to implement retry policy for a workflow block?

推荐答案

我觉得大家还是要做到这一点，你必须跟踪重试消息的剩余数量，你必须以某种方式安排重试的尝试。

I think you pretty much have to do that, you have to track the remaining number of retries for a message and you have to schedule the retried attempt somehow.

但是你可以通过一个单独的方法封装使其这更好。是这样的：

But you could make this better by encapsulating it in a separate method. Something like:

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

我加入的代码追踪例外，因为我认为失败不应该。被忽视，他们应该是最起码登录

I have added code to track the exceptions, because I think that failures should not be ignored, they should be at the very least logged.

此外，此代码不能与完成工作非常好：如果有重试等待他们的延迟和你完成（）块，它会立即完成，试将丢失。如果这是你一个问题，你将不得不跟踪优秀reties和完整目标在源完成，不进行重等待着。

Also, this code doesn't work very well with completion: if there are retries waiting for their delay and you Complete() the block, it will immediately complete and the retries will be lost. If that's a problem for you, you will have to track outstanding reties and complete target when source completes and no retries are waiting.