使用Reactive Extensions重试异步任务代码

我可以看到你的代码有几个潜在的问题:

在QueryWithRetryAsync例如调用的方法中将重试逻辑与主逻辑分开.这只是一个设计问题,但仍然存在问题

不要Catch直到之后的Retry.否则,SqlException将导致一个空列表,Retry操作员将永远不会看到异常

我认为这根本不是Throttle必要的,因为你只想通过管道获得一个价值

Retry(1)没有做你认为它做的事情(这对我来说也是一个惊喜).看来"重试"的定义包括第一次调用,所以你需要Retry(2)

这是一个独立的示例,其行为方式符合您的要求:

class Program
{
    static void Main(string[] args)
    {
        var pipeline = Observable
            .Defer(() => DoSomethingAsync().ToObservable())
            .Retry(2)
            .Catch(ex => Observable.Return("default"));

        pipeline
            .Do(Console.WriteLine)
            .Subscribe();

        Console.ReadKey();
    }

    private static int invocationCount = 0;

    private static async Task DoSomethingAsync()
    {
        Console.WriteLine("Attempting DoSomethingAsync");

        await Task.Delay(TimeSpan.FromSeconds(2));

        ++invocationCount;

        if (invocationCount == 2)
        {
            return "foo";
        }

        throw new InvalidOperationException();
    }
}