【发布时间】：2014-07-09 13:53:15
【问题描述】：
所以今天早上我的雷达上突然出现了一系列文章。它以 this question 开头，在 GitHub 上导致 the original example 和 source code。
我稍微重写了一下，所以我可以开始在控制台和服务应用程序中使用它：
public static class Extensions
{
static readonly TaskPoolScheduler Scheduler = new TaskPoolScheduler(new TaskFactory());
// Licensed under the MIT license with <3 by GitHub
/// <summary>
/// An exponential back off strategy which starts with 1 second and then 4, 8, 16...
/// </summary>
[SuppressMessage("Microsoft.Security", "CA2104:DoNotDeclareReadOnlyMutableReferenceTypes")]
public static readonly Func<int, TimeSpan> ExponentialBackoff = n => TimeSpan.FromSeconds(Math.Pow(n, 2));
/// <summary>
/// A linear strategy which starts with 1 second and then 2, 3, 4...
/// </summary>
[SuppressMessage("Microsoft.Security", "CA2104:DoNotDeclareReadOnlyMutableReferenceTypes")]
public static readonly Func<int, TimeSpan> LinearStrategy = n => TimeSpan.FromSeconds(1*n);
/// <summary>
/// Returns a cold observable which retries (re-subscribes to) the source observable on error up to the
/// specified number of times or until it successfully terminates. Allows for customizable back off strategy.
/// </summary>
/// <param name="source">The source observable.</param>
/// <param name="retryCount">The number of attempts of running the source observable before failing.</param>
/// <param name="strategy">The strategy to use in backing off, exponential by default.</param>
/// <param name="retryOnError">A predicate determining for which exceptions to retry. Defaults to all</param>
/// <param name="scheduler">The scheduler.</param>
/// <returns>
/// A cold observable which retries (re-subscribes to) the source observable on error up to the
/// specified number of times or until it successfully terminates.
/// </returns>
[SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope")]
public static IObservable<T> RetryWithBackoffStrategy<T>(
this IObservable<T> source,
int retryCount = 3,
Func<int, TimeSpan> strategy = null,
Func<Exception, bool> retryOnError = null,
IScheduler scheduler = null)
{
strategy = strategy ?? ExponentialBackoff;
scheduler = scheduler ?? Scheduler;
if (retryOnError == null)
retryOnError = e => true;
int attempt = 0;
return Observable.Defer(() =>
{
return ((++attempt == 1) ? source : source.DelaySubscription(strategy(attempt - 1), scheduler))
.Select(item => new Tuple<bool, T, Exception>(true, item, null))
.Catch<Tuple<bool, T, Exception>, Exception>(e => retryOnError(e)
? Observable.Throw<Tuple<bool, T, Exception>>(e)
: Observable.Return(new Tuple<bool, T, Exception>(false, default(T), e)));
})
.Retry(retryCount)
.SelectMany(t => t.Item1
? Observable.Return(t.Item2)
: Observable.Throw<T>(t.Item3));
}
}
现在为了测试它是如何工作的，我写了这个小程序：
class Program
{
static void Main(string[] args)
{
int tryCount = 0;
var cts = new CancellationTokenSource();
var sched = new TaskPoolScheduler(new TaskFactory());
var source = Observable.Defer(
() =>
{
Console.WriteLine("Action {0}", tryCount);
var a = 5/tryCount++;
return Observable.Return("yolo");
});
source.RetryWithBackoffStrategy(scheduler: sched, strategy: Extensions.LinearStrategy, retryOnError: exception => exception is DivideByZeroException);
while (!cts.IsCancellationRequested)
source.Subscribe(
res => { Console.WriteLine("Result: {0}", res); },
ex =>
{
Console.WriteLine("Error: {0}", ex.Message);
},
() =>
{
cts.Cancel();
Console.WriteLine("End Processing after {0} attempts", tryCount);
});
}
}
一开始我以为，订阅事件，会自动触发所有后续的retires。情况并非如此，所以我必须实现一个 Cancellation Token 并循环，直到它发出所有 reties 已用尽的信号。
另一种选择是使用 AutoResetEvent：
class Program
{
static void Main(string[] args)
{
int tryCount = 0;
var auto = new AutoResetEvent(false);
var source = Observable.Defer(
() =>
{
Console.WriteLine("Action {0}", tryCount);
var a = 5/tryCount++;
return Observable.Return("yolo");
});
source.RetryWithBackoffStrategy(strategy: Extensions.LinearStrategy, retryOnError: exception => exception is DivideByZeroException);
while (!auto.WaitOne(1))
{
source.Subscribe(
res => { Console.WriteLine("Result: {0}", res); },
ex =>
{
Console.WriteLine("Error: {0}", ex.Message);
},
() =>
{
Console.WriteLine("End Processing after {0} attempts", tryCount);
auto.Set();
});
}
}
}
在这两种情况下，它都会显示这些行：
Action 0
Error: Attempted to divide by zero.
Action 1
Result: yolo
End Processing after 2 attempts
我要向这群人提出的问题是：这是使用此扩展程序的最佳方式吗？或者有没有办法订阅 Observable 以便它自己重新触发，直到重试次数？
最终更新
根据 Brandon 的建议，这是正确的订阅方式：
internal class Program
{
#region Methods
private static void Main(string[] args)
{
int tryCount = 0;
IObservable<string> source = Observable.Defer(
() =>
{
Console.WriteLine("Action {0}", tryCount);
int a = 5 / tryCount++;
return Observable.Return("yolo");
});
source.RetryWithBackoffStrategy(strategy: Extensions.ExponentialBackoff, retryOnError: exception => exception is DivideByZeroException, scheduler: Scheduler.Immediate)
.Subscribe(
res => { Console.WriteLine("Result: {0}", res); },
ex => { Console.WriteLine("Error: {0}", ex.Message); },
() =>
{
Console.WriteLine("End Processing after {0} attempts", tryCount);
});
}
#endregion
}
输出会略有不同：
Action 0
Action 1
Result: yolo
End Processing after 2 attempts
结果证明这是一个非常有用的扩展。这是另一个如何使用它的示例，其中使用委托给出策略和错误处理。
internal class Program
{
#region Methods
private static void Main(string[] args)
{
int tryCount = 0;
IObservable<string> source = Observable.Defer(
() =>
{
Console.WriteLine("Action {0}", tryCount);
int a = 5 / tryCount++;
return Observable.Return("yolo");
});
source.RetryWithBackoffStrategy(
strategy: i => TimeSpan.FromMilliseconds(1),
retryOnError: exception =>
{
if (exception is DivideByZeroException)
{
Console.WriteLine("Tried to divide by zero");
return true;
}
return false;
},
scheduler: Scheduler.Immediate).Subscribe(
res => { Console.WriteLine("Result: {0}", res); },
ex => { Console.WriteLine("Error: {0}", ex.Message); },
() =>
{
Console.WriteLine("Succeeded after {0} attempts", tryCount);
});
}
#endregion
}
输出：
Action 0
Tried to divide by zero
Action 1
Result: yolo
Succeeded after 2 attempts
【问题讨论】：
标签： c# system.reactive reactive-programming