检查 Parallel.Foreach 中的线程花费的时间，如果花费更多时间则退出

Question

我有 Parallel.Foreach 的代码，它正在处理文件并对每个文件并行执行一些操作。

Parallel.ForEach(lstFiles, file=>
{
 // Doing some operation on file
 // Skip file and move to next if it is taking too long
});

我想跳过一个文件并移动到下一个文件（但不想退出 Parallel.Foreach），如果特定文件花费的时间太长（比如 2 分钟）。 Parallel.Foreach 有什么方法可以检查线程处理单个文件所花费的时间。

谢谢

Answer 1

我建议您不要使用Parallel.ForEach，而是使用 Mirosoft 极其强大的响应式框架。然后你可以这样做：

var query =
    from file in lstFiles.ToObservable()
    from result in Observable.Amb(
        Observable.Start(() => SomeOperation(file)).Select(_ => true),
        Observable.Timer(TimeSpan.FromMinutes(2.0)).Select(_ => false))
    select new { file, result };

IDisposable subscription =
    query
        .Subscribe(x =>
        {
            /* do something with each `new { file, result }`
               as they arrive. */
        }, ex =>
        {
            /* do something if an error is encountered */
            /* (stops processing on first error) */
        }, () =>
        {
            /* do something if they have all finished successfully */
        })

这一切都是并行完成的。 Observable.Amb 运算符启动在其参数列表中定义的两个可观察对象，并获取两者中的任何一个首先产生值的值 - 如果是 Start 可观察对象，它已经处理了您的文件，如果是 Timer 可观察对象，那么2.0 分钟过去了，文件没有结果。

如果您想在处理到一半时停止处理，只需致电subscription.Dispose()。

使用 NuGet "System.Reactive" 获取位。

---

根据 cmets 中的请求以 lambda 形式查询：

var query =
    lstFiles
        .ToObservable()
        .SelectMany(
            file =>
                Observable.Amb(
                    Observable.Start(() => SomeOperation(file)).Select(_ => true),
                    Observable.Timer(TimeSpan.FromMinutes(2.0)).Select(_ => false)),
            (file, result) => new { file, result });