Rxcpp中的调度程序

Question

我正在试图找出Rx的C++ 版本中的调度模型。

了解有一个调度方法的简单接口的C#版本；C++版本看起来相当复杂，包括调度器、工作人员和协调。

对于我来说，缺少的一个主要部分是线程池调度器的实现，它是否与其他名称一起存在？我该如何实现我自己呢？我应该把它写在PPL (视窗)上面吗？如果在上面需要一个序列化(类似于Actor)的观察者，我应该使用什么？查看这里和这里可以表明这不是一个简单的任务。

由于官场文档是自动生成的，而且仍然非常稀疏，这将有助于获得关于这个主题的某种概述。

Answer 1

是的，生成的文档是新的，日程安排还没有记录下来。

rxcpp v2中的调度器基于RxJava使用的调度器和辅助结构(涉及Eric)，RxJava的文档将对调度程序和工作人员进行解释。rxcpp增加了可调度、协调和协调。

scheduler拥有由now()方法公开的时间线。在这个时间线上，scheduler也是worker的工厂。由于调度程序拥有时间线，所以可以构建时间旅行调度器。虚拟调度器是测试调度器的一个基础，它使用这个命令在ms中完成多秒钟的测试。

worker拥有一个用于时间线的挂起的schedulable队列，并且有一个生存期。当到达schedulable的时间时，将运行schedulable。队列保持插入顺序，以便当N个schedulable具有相同的目标时间时，它们按照插入队列的顺序运行。worker保证每个schedulable在下一个schedulable启动之前完成。当取消订阅worker的生存期时，所有挂起的schedulable都会被丢弃。

schedulable拥有一个函数，并且有一个工作人员和一个终身工作人员。当schedulable的生存期被取消订阅时，schedulable函数将不会被调用。schedulable被传递给该函数，并允许该函数重新安排自身或在同一个工作人员上调度其他内容。

新概念是协调和协调。我添加这些代码是为了简化运算符实现，并在运算符实现中引入有偿使用。具体来说，在Rx.NET和RxJava中，操作符使用原子操作和同步原语来协调来自多个流的消息，即使所有流都在同一个线程上(比如UI事件)。默认情况下，rxcpp中的identity_. . .坐标是使用的，并且没有开销。syncronize_. . .和observe_on_. . .协调分别使用互斥和队列对工人，以安全地交织多个流.

coordination是coordinators的工厂，有一个scheduler。

coordinator有一个worker，是一个用于协调observables、subscribers和schedulable函数的工厂。

所有接受多个流或及时处理的操作符(甚至subscribe_on和observe_on)都使用一个协调参数，而不是调度程序。

下面是一些提供的函数，它们将使用特定的调度程序生成协调。

• identity_immediate()
• identity_current_thread()
• Identity_same_worker(工人w)
• serialize_event_loop()
• serialize_new_thread()
• Serialize_same_worker(工人w)
• observe_on_event_loop()
• observe_on_new_thread()

目前还没有线程池调度程序。线程池调度程序需要接受对线程池实现的依赖，因为我不希望编写线程池。我的计划是为windows线程池和apple线程池以及boost asio执行器池制定一个调度程序。要回答的一个问题是，这些特定于平台的构造是否应该驻留在rxcpp回购中，还是应该具有平台特定的repos。

欢迎您的贡献、意见和想法！