在进行长时间运行操作之前立即取消操作？

Question

我使用AsParallel与WithDegreeOfParallelism和WithCancellation相结合的方式如下

AsParallel().WithCancellation(cs.Token).WithDegreeOfParallelism(2)

这是我对此的理解。一次只处理两个传入序列。一旦其中一个请求完成，那么将处理更多的项。但是，如果已启动取消请求，则将处理来自传入队列中尚未被拾取的项目。基于这一理解，我创建了以下代码。

class Employee
    {
        public int ID { get; set;}
        public string FirstName { get; set;}
        public string LastName { get; set;}
    }

    class Program
    {

        private static List<Employee> _Employees;
        static CancellationTokenSource cs = new CancellationTokenSource();
        static Random rand = new Random();

        static void Main(string[] args)
        {
            _Employees = new List<Employee>() 
            {
                new Employee() { ID = 1, FirstName = "John", LastName = "Doe" },
                new Employee() { ID = 2, FirstName = "Peter", LastName = "Saul" },
                new Employee() { ID = 3, FirstName = "Mike", LastName = "Sue" },
                new Employee() { ID = 4, FirstName = "Catherina", LastName = "Desoza" },
                new Employee() { ID = 5, FirstName = "Paul", LastName = "Smith" },
                new Employee() { ID = 6, FirstName = "Paul2", LastName = "Smith" },
                new Employee() { ID = 7, FirstName = "Paul3", LastName = "Smith" },
                new Employee() { ID = 8, FirstName = "Paul4", LastName = "Smith" },
                new Employee() { ID = 9, FirstName = "Paul5", LastName = "Smith" },
                new Employee() { ID = 10, FirstName = "Paul6", LastName = "Smith" },
                new Employee() { ID = 5, FirstName = "Paul7", LastName = "Smith" }
            };

            try
            {
                var tasks = _Employees.AsParallel().WithCancellation(cs.Token).WithDegreeOfParallelism(2).Select(x => ProcessThisEmployee(x, cs.Token)).ToArray();
                Console.WriteLine("Now waiting");
                Thread.Sleep(1000);
                cs.Cancel();
                Task.WaitAll(tasks);
            }
            catch (AggregateException ae)
            {
                // error handling code
                Console.WriteLine("something bad happened");
            }
            catch (Exception ex)
            {
                // error handling code
                Console.WriteLine("something even worst happened");
            }
            // other stuff
            Console.WriteLine("All Done");
        }

        private static async Task ProcessThisEmployee(Employee x, CancellationToken token)
        {
            if (token.IsCancellationRequested)
            {
                Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} -> Cancelled", System.Threading.Thread.CurrentThread.ManagedThreadId));
                return;
            }
            int Sleep = rand.Next(800, 2000);
            Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} -> Sleeping for {2}", System.Threading.Thread.CurrentThread.ManagedThreadId, x.ID, Sleep));
            await TaskEx.Run(() => System.Threading.Thread.Sleep(Sleep));

            Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} finished", System.Threading.Thread.CurrentThread.ManagedThreadId, x.ID));
        }

    }

这是我运行它时的输出。

ThreadID =3 ->雇员1 ->睡眠1058 ThreadID =1 ->雇员7 -> 1187年睡眠 ThreadID =1 ->雇员8 -> 1296睡眠 ThreadID =1 ->雇员9 -> 1614年睡眠 ThreadID =1 ->雇员10 -> 1607年睡眠 ThreadID =1 ->雇员5 -> 1928年睡眠 ThreadID =3 ->雇员2 -> 1487年睡眠 ThreadID =3 ->雇员3 -> 1535年睡眠 ThreadID =3 ->雇员4 -> 1265睡眠 ThreadID =3 ->员工5 -> 1248睡眠 ThreadID =3 ->雇员6 ->睡眠807 现在等着 ThreadID =3 ->雇员6完成 ThreadID =4 ->雇员1完成 ThreadID =5 ->雇员7完成 ThreadID =6 ->雇员8完成 ThreadID =3 ->雇员5完成 ThreadID =4 ->雇员9完成 ThreadID =5 ->雇员10完成 ThreadID =6 ->雇员5完成 ThreadID =3 ->雇员4完成 ThreadID =7 ->雇员2完成 ThreadID =8 ->雇员3完成 全都做完了

以下是我的问题(根据我对事物的理解)。

1. 我原以为对一些员工来说，ProcessThisEmployee根本不会被调用，因为它将被取消，但它会被所有员工调用。
2. 即使调用了ProcessThisEmployee方法，它也将经历以下代码路径，这也是没有发生的。 如果( token.IsCancellationRequested ){ Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} ->已取消“)，则返回；}

因此，我更改了ProcessThisEmployee，基本上是在睡眠后移动token.IsCancellationRequested消息，如下所示。

private static async Task ProcessThisEmployee(Employee x, CancellationToken token)
{

    int Sleep = rand.Next(800, 2000);
    Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} -> Sleeping for {2}", System.Threading.Thread.CurrentThread.ManagedThreadId, x.ID, Sleep));
    await TaskEx.Run(() => System.Threading.Thread.Sleep(Sleep));
    if (token.IsCancellationRequested)
    {
        Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} -> Cancelled", System.Threading.Thread.CurrentThread.ManagedThreadId));
        return;
    }
    Console.WriteLine(string.Format("ThreadID = {0} -> Employee {1} finished", System.Threading.Thread.CurrentThread.ManagedThreadId, x.ID));
}

现在我得到了后续输出。

ThreadID = 3 -> Employee 1 -> Sleeping for 1330  
ThreadID = 1 -> Employee 7 -> Sleeping for 1868  
ThreadID = 3 -> Employee 2 -> Sleeping for 903  
ThreadID = 3 -> Employee 3 -> Sleeping for 1241  
ThreadID = 3 -> Employee 4 -> Sleeping for 1367  
ThreadID = 3 -> Employee 5 -> Sleeping for 1007  
ThreadID = 3 -> Employee 6 -> Sleeping for 923  
ThreadID = 1 -> Employee 8 -> Sleeping for 1032  
ThreadID = 1 -> Employee 9 -> Sleeping for 1948  
ThreadID = 1 -> Employee 10 -> Sleeping for 1456  
ThreadID = 1 -> Employee 5 -> Sleeping for 1737  
Now waiting  
ThreadID = 5 -> Employee 2 finished  
ThreadID = 3 -> Employee 6 finished  
something bad happened  
All Done  

我的问题是，我对这个工作流有什么误解？基本上，我想尽快取消手术，而不做长时间的手术(在这种情况下，睡眠只是一个例子，但它可能会非常昂贵)

Answer 1

该代码存在一些问题：

1. ToArray()实现了序列，即只有在源序列的所有输入都通过Select(...)后才会返回。

因为您在那之后调用了cs.Cancel()，所以它不会在token.IsCancellationRequested开始时立即触发ProcessThisEmployee

2. WithDegreeOfParallelism(2).Select(x => ProcessThisEmployee(x, cs.Token))看起来不错，但实际上并没有真正做您希望它做的事情，因为ProcessThisEmployee是一个异步方法，在到达第一个返回或第一个等待时立即返回。

您可能想要做的是执行长时间运行的ProcessThisEmployee方法，只需2级并行。实际上，您要做的是创建一组只具有2级并行性的Tasks。在此之后，任务本身都会同时运行。

我不知道如何解决你的特殊情况，因为我不知道背景。但也许这已经对你有所帮助了。

更新以回复您的评论:我正在做ToArray，ProcessThisEmployee是一个异步方法，因为这段代码将成为库的一部分，可以从WPF应用程序中使用。最终用户可能希望在UI上提供更新，所以我不想阻止直到操作完成(约翰史密斯)

不要为本质上不是异步的东西编写异步包装器，即主要是文件、网络或数据库访问。如果使用库的开发人员想在异步上下文中调用某个东西，他仍然可以执行await Task.Run(...)。有关这方面的更多信息，您可以查看这篇关于您是否使用应该公开同步方法的异步包装器。的文章。

在我看来，如果您已经有了一个工作的LINQ查询，那么PLINQ是非常有用的，并且希望加快它的速度，因为这个查询适合并行处理。

在您的情况下，最简单的方法可能是使用两个线程的工作队列。我很肯定在网络上也有这些例子。