取消订阅套接字流后取消特定的子线程或任务

Question

我正在尝试使用以下代码在主任务下运行多线程任务，但无法正常工作。

CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
CancellationToken cancellationToken = cancellationTokenSource.Token;

using (var socketClient = new SocketClient()) {
  Task.Factory.StartNew(() => {
    Console.WriteLine("Starting child task...");
    if (cancellationToken.IsCancellationRequested) {
        Console.WriteLine("Task cancellation requested");
        throw new OperationCanceledException(cancellationToken);
    }

    List<string> pairsList = db.GetPairsList(); //loads from db and it's dynamic
    try {
        // loop through each pair
        Parallel.ForEach(pairsList, pair => {
            Console.WriteLine("Item {0} has {1} characters", pair, pair.Length);
            SubscribeToStream(socketClient);//I'm subscribing to some socket streams 
        });
    } catch (OperationCanceledException ex) {
        Console.WriteLine(ex.Message);
    }
  }, cancellationToken);
}

订阅方法如下：

private void SubscribeToStream(string pair, SocketClient socketClient) {
    var subscribePair = socketClient.SubscribeToPair(pair, data => {             
         //here socket stream returns callback each second for each pair
         Application.Current.Dispatcher.Invoke(() => {
              //if perfect match found then Remove pair from the list
              if(a == b) {
                 db.removePair(pair);
              }
              //after removing refresh new pairs list to see if new pair added or removed from the main list from db..
         }
         //after match, we need to unsubscribe from the current pair
         socketClient.UnsubscribeFromPair(subscribePair.Data);

         //IT CALLS SAME PAIR AGAIN EVEN AFTER UNSUBSCRIBING FROM SOCKET STREAM
    }
}

我想运行一个单独保存子任务的主任务。开始按钮将启动主线程，停止按钮将安全地停止所有线程以及主线程。

每个子线程订阅了一些套接字流，这些套接字流为每对运行。如果找到匹配的数据，那么我将从对列表中删除该对，但由于流订阅或线程可能未成功取消，线程会继续。但是，我可以取消主线程，但我想取消特定线程/任务而不是主任务，并允许主任务与现有或新添加的对一起继续。

我们怎样才能安全地实现这个场景？

Answer 1

对于您要取消的每个线程/任务，您需要将其转发取消标记并反复检查。因此，您需要确保您的内部任务（在 Parallel.ForEach 循环中）也反复检查 cancelToken 并在它被触发时中止。

提示：最好始终保留对已创建任务的引用，以便能够检查它是否真的结束了。所以保留对 Factory.StartNew() 的返回值的引用，稍后使用它来检查任务是否真的结束了。