长期运行任务与线程 - 性能

Question

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

目前使用Tasks实现.我有一个JobStream在循环中一次运行一个作业.我可以同时运行5,15或50个这些流,具体取决于负载.

JobManager

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

你真的应该使用微软的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;
}

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

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