我怎样才能杀死或停止后台工作人员？

Question

为了快速加载几个大文件，我启动了后台工作程序作为文件数量。

每个后台工作者需要很长时间来分别加载其文件。当他们加载文件时，我想停止所有加载。我了解 backgroundworker.CancelAsync() 向线程发送取消消息，但线程没有时间接受消息。因为每个线程只加载一个文件，并且没有循环操作来检查取消。在这种情况下，我该如何阻止这些后台工作人员？

让我在这里展示我的代码。 //主线程调用50个子线程。

private List<BackgroundWorker> bgws = new List<BackgroundWorker>();
private bool ChildThreadCompleted;
 private void MainThread_DoWork(object sender, DoWorkEventArgs e)
{
// 50 sub threads will be started here
    for (int i=1; i<=50; i++)
    {
        if (mainThread.CancellationPending) return;
        BackgroundWorker childThread = new BackgroundWorker();
        childThread.WorkerSupportsCancellation = true;
        childThread.DoWork += ChildThread_DoWork;
        childThread.RunWorkerCompleted += ChildThread_RunWorkerCompleted;
        bgws.Add(childThread);
        childThread.RunWorkerAsync(i);
    }
    while (!ChildThreadCompleted)
    {
        if (mainThread.CancellationPending)
        {
            foreach (BackgroundWorker thread in bgws)
                if (thread.IsBusy) thread.CancelAsync();
        }
        Application.DoEvents();
    }
}

 private void ChildThread_DoWork(object sender, DoWorkEventArgs e)
{
    int arg = Convert.ToInt32(e.Argument);
    System.Threading.Thread.Sleep(1000);
    BackgroundWorker thread = (BackgroundWorker)sender;
    if (thread.CancellationPending) return;
    // In case of loading the image no longer makes sense, I'd like to stop.
    // At this point, i can't stop this process.
    //loading large file here. Just one image file for example. <= this takes one or more seconds

    e.Result = arg;
}

private void ChildThread_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
{
    BackgroundWorker thread = sender as BackgroundWorker;
    bgws.Remove(thread);
    if (bgws.Count == 0) ChildThreadCompleted = true;
}

Answer 1

短版

BGW 无法做到这一点，除非没有复杂的编码。 ActionBlock 是专门为处理具有取消支持的输入流而构建的。

我使用数据流类每 15 分钟查找、下载和处理数千条机票记录。

加长版

BackgroundWorker 已过时，完全被 TPL 类（如任务、CancellationToken、IPProgress 等）取代。

这个特殊问题最好由更高级别的类ActionBlock 解决。 ActionBlock 和 TPL Dataflow 命名空间中的其他类允许您创建类似于 Powershell 管道的块管道。

每个块都在自己的任务上运行，接收输入并将输出传递给下一个块。您甚至可以指定一个块可以通过使用多个任务来处理多个输入。

ActionBlock 不产生任何输出，它只处理输入。像大多数块一样，它有一个输入缓冲区并支持取消。

文件处理块可能如下所示。 ：

var block=new ActionBlock<string>(file => SomeSlowFileProcessing(file));

var files = Directory.EnumerateFiles(someFolder,somePattern);

//Post all files
foreach(file in files)
{
    //Post doesn't block
    block.Post(file);
}    

当我们完成使用块时，我们应该告诉它我们已经完成了：

block.Complete();

并异步等待，直到处理完所有剩余文件：

await block.Completion;

您可以通过指定最大并发任务数来指示块process multiple messages in parallel：

var options = new ExecutionDataflowBlockOptions
     {
        MaxDegreeOfParallelism = 20
     };

var block=new ActionBlock<string>(file => SomeSlowFileProcessing(file), options);

启动 100 个线程来处理 100 个文件可能会导致线程相互阻塞。通过限制并发任务的数量，您可以确保所有 CPU 都可以执行有用的工作。

您也可以停止阻止。一种方法是通过调用Fault() 方法来“核对”它：

block.Fault();
try 
{
    await block.Completion;
}
catch(Exception exc)
{
     ...
}

这将丢弃输入缓冲区中剩下的所有内容，并将异常传播到管道中的下一个块。就好像块的方法抛出了异常。在这种情况下，没有其他块，await block.Completion; 将抛出。

另一个，more cooperative way 是使用 CancellationTokenSource 来取消块并向工作方法发出信号，它应该取消。

CancellationTokenSource 是一个类，可用于向任何任务、线程或其他代码发出取消信号。它通过提供一个 CancellationToken 来实现，当有人在 CTS 上调用 Cancel() 或其超时间隔到期时，其 IsCancellationRequested 属性变为 true。

这样，您可以通过创建具有超时期限的 CTS 来为您的区块提供超时功能：

var cts=new CancellationTokenSource(60000); //Timeout in 1 minute
var options = new ExecutionDataflowBlockOptions
     {
        MaxDegreeOfParallelism = 20,
        CancellationToken=cts.Token
     };

var block=new ActionBlock<string>(file => SomeSlowFileProcessing(file), options);

//.....

try
{
    await block.Completion;
}
catch (OperationCanceledException)
{
    Console.WriteLine("Timed out!");
}

如果 CTS 存储在字段中，则可以使用按钮事件发出取消信号： CancellationTokenSource _cts; 动作块_block；

public void Start_Click(object sender, EventArgs args)
{
    //Make sure both the CTS and block are created before setting the fields  
    var cts=new CancellationTokenSource(60000); //Timeout in 1 minute
    var token=cts.Token;

    var options = new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 20,
        CancellationToken=token
     };
   var block=new ActionBlock<string>(file => SomeSlowFileProcessing(file,token), 
                                     options);
   //Once preparation is over ...
   _cts=cts;
   _block=block;

   //Start posting files
  ...

}

public async void Cancel_Click(object sender, EventArgs args)
{
   lblStatus.Text = "Cancelling";
    _cts.Cancel();

   try
   {
       await _block.Completion;           
   }
   lblStatus.Text = "Cancelled!";
}

加载大文件并取消

异步文件操作也接受取消令牌，例如 FileStream.ReadAsync 有一个接受 CancellationToken 的重载

这意味着如果worker方法转换为异步方法，则可以取消它，例如

async Task MySlowMethod(string fileName,CancellationToken token)
{
    try 
    {
        using (FileStream SourceStream = File.Open(filename, FileMode.Open))
        {
            var data = new byte[SourceStream.Length];
            await SourceStream.ReadAsync(data, 0, (int)SourceStream.Length,token);
            // Use the data
        }
}