在 VB.NET 中使用 BlockingCollection 添加和删除多线程生产者和消费者

Question

专家，

当您在 Internet 上的任何地方都找不到解决方案时，就会发生这种情况，您只需要破解它，直到它看起来不错 [足够]。

我有一种情况，我需要解析高速传入的对象，但解析需要相对较长的时间——比传入的速度慢得多......我可以很容易地判断一个盒子上有多少个核心，然后我可以划分一组工作线程（任务）来分而治之！但是任何多线程应用程序的问题是“有多少线程？”

没有硬性和快速的答案，所以不要费心去寻找。我的方法是采用一种灵活的方法，我的主线程可以监视它以查看吞吐量（在 X 时间内完成的总工作量）对于它恰好正在运行的机器是否达到最大值。此外，同一台机器的负载、可用 RAM 等可能会随着时间的推移而变化，因此您不能只是设置它然后忘记它...

我只是想回答我自己的问题，这是鼓励。

Answer 1

如果预计解析器一直很忙（或几乎如此），那么拥有比可以处理它们的 CPU 线程更多的解析器线程是没有意义的。当您只有 4 个内核时拥有 10 个解析器线程是没有意义的，因为线程上下文切换会产生开销。所以分配 3 或 4 个为队列服务的工作线程（消费者）。

请参阅https://stackoverflow.com/a/2670568/56778 以确定系统上的逻辑处理器数量。拥有比这更多的工作线程是没有意义的。使用复杂的动态分配工人的方案是否有意义……这是一个见仁见智的问题。我更倾向于拥有一个每分钟检查一次队列状态（项目数）的计时器，并让它适当地分配或取消分配工作线程，每分钟添加或删除一个工作线程以避免超出标记。这可能会很好地利用可用资源。

移除工人将非常容易。如果您创建一个AutoResetEvent，每个线程每次都通过其循环检查，那么第一个看到它的线程可以退出。例如：

private AutoResetEvent _killAThread = new AutoResetEvent(false);

// in each thread
while (!_killAThread.Wait(0) && !cancelToken.IsCancellationRequested)
{
    // thread waits for an item from the queue and processes it.
}

当然，如果队列长时间为空，这可能会导致您有太多线程等待，但我感觉您不希望这种情况经常发生（如果有的话）。

当然，添加新的消费者也很容易。您的计时器滴答处理程序将现有队列大小与某个阈值进行比较，并在需要时启动一个新线程。或者，如果线程过多，它会调用_killATheread.Set()，并且下一个完成其处理的线程将检查事件，查看它是否已设置，然后退出。

Answer 2

[注意：行号指的是在原始博客文章here 中插入的 Crayon for Wordpress 行号。在该链接，您还可以下载整个 VS 2012 解决方案的 7z。]

这就是这个想法：

那么让我们看看我是如何解决阻塞收集的问题，然后将我们的工作交给解析器任务...我使用了一种非常通用且易于重用的生产者和消费者方法。事实上，在我下面的示例中，您甚至可以调整生产者的数量。您可以在下面调整我的代码以更接近您的工作，然后调整线程/任务/消费者的数量以查看并行性对您的效果。

首先是 Imports 中的常见嫌疑人..

Imports System.Threading
Imports System.Threading.Tasks
Imports System.Collections.Concurrent

Module modStartHere
    '
    ' Producer
    '
    Dim itemsToProduce = 10
    Dim sleepProducer = 10 ' in milliseconds
    Dim producerStartID = 1
    Dim producersNumToStart = 1
    Dim ProducerCTSs As New ConcurrentBag(Of CancellationTokenSource)
    Dim moreItemsToAdd As Boolean = True
    '
    ' Consumer
    '
    Dim sleepConsumer = 1000 ' in milliseconds
    Dim consumerStartID = 100
    Dim consumersNumToStart = 3
    Dim ConsumerCTSs As New ConcurrentBag(Of CancellationTokenSource)

生产者最初是按照上述设置的。虽然 itemsToProduce 在程序期间不会改变，但生产者和消费者的数量会改变。这是一个非常粗略的草稿，毫无疑问会在某些时候在您自己的代码中进行精简，但这展示了如何很好地解决这个问题。

我使用“ID”在输出中显示哪个线程在做什么。它们在生产中唯一需要的是 CTS 实例列表：

'
' the multi-thread-safe queue that is produced to and consumed from
'
Dim bc As New BlockingCollection(Of Integer)
'
' this # will be what is actually produced & consumed (1, 2, 3, ...)
'
Dim itemId As Integer = 0
'

这里的主要机器就是那条小线：

将 bc 调暗为新的 BlockingCollection(Of Integer)

微软说：

BlockingCollection 概述 .NET Framework 4.5

BlockingCollection(Of T) 是一个线程安全的集合类，它 提供以下功能：

An implementation of the Producer-Consumer pattern.
Concurrent adding and taking of items from multiple threads.
Optional maximum capacity.
Insertion and removal operations that block when collection is empty or full.
Insertion and removal "try" operations that do not block or that block up to a specified period of time.
Encapsulates any collection type that implements IProducerConsumerCollection(Of T)
Cancellation with cancellation tokens.
Two kinds of enumeration with foreach (For Each in Visual Basic):

-Read-only enumeration.
-Enumeration that removes items as they are enumerated.

itemId 只是一个保存假负载的变量。生产者会将其加一以模拟不同的对象实例或工作单元。您只需更改 BlockingCollection 保存的类型...

现在我不是以 FIFO 方式执行此操作（我将在生产环境中这样做），但您可以按照 Microsoft 的方式执行此操作，甚至可以使用 FILO：

当你创建一个 BlockingCollection(Of T) 对象时，你可以指定 not 只有有限的容量，还有要使用的集合类型。为了 例如，您可以先指定一个 ConcurrentQueue(Of T) 对象 先进先出 (FIFO) 行为，或 ConcurrentStack(Of T) 对象 后进先出 (LIFO) 行为。

现在这很有用！在这个演示中，我不顾一切地做了这一切......但就像我说的，为了我的特定需要，我需要 FIFO，如顶部的图表所示......

稍后，您将看到函数和子例程，但这里真正的魔力在于 2 个集合 - 一个用于生产者，一个用于消费者：

Dim ProducerCTSs 作为新的 ConcurrentBag(Of CancellationTokenSource) Dim ConsumerCTSs 作为新的 ConcurrentBag(Of CancellationTokenSource)

The Magic: As each Task(thread) is either created or closed, the corresponding CancellationTokenSource is either added or removed from the appropriate collection above.

说真的，就是这样！ :)

接下来在代码中，创建初始生产者和消费者：

'===============================
'
' start demo
'
Sub Main()
    '
    '===============================
    '
    ' initial state:
    '
    ' start our producer(s)
    '
    For ps As Integer = producerStartID To producerStartID + producersNumToStart - 1
        CreateTask(ps, "Producer")
    Next
    '
    ' start our consumer(s)
    '
    For cs As Integer = consumerStartID To consumerStartID + consumersNumToStart - 1
        CreateTask(cs, "Consumer")
    Next
    '
    '=========================================

除了几个 Thread.Sleep() 调用之外，下一部分所做的就是添加或删除生产者和消费者任务（线程）。您可以更改顶部的初始值以逐步完成。

创建任务... - CreateTask(, )

要删除一个任务，你（在一行中）都会得到一个随机的 CTS，然后 .Cancel() 它：

GetRandomCTS(ProducerCTSs).Cancel() GetRandomCTS(ConsumerCTSs).Cancel()

GetRandomCTS() 获取 CTS 实例的集合，随机选择一个，然后对其调用 Cancel()。

'
Thread.Sleep(2000)
'
' create a producer
'
Console.WriteLine("creating producer 555...")
CreateTask(555, "Producer")
Thread.Sleep(1000)
'
' cancel a consumer
'
Console.WriteLine("cancelling random consumer...")
GetRandomCTS(ConsumerCTSs).Cancel()
Thread.Sleep(2000)
'
' cancel a consumer
'
Console.WriteLine("cancelling random consumer...")
GetRandomCTS(ConsumerCTSs).Cancel()
Thread.Sleep(1000)
'
' create a consumer
'
Console.WriteLine("creating consumer 222...")
CreateTask(222, "consumer")
Thread.Sleep(1000)
'
' cancel a producer
'
Console.WriteLine("cancelling random producer...")
GetRandomCTS(ProducerCTSs).Cancel()
Thread.Sleep(1000)
'
' cancel a consumer
'
Console.WriteLine("cancelling random consumer...")
GetRandomCTS(ConsumerCTSs).Cancel()
'
'==========================================
'
Console.ReadLine()

结束子

就是这样！

现在是有趣的部分：

#Region "Utilites"
    ''' <summary>
    ''' Retrieves a random cancellation token source from the given list of current threads...
    ''' Works for either producer or consumer
    ''' </summary>
    ''' <param name="ctsBag">ConcurrentBag(Of CancellationTokenSource)</param>
    ''' <returns>CancellationTokenSource</returns>
    ''' <remarks></remarks>
    Function GetRandomCTS(ctsBag As ConcurrentBag(Of CancellationTokenSource)) As CancellationTokenSource
        Dim cts As CancellationTokenSource = Nothing
        Dim rndNum As Random = Nothing
        Dim rndIndex As Integer = Nothing
        Try
            If ctsBag.Count = 1 Then
                Console.WriteLine("There are no threads to cancel!")
            Else
                rndNum = New Random(12345)
                rndIndex = rndNum.Next(0, ctsBag.Count - 1) ' because ElementAt() is zero-based index
                cts = ctsBag.ElementAt(rndIndex)
            End If
        Catch ex As Exception
            Console.WriteLine("GetRandomCTS() Exception: " & ex.StackTrace)
        End Try
        Return cts
    End Function

第 7 行：这就是我们要返回的，一个 CancellationTokenSource

第 16 行：ctsBag.ElementAt() 允许我们按编号提取特定的 CTS 实例。

下面，CreateTask 接受一个参数作为你希望它在运行时显示的#（只是为了演示，看看哪个线程在做什么），以及一个告诉它你是否想要一个新的消费者生产者的字符串。当然，我本可以让它变得更复杂，但这只是一个粗略的草稿。 :)

    Private Function CreateTask(taskId As Integer, taskType As String) As CancellationTokenSource
        Dim t As Task = Nothing
        Dim cts As New CancellationTokenSource()
        Dim token As CancellationToken = cts.Token
        Try
            If taskType.ToLower = "producer" Then
                t = Task.Factory.StartNew(Sub() Producer(taskId, token), token, TaskCreationOptions.LongRunning)
                ProducerCTSs.Add(cts)
            ElseIf taskType.ToLower = "consumer" Then
                t = Task.Factory.StartNew(Sub() Consumer(taskId, token), token, TaskCreationOptions.LongRunning)
                ConsumerCTSs.Add(cts)
            Else
            End If
            Console.WriteLine("{0} Task {1} ({2}) running!", taskType, taskId.ToString("000"), t.Id)
        Catch ex As Exception
            Console.WriteLine("Task {0} CreateTask({1}) Exception: ", taskId.ToString("000"), taskType & ex.StackTrace)
        End Try
        Return cts
    End Function
#End Region

第 7 行和第 10 行：它们调用下面的 Producer() 或 Consumer() 类，向它们传递所需的 CancellationTokenSource，让它们能够在运行时优雅地取消而不会损坏任何数据。

t = Task.Factory.StartNew(Sub() Producer(taskId, token), token, TaskCreationOptions.LongRunning)

您注意到TaskCreationOptions.LongRunning了吗？这对我来说很好，它通过告诉程序不要太担心取消太密切地会发生什么来提高性能。

那么 Producer() 类是什么样的？

#Region "Producer(s)"
    Public Sub Producer(ByVal taskNum As Integer, ByVal ct As CancellationToken)
        ' Was cancellation already requested?  
        If ct.IsCancellationRequested = True Then
            Console.WriteLine("Producer Task {0} was cancelled before Producer thread created!", taskNum.ToString("000"))
            ct.ThrowIfCancellationRequested()
        End If
        '
        'Dim r As Random = New Random(123)
        Dim sw As New Stopwatch
        Dim numAdded As Integer = 0
        sw.Start()
        While moreItemsToAdd = True
            ' Dim itemIn As Integer = r.Next(1, 1000)
            itemId += 1 ' the payload
            Try
                bc.Add(itemId)
                Console.WriteLine("--> " & taskNum.ToString("000") & " --> [+1 => Q has: " & bc.Count & "] added: " & itemId)
                numAdded += 1
                If ct.IsCancellationRequested Then
                    Console.WriteLine("Producer Task {0} cancelled", taskNum.ToString("000"))
                    ct.ThrowIfCancellationRequested()
                End If
                Thread.Sleep(sleepProducer)
            Catch ex As OperationCanceledException
                Console.WriteLine("Task " & taskNum.ToString("000") & " cancelling by request!")
                Exit While
            Catch ex As Exception
                Console.WriteLine("Producer() Exception: " & ex.StackTrace)
            End Try
            If bc.Count >= itemsToProduce Then
                moreItemsToAdd = False
            End If
        End While
        sw.Stop()
        ' Let consumer know we are done.
        Console.WriteLine("Producer stopped adding items! Added " & numAdded & " items in " & sw.Elapsed.TotalSeconds & " seconds!")
        bc.CompleteAdding()
    End Sub
#End Region

我知道，我知道……看起来很复杂！但实际上并非如此。我没那么聪明！ 1/2 代码只是为了捕获和处理取消请求，以便处理不会破坏任何数据。那，还有一个俗气的 StopWatch() 来计时……是的，早期版本的工件仍然被注释掉了。就像我说的“粗糙”...

第 17 行：只需将 itemId（我们的有效负载，可以是任何东西）添加到 BlockingCollection (bc)。

第 20 行：如果取消，我们在这里处理它，而不是函数的随机部分，这可能会破坏各种东西......

第 31 行：我添加了这个作为一种俗气的方式来告诉 Producers 何时停止......生产。这个变量（limit）设置在代码的顶部。

第 38 行：bc.CompleteAdding() - 这是向使用 bc (BlockingCollection) 的每个人发出的信号，表示不会再添加任何项目。这样，消费者就知道何时停止……消费！

“他们为什么要这样做？”

好吧，假设您想要一个或多个短期运行的任务，并且需要知道它们已完成才能继续......是的，就我而言，它们是长期运行的，在生产中我会以“TaskCreationOptions.LongRunning”开始每个任务

Consumer() 类几乎相同，只有一些细微差别：

#Region "Consumer(s)"
    Public Sub Consumer(ByVal taskNum As Integer, ByVal ct As CancellationToken)
        If ct.IsCancellationRequested = True Then ' Was cancellation already requested?
            Console.WriteLine("Consumer Task {0} was cancelled before Consumer thread created!", taskNum.ToString("000"))
            ct.ThrowIfCancellationRequested()
        End If
        Dim totalTaken As Integer = 0
        Dim sw As New Stopwatch
        sw.Start()
        While bc.IsCompleted = False
            Dim itemOut As Integer = Nothing ' the payload
            Try
                itemOut = bc.Take()
                Console.WriteLine("<-- " & taskNum.ToString("000") & " <-- [-1 => Q has: " & bc.Count & "] took: " & itemOut)
                If ct.IsCancellationRequested Then
                    Console.WriteLine("Consumer Task {0} cancelled", taskNum.ToString("000"))
                    ct.ThrowIfCancellationRequested()
                End If
                totalTaken += 1
            Catch ex As OperationCanceledException
                Console.WriteLine("Task " & taskNum.ToString("000") & " cancelling by request!")
                Exit While
            Catch e As InvalidOperationException
                ' IOE means that Take() was called on a completed collection.
                ' In this example, we can simply catch the exception since the  
                ' loop will break on the next iteration.
            End Try
            If (Not itemOut = Nothing) Then
                Thread.Sleep(sleepConsumer)
            End If
        End While
        sw.Stop()
        If bc.IsCompleted = True Then
            Console.WriteLine(vbCrLf & "Task " & taskNum.ToString("000") & " - No more items to take. Took " & totalTaken & " items in " & sw.Elapsed.TotalSeconds & " seconds!")
        End If
    End Sub
#End Region
End Module

第 3 行：在这两个课程中，我们确保在顶部查看我们是否已被取消。这样，如果另一个任务/线程在我们被实例化时完成了最后一项工作，我们就不会浪费时间或资源。

第 13 行：itemOut = bc.Take() - 这里我们抓取下一个项目（取决于 FIFO 或 FILO/LIFO，如上面讨论的配置。这个 BlockingCollection 完成了所有工作！

当你坐下来看看它，这个类的所有其他代码只是为了装饰第 13 行！

所以让我们把这只小狗烧起来吧！

Producer Task 001 (1) running!
--> 001 --> [+1 => Q has: 1] added: 1
<-- 100 <-- [-1 => Q has: 0] took: 1
Consumer Task 100 (2) running!
Consumer Task 101 (3) running!
Consumer Task 102 (4) running!
--> 001 --> [+1 => Q has: 1] added: 2
--> 001 --> [+1 => Q has: 2] added: 3
--> 001 --> [+1 => Q has: 3] added: 4
--> 001 --> [+1 => Q has: 4] added: 5
--> 001 --> [+1 => Q has: 5] added: 6
--> 001 --> [+1 => Q has: 6] added: 7
--> 001 --> [+1 => Q has: 7] added: 8
--> 001 --> [+1 => Q has: 8] added: 9
--> 001 --> [+1 => Q has: 9] added: 10
--> 001 --> [+1 => Q has: 10] added: 11
Producer stopped adding items! Added 11 items in 0.1631605 seconds!
<-- 101 <-- [-1 => Q has: 9] took: 2
<-- 100 <-- [-1 => Q has: 8] took: 3
<-- 101 <-- [-1 => Q has: 7] took: 4
<-- 102 <-- [-1 => Q has: 6] took: 5
creating producer 555...
Producer Task 555 (5) running!
<-- 100 <-- [-1 => Q has: 5] took: 6
Producer stopped adding items! Added 0 items in 1.09E-05 seconds!
<-- 101 <-- [-1 => Q has: 4] took: 7
<-- 102 <-- [-1 => Q has: 3] took: 8
cancelling random consumer...
<-- 100 <-- [-1 => Q has: 2] took: 9
<-- 101 <-- [-1 => Q has: 1] took: 10
<-- 102 <-- [-1 => Q has: 0] took: 11
Consumer Task 102 cancelled
Task 102 cancelling by request!

Task 102 - No more items to take. Took 2 items in 2.0128301 seconds!

Task 100 - No more items to take. Took 4 items in 4.0183264 seconds!

Task 101 - No more items to take. Took 4 items in 4.0007338 seconds!
cancelling random consumer...
creating consumer 222...

Task 222 - No more items to take. Took 0 items in 2.8E-06 seconds!
consumer Task 222 (6) running!
cancelling random producer...
cancelling random consumer...

这是您期望的输出吗？

在下面的链接中为您获取 7z 的整个解决方案...

从HERE下载解决方案！

我花了一段时间才弄清楚整个 CancellationToken 概念，但现在我正在使用它，而且 BlockingCollection 的防弹性，我相信我的应用程序可以每秒处理数百个对象而不会搞砸任何事情。

我的生产应用程序将读取主机上的核心数量，并使用它来设置初始消费者数量。然后我将上下调整，监控完成时间（以聚合方式），从而充分利用主机的资源，了解主机可能与我的应用程序同时做许多其他事情。

谢谢大家！