c# - 长时间运行的任务与线程——性能

Question

假设我有一些长时间运行的后台作业。每个作业都会做一些工作，然后抓取下一个作业并运行它，一直持续到时间结束。

目前这是使用任务实现的。我有一个 JobStream，它在一个循环中一次运行一个作业。我可以同时运行 5、15 或 50 个这些流，具体取决于负载。

作业管理器

public Task Run(CancellationToken cancellationToken) {
    var jobTasks = Enumerable
        .Range(0, _config.BackgroundProcessor.MaximumSimultaneousJobs)
        .Select(o => JobStream.StartNew(..., () => RunNextJob(cancellationToken), cancellationToken));

    return Task.WhenAll(jobTasks);
}

工作流

public static Task StartNew(Func<Task> nextJobRunner, CancellationToken cancellationToken) {
    var jobStream = new JobStream(nextJobRunner, cancellationToken);

    return jobStream.Start();
}

private Task Start() {
    return Task.Run(async () => {
        do {
            await _nextJobRunner();
        } while (!_cancellationToken.IsCancellationRequested);
    });
}

我的问题是，这里的任务是一个很好的举动，还是我应该只用老式的方式创建线程？我最关心的是性能，并确保作业可以独立运行而不会因为另一个正在努力工作而被束缚。

Answer 1

您确实应该为此使用 Microsoft 的 Reactive Framework(NuGet“System.Reactive”)。它更强大、更简单。

这是一个例子:

void Main()
{
    int number_of_streams = 10;

    IObservable<int> query =
        Observable
            .Range(0, number_of_streams)
            .Select(stream_number =>
                Observable
                    .Defer(() => Observable.Start(() => nextJob(stream_number)))
                    .Repeat())
            .Merge();

    IDisposable subscription =
        query
            .Subscribe(x => Console.WriteLine(x));
}

public int nextJob(int streamNumber)
{
    Thread.Sleep(10000);
    return streamNumber;
}

这会同时运行 10 个流并在每个流中调用 int nextJob(int streamNumber)。我为每个作业模拟了 10 秒的工作，但输出每秒产生一个结果。

此查询在 10 个流上永远重复，直到您调用 subscription.Dispose() 并且它会全部停止。