如何让 Rx 回调在 ThreadPool 上运行?

问题描述

您是否希望下面的程序打印 False?

Would you expect that the program below prints False?

using System;
using System.Threading;
using System.Reactive.Linq;
using System.Reactive.Concurrency;

public static class Program
{
    public static void Main()
    {
        Observable
            .Return(1)
            .ObserveOn(ThreadPoolScheduler.Instance)
            .Do(x => Console.WriteLine(Thread.CurrentThread.IsThreadPoolThread))
            .Wait();
    }
}

输出:

False

我不明白的是 ThreadPoolScheduler 用于在 ThreadPool 上调度工作，但显然这不是正在发生的事情.它的名称可能指的是 Rx 内部的其他一些线程池，而不是实际的 System.Threading 命名空间中的nofollow noreferrer">ThreadPool 类.

My undestanding was that the ThreadPoolScheduler is intended for scheduling work on the ThreadPool, but apparently this is not what happening. Its name probably refers to some other thread pool, internal to Rx, and not to the actual ThreadPool class in the System.Threading namespace.

我进行了各种尝试以强制回调在 ThreadPool 上运行，但没有成功.我尝试过的一些事情:

I made various attempts to force the callback to run on the ThreadPool, but with no success. Some things I've tried:

.ObserveOn(Scheduler.Default)

.ObserveOn(DefaultScheduler.Instance)

.ObserveOn(TaskPoolScheduler.Default)

.ObserveOn(new TaskPoolScheduler(Task.Factory))

.ObserveOn(new TaskPoolScheduler(new TaskFactory(TaskScheduler.Default)))

ThreadPoolScheduler.Instance.DisableOptimizations(); // At the start of the program

以上最后一次尝试是在阅读这个 Microsoft 论坛中的问题.无论我尝试过什么，Thread.IsThreadPoolThread 属性不断返回 false.

The last attempt above was after reading this question in Microsoft's forums. No matter what I've tried, the Thread.IsThreadPoolThread property keeps returning false.

我是否应该编写自己的 IScheduler 实现以确保我的代码在 ThreadPool 上运行?对于这样一个微不足道的目标来说，这听起来是一项非常重要的任务.

Am I supposed to wrote my own IScheduler implementation to ensure that my code runs on the ThreadPool? It sounds like a quite significant undertaking for such a trivial goal.

.NET 5.0.1、System.Reactive 5.0.0、C# 9

.NET 5.0.1, System.Reactive 5.0.0, C# 9

推荐答案

你分享的链接居然有解决方案:

The link you shared actually has a solution:

var modScheduler = ThreadPoolScheduler.Instance.DisableOptimizations(new[] { typeof(ISchedulerLongRunning) });
Observable
    .Return(1)
    .ObserveOn(modScheduler)
    .Select(_ => Thread.CurrentThread)
    .Subscribe(t => Console.WriteLine(t.IsThreadPoolThread));

如果你查看ThreadPoolScheduler，你会看到所有的工作都被发送到了 ThreadPool，除了 ScheduleLongRunning.如果禁用该优化，则所有工作都将发送到 ThreadPool.

If you look at the source of ThreadPoolScheduler, you'll see that all the work gets sent to the ThreadPool, except for ScheduleLongRunning. If you disable that optimization, then all work will get sent to the ThreadPool.

这篇关于如何让 Rx 回调在 ThreadPool 上运行?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！