具有可观察的重试机制

Question

我正在尝试使用 C# 可观察量编写重试机制。

重试

1. 具有重试计数和重试间隔
2. 重试应执行"OnExecute"方法。
3. 在每个异常中，它将执行"OnCatch"方法。

这是我尝试做的：

public static IObservable<T> Retry(GenericRetryExecutorRequest<T> request)
{
var source = Observable.Timer(TimeSpan.Zero, request.Interval)
.Select(item =>
{
return request.GenericRetryActions.OnExecute();
});
var retryObservable = source
.Retry(request.RetryCount)
.Catch(source);
return retryObservable;
}
public class GenericRetryExecutorRequest<T>
{
public int RetryCount { get; set; } = 3; 
public TimeSpan Interval { get; set; } = new TimeSpan(0,0,0,5);
public IGenericRetryActions<T> GenericRetryActions { get; set; }
}
public interface IGenericRetryActions<out T>
{
T OnExecute();
void OnCatch();
}

不幸的是 - 它表现不佳：

我不知道

1. 在抛出异常时如何执行OnCatch。 我尝试了很多方法，但没有成功。
2. OnExecute似乎没有重复执行(与请求 间隔(，以防引发异常。

Answer 1

试试这个：

public static IObservable<T> Retry<T>(this GenericRetryExecutorRequest<T> request)
{
return Observable.Timer(Timespan.Zero, request.Interval)
.Select(item =>
{
try
{
var value = request.GenericRetryActions.OnExecute();
return Notification.CreateOnNext(value);
}
catch(Exception e)
{
request.GenericRetryActions.OnCatch();
return Notification.CreateOnError<T>(e);
}
})
.Dematerialize()
.Retry(request.RetryCount);
}

通常，在可观察量中使用 try-catch 是不受欢迎的;最好使用可观察的 On-Error 异常处理。但是，您的接口OnExecute不会返回IObservable<T>，而只是返回T。所以你被迫使用try-catch。如果您要更改接口以返回IObservable<T>，那么我认为这将起作用：

public class GenericRetryExecutorRequest2<T>
{
public int RetryCount { get; set; } = 3;
public TimeSpan Interval { get; set; } = new TimeSpan(0, 0, 0, 5);
public IGenericRetryActions2<T> GenericRetryActions { get; set; }
}
public interface IGenericRetryActions2<out T>
{
IObservable<T> OnExecute();
void OnCatch();
}
public static IObservable<T> Retry2<T>(this GenericRetryExecutorRequest2<T> request)
{
return Observable.Timer(Timespan.Zero, request.Interval)
.SelectMany(_ => request.GenericRetryActions.OnExecute())
.Catch((Exception e) => Observable.Return(Unit.Default)
.Do(_ => request.GenericRetryActions.OnCatch())
.SelectMany(Observable.Throw<T>(e))
)
.Retry(request.RetryCount);
}

这一切都是假设您希望该机制在成功时继续启动。如果没有，请在任一解决方案的末尾添加一个Take(1)。