c# - 取消 BlockingCollection.GetConsumingEnumerable() 并处理剩余的内容

Question

我有一个进程生成工作，第二个进程带有 BlockingCollection<>那消耗了那份工作。当我关闭我的程序时，我需要我的消费者停止消费工作，但我仍然需要快速记录未决但尚未消费的工作。

现在，我的消费者生成了一个线程，该线程具有 foreach (<object> in BlockingCollection.GetConsumingEnumerable())环形。当我停止我的程序时，我的制作人调用 Consumer.BlockingCollection.CompleteAdding() .我发现我的消费者继续处理队列中的所有内容。

谷歌搜索问题告诉我我需要使用 CancellationToken .所以我试了一下:

private void Process () { // This method runs in a separate thread
    try {
        foreach (*work* in BlockingCollection.GetConsumingEnumerable(CancellationToken)) {
            // Consume
        }
    }
    catch (OperationCancelledException) {
        foreach (*work* in BlockingCollection.GetConsumingEnumerable()) {
            // quickly log
        }
    }
}

我的制作人有:

private CancellationTokenSource StopFlag = new CancellationTokenSource ();
MyConsumer.CancellationToken = StopFlag.Token;
// Make the consumer spawn it's consuming thread...
StopFlag.Cancel ();
MyConsumer.BlockingCollection.CompleteAdding ();

当我尝试此操作时，没有任何迹象表明 OperationCancelledException 曾经发生过。

This question试图解释取消 token 的使用，但似乎没有正确使用它。 (论点:如果有效，那么它“足够正确”。)和this question似乎与我的问题完全相同，但没有示例。 (相同 here 。)

所以重申一下:我如何正确使用 CancellationToken在 BlockingCollection.GetConsumingEnumerable()需要注意的是，在使用不同的方法取消队列后，我需要处理队列中的剩余项目？

(我认为我的问题集中在 CancellationToken 的正确使用上。我的测试都没有表明进程实际上被取消了。(StopFlag.IsCancellationRequested 总是等于 false。))

Answer 1

当您将 CancellationToken 传递给 GetConsumingEnumerable 时，它不会抛出请求取消的异常，它只会停止吐出项目。与其捕获异常，不如检查 token :

foreach (var item in BlockingCollection.
    GetConsumingEnumerable(CancellationToken))
{
    //consume item
}
if (CancellationToken.IsCancellationRequested)
    foreach (var item in BlockingCollection)
    {
        //log item
    }

另请注意，如果请求取消，并且可能尚未调用 CompletedAdding，那么您应该只迭代集合，而不是调用 GetConsumingEnumerable。如果您知道生产者将在取消操作时完成添加，那么这不是问题。