问

为我的Observable扩展方法选择"默认"调度程序

伊那里发布于 2023-02-09 21:41

扩展

(我见过这个问题,但......)

我正在尝试研究如何为我的Observable扩展方法选择默认调度程序.

第一个例子:如果我不通过使用Scheduler.Default以下方式离开当前线程,则会在"生产"代码中"卡住" :

public static IObservable ResponsiveSample(this IObservable src, 
                                TimeSpan interval, IScheduler scheduler = null)
{
    scheduler = scheduler ?? Scheduler.Default;
    return src.Publish(xs => xs.Take(1).Concat(xs.Sample(interval, scheduler)));
}

第二个例子(从这里偷来).这个在当前线程上是可以的.

public static IObservable RetryAfterDelay(this IObservable source, 
                       TimeSpan dueTime, IScheduler scheduler = null)
{
    return RepeateInfinite(source, dueTime, scheduler).Catch();
}

private static IEnumerable> RepeateInfinite(IObservable src, 
                               TimeSpan dueTime, IScheduler scheduler = null)
{
    yield return source; // Don't delay the first time

    scheduler = scheduler ?? Scheduler.CurrentThread;
    while(true)
        yield return source.DelaySubscription(dueTime, scheduler);
}

问题:是否有经验法则可以帮助我预测哪个默认调度程序可能会导致调用代码的麻烦最少？

1 个回答

摘要

通常,如果操作员调度未来事件(例如,在适当的时间调用调度,或等待某些其他事件响应),或者他们迭代地生成事件,则应提供调度程序重载.

如果源流本质上是阻塞的,那实际上就是源流的业务.是否解决阻止取决于您的扩展 - 但是您不应该在不提供选项(通过调度程序参数)的情况下将非阻塞流转换为阻塞流以避免这种情况.

细节

你引用的问题是一个很好的起点.回顾一下,您可以看到框架将其运算符分为5个类:

AsyncConversions

ConstantTimeOperations

迭代

TailRecursion

TimeBasedOperations

但是,我认为对这些类别的调整使它们对于这个讨论更有用:

产生未来事件(例如缓冲区,窗口,延迟)

迭代(例如,范围,生成,重复)

其他

制作未来活动

Rx库提供的属于此类别的运算符将始终允许您指定调度程序.如果不这样做,他们将使用Scheduler.Default,它将使用始终具有非阻塞语义的平台默认调度程序.基本上这些是将在未来某个时间点安排事件的运营商.需要明确的是,未来的意思是,他们将通过在将来的某个时间点发送事件和/或通过调用Schedule指定到期时间来对传入的事件做出反应.

因此,您几乎不会希望它们被阻塞,因此使用将在不同线程上运行的调度程序(或任何与之相当的平台)是有意义的.

迭代

迭代运算符是那些会产生许多事件的运算符 - 可能会或可能不会在将来安排.例如,Range运算符会调度它的事件以立即分派.然而,一个简单的实现将整个范围立即推到观察者上会有问题 - 因为生成一个范围需要一些非零的时间,你不想删除将范围调度到某个指定的调度程序的选项,特别是在大范围的情况下.此外,您要小心谨慎地将事件推送到调度程序 - 如果您将整个范围转储到一个计划操作中的单个循环中,则可能会不公平地剥夺对调度程序的访问权限,从而对使用该调度程序的时间关键的计划操作产生令人讨厌的后果调度程序(特别是如果它是单线程的).

因此,实现像Range这样的迭代运算符,以便每次迭代负责调度下一次迭代.这意味着它们的行为可能会有很大差异,具体取决于它们使用的调度程序 - 使用即时调度程序,您会看到阻塞,调度程序调度程序您不会看到阻塞或饥饿,因为事件将与其他调度操作交错.

其他类别

剩下的类别是那些操作员立即转换或生成事件 - 因此它们本身是非阻塞的.这些运算符通常不会提供重载以允许您指定调度程序 - 它们只会使用调用它们的线程生成事件 - 这些事件在其生命周期内仍然会有所不同,并且取决于它们应用于的流.如果它们以恒定时间(如Observable.Return)发起事件,它们通常会在Subscribe呼叫返回之前完成.

怎么决定

所以要回答你的问题,你真的需要考虑整个运营商链来做出适当的决定.链中的每个链接都可能引入未来或迭代调度.此外,IObservable<T>没有揭示个体操作员的行为 - 给定的操作员是否包含带有计划参数的过载是一个很好的线索(对于内置操作员来说非常可靠),但不能保证.

指南中有很好的建议(必要的读物)可以帮助您决定是使用还是提供调度程序过载.以下是相关摘录:

5.4.考虑将特定的调度程序传递给并发引入运算符 而不是使用ObserveOn运算符来更改可观察序列生成消息的执行上下文,最好在正确的位置创建并发以开始.由于运算符通过提供调度程序参数重载来参数化并发的引入,因此传递正确的调度程序将导致必须使用ObserveOn运算符的位置更少.

[...]

何时忽略此准则 当组合源自不同执行上下文的多个事件时,请使用准则5.5尽可能晚地将所有消息放在特定的执行上下文中.

5.5.尽可能少地调用ObserveOn运算符 通过使用ObserveOn运算符,将为通过原始可观察序列的每条消息调度一个操作.这可能会改变时序信息,并给系统带来额外的压力.稍后在查询中放置此运算符将减少这两个问题.

[...]

何时忽略本指南如果您对可观察序列的使用未绑定到特定执行上下文,请忽略此准则.在这种情况下,请不要使用ObserveOn运算符.

6.12.避免引入并发性 通过添加并发性,我们可以更改可观察序列的及时性.邮件将安排在稍后到达.传递消息所花费的时间就是数据本身,通过添加并发性来纠正数据.

[...]

何时忽略本指南在引入并发性是运营商所做工作的重要部分的情况下,请忽略此指南.注意:当我们使用立即调度程序或直接从对Subscribe的调用中调用观察者时,我们使订阅调用阻塞.在这种情况下任何昂贵的计算都表明引入并发的候选者.

6.14.运算符不应该阻塞 Rx是一个使用可观察集合编写异步和基于事件的程序的库.通过使运算符阻塞,我们失去了这些异步特性.我们还可能松散可组合性(例如,通过返回类型为T而不是IObservable的值).

根据经验,我在重载中提供了调度程序选项,当且仅当我在实现中使用调度程序时 - 或者因为我直接使用调度程序或者因为我调用具有调度程序重载的运算符.如果它们可用,我总是使用调度程序重载.这样我就可以通过TestScheduler(在nuget包装rx测试中)进行单元测试.

总是值得使用类似我的Spy方法来确定Subscribe调用是否立即返回并且OnXXX观察者的方法是否在不同的线程上调用,或者操作符是否在调用线程上的常量时间运行事件中立即完成并从Subscribe调用返回几乎立即(如在Return) - 如果是这种情况,给定的配置是非阻塞的,Scheduler.Default不是必需的.

对于您的具体示例,源流是关键因素.

在这两个示例中,您都可以通过提供当前线程调度程序来阻止.但是,您是否将阻止取决于源流Subscribe是阻塞还是非阻塞,以及它调度的事件是什么调度程序.

它也取决于你所说的"坚持" - 如果你只是在谈论Subscribe坚持,那么你只需要检查(从最后一个操作员向后应用)如果阻塞Subscribe出现在非阻塞之前.如果您正在讨论导致阻塞的中流事件,您需要考虑这些运算符的工作方式,并且:

通常,如果操作员调度未来事件(例如,在适当的时间调用调度,或等待某些其他事件响应),或者他们迭代地生成事件,则应提供调度程序重载.

如果源流本质上是阻塞的,那实际上就是源流的业务.是否解决阻止取决于您的扩展 - 但是您不应该在不提供选项(通过调度程序参数)的情况下将非阻塞流转换为阻塞流以避免这种情况.

2023-02-09 21:46 回答

入贝_272

撰写答案

今天，你开发时遇到什么问题呢？

立即提问

热门标签

为我的Observable扩展方法选择"默认"调度程序

摘要

细节

制作未来活动

迭代

其他类别

怎么决定