c# - 如何将 "Parallel.ForEach"作为后台任务执行，将控制立即返回给调用方法？

Question

如何将“Parallel.ForEach”作为后台任务执行，将控制立即返回给调用方法？

我想使用 C#/Linq“Parallel.Foreach”，但我不想在所有并行任务完成之前继续执行调用方法中的下一条语句。我正在寻找一种方法来从另一个线程异步处理“Parallel.Foreach”的完成。

using System.Threading.Tasks;
using System.Collections.Concurrent;

public class stuff {    
    int[]                jobs;
    ConcurrentQueue<int> done;

    void DoSomething(int job_id) {
       //This takes a long time...
       Sleep(10000);
    }

    void Button_Click() {
        jobs = new int[] {0,20,10,13,12};

        // parallel.foreach...ok great...now all CPUs will 
        // be loaded in parallel with tasks... 
        // … But, How to get this to return 
        // immediately without awaiting completion?
        Parallel.ForEach(jobs, job_i => {
            DoSomething(job_i);
            done.Enqueue(job_i); //tell other thread that job is done
        });
        // now my gui is blocked and unresponsive until this 
        // foreach fully completes...instead, 
        // I want this running in background and returning immediately to gui
        // so that its not hung...
    }
 }

Answer 1

对于初学者来说，我会推荐一个async void EventHandler。

带有await 方法的async eventhandler 将防止UI/主线程被卡住。要从上述异步性中受益，您需要在代码中使用异步/等待。

建议对此进行更改:

void Button_Click() {
    jobs = new int[] {0,20,10,13,12};

    // parallel.foreach...ok great...now all CPUs will 
    // be loaded in parallel with tasks... 
    // … But, How to get this to return 
    // immediately without awaiting completion?
    Parallel.ForEach(jobs, job_i => {
        DoSomething(job_i);
        done.Enqueue(job_i); //tell other thread that job is done
    });
    // now my gui is blocked and unresponsive until this 
    // foreach fully completes...instead, 
    // I want this running in background and returning immediately to gui
    // so that its not hung...
}

为此:

    public async void Button_Click()
    {
        await DoMyWorkAsync();
    }

    private async Task DoMyWorkAsync()
    {
        await Task.Run(() =>
        {
            jobs = new int[] { 0, 20, 10, 13, 12 };

            Parallel.ForEach(jobs, job_i =>
            {
                DoSomething(job_i);
                done.Enqueue(job_i);
            });
        });
    }

注意:可能还有其他注意事项，例如有人双击。但是，要回答最初的问题 - 这应该是您需要的所有更改。请参阅下面的快速而肮脏的 ThreadSafeBoolean。

对于 TAP/C# 纯粹主义者:
如果 Toub、Cleary 或 Skeet 在这里，他们会警告我不要使用 await Task 包装器——但是我发现在现实世界中要维护 Async/Await 模式——偶尔需要这种模式。尽管 Parallel 也支持异步 lambda，但其行为非常难以预测。您可能需要 Yuri 的 NitroEx 或 ForEachAsync 扩展之类的东西。但是，如果您想并行和异步地运行类似这些同步触发和忘记的东西 - 异步任务包装器通常是最简单的解决方案。

下面我演示了使用由 Interlocked 支持的线程安全 bool 值处理双击。对于更具可读性/通用性的代码，我还会考虑 if (Monitor.TryEnter(myObjectLock, 0)) 或 SemaphoreSlim(1,1)。每个都有不同的使用方式。如果我在这种情况下坚持使用一种模式，那么 Monitor.TryEnter 可能是最干净的方法，因为它会返回 false，您可以像线程安全 bool 一样退出。为了提高效率和简单性，我假设如果事件已经在运行，我们将只通过该事件(即什么也不做)。

使用线程安全 bool 值的忙检查示例:

using System.Threading;

// default is false, set 1 for true.
private static int _threadSafeBoolBackValue = 0;

public bool ThreadSafeBusy
{
    get { return (Interlocked.CompareExchange(ref _threadSafeBoolBackValue, 1, 1) == 1); }
    set
    {
        if (value) Interlocked.CompareExchange(ref _threadSafeBoolBackValue, 1, 0);
        else Interlocked.CompareExchange(ref _threadSafeBoolBackValue, 0, 1);
    }
}

public async void Button_Click(object sender, EventArgs e)
{
    if (!ThreadSafeBusy)
    {
        ThreadSafeBusy = true;
        await DoMyWorkAsync();
        ThreadSafeBusy = false;
    }
}

private async Task DoMyWorkAsync()
{
    await Task.Run(() =>
    {
        jobs = new int[] { 0, 20, 10, 13, 12 };

        Parallel.ForEach(jobs, job_i =>
        {
            DoSomething(job_i);
            done.Enqueue(job_i);
        });
    });
}

对于其他性能优化，请考虑并发插入的替代方案，如果工作量很大，请将并发处理的并行 foreach 限制为 new ParallelOptions {MaxDegreeOfParallelism = Environment.ProcessorCount}。

private async Task DoMyWorkAsync()
{
    await Task.Run(() =>
    {
        jobs = new int[] { 0, 20, 10, 13, 12 };

        Parallel.ForEach(
            parallelOptions: new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
            source: jobs,
            body: job_i =>
            {
                DoSomething(job_i);
                done.Enqueue(job_i);
            });
    });
}