小贝子编程

Rx后退并重试



这是基于本SO中提供的代码：编写Rx"RetryAfter"；扩展方法

我使用的是Markus Olsson的代码（目前仅进行评估），在有人询问之前，我曾试图在Github上联系Markus，但在我工作的地方被阻止了，所以我觉得我唯一能做的就是在so这里询问。很抱歉，如果这对任何人来说都不好的话。

所以我在一个小演示中使用了以下代码：

class Attempt1
{
    private static bool shouldThrow = true;
    static void Main(string[] args)
    {
        Generate().RetryWithBackoffStrategy(3,
               MyRxExtensions.ExponentialBackoff,
            ex =>
            {
                return ex is NullReferenceException;
            }, Scheduler.TaskPool)
            .Subscribe(
                OnNext,
                OnError
            );
        Console.ReadLine();
    }
    private static void OnNext(int val)
    {
        Console.WriteLine("subscriber value is {0} which was seen on threadId:{1}",
            val, Thread.CurrentThread.ManagedThreadId);
    }
    private static void OnError(Exception ex)
    {
        Console.WriteLine("subscriber bad {0}, which was seen on threadId:{1}",
            ex.GetType(),
            Thread.CurrentThread.ManagedThreadId);
    }
    static IObservable<int> Generate()
    {
        return Observable.Create<int>(
            o =>
            {
                Scheduler.TaskPool.Schedule(() =>
                {
                    if (shouldThrow)
                    {
                        shouldThrow = false;
                        Console.WriteLine("ON ERROR NullReferenceException");
                        o.OnError(new NullReferenceException("Throwing"));
                    }
                    Console.WriteLine("Invoked on threadId:{0}",
                        Thread.CurrentThread.ManagedThreadId);
                    Console.WriteLine("On nexting 1");
                    o.OnNext(1);
                    Console.WriteLine("On nexting 2");
                    o.OnNext(2);
                    Console.WriteLine("On nexting 3");
                    o.OnNext(3);
                    o.OnCompleted();
                    Console.WriteLine("On complete");
                    Console.WriteLine("Finished on threadId:{0}",
                        Thread.CurrentThread.ManagedThreadId);
                });
                return () => { };
            });
    }
}
public static class MyRxExtensions
{
    /// <summary>
    /// An exponential back off strategy which starts with 1 second and then 4, 9, 16...
    /// </summary>
    public static readonly Func<int, TimeSpan>
        ExponentialBackoff = n => TimeSpan.FromSeconds(Math.Pow(n, 2));
    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 ?? MyRxExtensions.ExponentialBackoff;
        int attempt = 0;
        return Observable.Defer(() =>
        {
            return ((++attempt == 1) ? source : source.Delay(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));
    }
}

代码对我来说是有意义的，我们尝试并执行一个操作，如果失败，我们会后退并重试。可以指定我们想要重试的异常类型，我们还可以看到订阅者在重试后只看到一次最终值，这是有效的（在上面的演示代码中，只完成了异常（第一次出现OnError））。

所以一般来说，除了一件事之外，代码都能按预期工作。

如果我看看上面代码产生的输出，我会得到：

ON ERROR NullReferenceException 
Invoked on threadId:10 
On nexting 1
Invoked on threadId:11 
On nexting 1 
On nexting 2 
On nexting 3 
On complete 
Finished on threadId:10 
On nexting 2 
On nexting 3 
On complete 
Finished on threadId:11 
subscriber value is 1 which was seen on threadId:10 
subscriber value is 2 which was seen on threadId:10
subscriber value is 3 which was seen on threadId:10

这里对我来说有趣的是，subscriber值都是一次性的，我本以为当调用Generate（）方法中的OnNext时，subscriber OnNext会将OnNext’ed值写入Console输出。

有人能解释为什么会这样吗？

这是因为您将Delay放在结果流上。（在第二次迭代中传递给ExponentialBackoff的n的值为1，给出1秒的延迟。）

Delay在源上运行，但源照常运行Delay将其接收到的结果安排为在指定的持续时间之后发射。因此，订阅者是在Generate的逻辑运行完成后获得结果的。

如果你仔细想想，Delay就是这样——否则Delay会以某种方式干扰上游运营商！

作为一个缓慢的消费者，有可能干扰上游运营商（不抛出异常）。但对于一个简单的Delay来说，这肯定是一种非常糟糕的行为方式。

我不认为Delay是你想要的，因为Delay不会延迟订阅。如果你用DelaySubscription代替，我想你会得到你想要的这也是链接问题中使用的内容

你的问题很好地说明了Delay和DelaySubscription之间的区别！这里也值得一想Defer。

这三者之间的区别是微妙但重要的，所以让我们总结一下这三者：

  • Delay-立即调用目标操作员以获取IObservable，在其Subscribe上立即调用目标上的Subscribe，在指定的Scheduler上调度指定延迟后交付的事件。

  • DelaySubscription-立即呼叫目标操作员以获取IObservable。在其Subscribe上调度目标上的Subscribe以在指定的Scheduler上的指定延迟之后执行。

  • Defer-没有目标运算符。在Subscribe上运行提供的工厂函数以获取目标IObservable，并立即调用Subscribe。没有添加延迟，因此没有指定Scheduler。

对于这篇文章中发生的其他人来说，James World和Brandon的建议确实解决了这个问题（谢谢各位）。

这是完整的工作代码

class Attempt1
{
    private static bool shouldThrow = true;
    static void Main(string[] args)
    {
        Generate().RetryWithBackoffStrategy(3, MyRxExtensions.ExponentialBackoff,
            ex =>
            {
                return ex is NullReferenceException;
            }, Scheduler.TaskPool)
            .Subscribe(
                OnNext,
                OnError
            );
        Console.ReadLine();
    }
    private static void OnNext(int val)
    {
        Console.WriteLine("subscriber value is {0} which was seen on threadId:{1}", 
            val, Thread.CurrentThread.ManagedThreadId);
    }
    private static void OnError(Exception ex)
    {
        Console.WriteLine("subscriber bad {0}, which was seen on threadId:{1}", 
            ex.GetType(),
            Thread.CurrentThread.ManagedThreadId);
    }
    static IObservable<int> Generate()
    {
        return Observable.Create<int>(
            o =>
            {
                Scheduler.TaskPool.Schedule(() =>
                    {
                        if (shouldThrow)
                        {
                            shouldThrow = false;
                            Console.WriteLine("ON ERROR NullReferenceException");
                            o.OnError(new NullReferenceException("Throwing"));
                        }
                        Console.WriteLine("Invoked on threadId:{0}", 
                            Thread.CurrentThread.ManagedThreadId);
                        Console.WriteLine("On nexting 1");
                        o.OnNext(1);
                        Console.WriteLine("On nexting 2");
                        o.OnNext(2);
                        Console.WriteLine("On nexting 3");
                        o.OnNext(3);
                        o.OnCompleted();
                        Console.WriteLine("On complete");
                        Console.WriteLine("Finished on threadId:{0}", 
                            Thread.CurrentThread.ManagedThreadId);
                    });
                return () => { };
            });
    }
}
public static class MyRxExtensions
{
    /// <summary>
    /// An exponential back off strategy which starts with 1 second and then 4, 9, 16...
    /// </summary>
    public static readonly Func<int, TimeSpan> ExponentialBackoff = n => TimeSpan.FromSeconds(Math.Pow(n, 2));
    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 ?? MyRxExtensions.ExponentialBackoff;
        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));
    }
    public static IObservable<T> DelaySubscription<T>(this IObservable<T> source, 
        TimeSpan delay, IScheduler scheduler = null)
    {
        if (scheduler == null)
        {
            return Observable.Timer(delay).SelectMany(_ => source);
        }
        return Observable.Timer(delay, scheduler).SelectMany(_ => source);
    }
}

产生所需输出的

ON ERROR NullReferenceException
Invoked on threadId:11
On nexting 1
On nexting 2
On nexting 3
On complete
Finished on threadId:11
Invoked on threadId:11
On nexting 1
subscriber value is 1 which was seen on threadId:11
On nexting 2
subscriber value is 2 which was seen on threadId:11
On nexting 3
subscriber value is 3 which was seen on threadId:11
On complete
Finished on threadId:11

相关内容

  • 没有找到相关文章

最新更新


热门标签：

javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium