使用轮询线程的BlockingCollection出现重复获取元素问题

云原生阿岚

2026-5-29

嘿，我看了你的代码和日志，问题很明显——你自定义的IThreadPool没有正确实现线程安全的元素取出逻辑，导致多个线程都拿到了同一个BatchRecord。咱们一步步来解决这个问题：

首先，先明确你的需求：从日志里的不同BatchId来看，你应该是要做生产-消费的消息队列（每个BatchRecord被处理一次），这种场景完全没必要自己造轮子，.NET自带的BlockingCollection就是专门干这个的，原生支持线程安全，绝不会出现重复取元素的问题。

问题根源

你自己写的IThreadPool(Of BatchRecord)的GetRecord()方法大概率存在两个问题之一：

取出元素时没有从集合中移除它，导致所有线程都能拿到同一个元素；
同步逻辑缺失，多个线程同时进入取元素的代码块，都拿到了同一个实例。
而BlockingCollection的Take()方法是原子操作，能保证每个元素只会被一个线程取走，完美规避这些问题。

解决方案

直接替换掉你的自定义集合，用BlockingCollection来实现：

1. 修改集合定义和相关方法

' 替换原来的IThreadPool，直接用原生的BlockingCollection
Public Shared BatchRecords As New BlockingCollection(Of BatchRecord)()

Public Shared Function GetOne(ThreadName As String) As BatchRecord
    ' Take()是线程安全的：会移除并返回集合中的元素，无元素时自动阻塞等待
    Dim record As BatchRecord = BatchRecords.Take()
    log.WriteLog($"Thread:{ThreadName} picking up Batch ID:{record.Id} {DateTime.Now:yyyy/MM/dd HH:mm:ss}")
    Return record
End Function

Public Shared Sub AddOne(record As BatchRecord)
    ' Add()也是线程安全的，负责向集合中添加元素
    BatchRecords.Add(record)
    log.WriteLog($"Adding BatchId: {record.Id} {DateTime.Now:yyyy/MM/dd HH:mm:ss}")
End Sub

2. 修正GeneratorThread的Run方法

之前的代码里用了MyCollection.Take()，看起来像是笔误，要确保消费线程从正确的集合取元素：

Public Sub Run()
    While True
        ' 调用GetOne方法获取元素，保证线程安全
        Dim record As BatchRecord = YourClassName.GetOne(Thread.CurrentThread.Name)
        ' 这里处理你的业务逻辑
        ' Do Work...
        
        ' 注意：如果是消息队列场景，处理完不需要把元素放回集合；如果是对象池场景，需要用另一种实现
    End While
End Sub

3. 启动任务的代码可保留（按需调整线程数）

你启动20个消费线程的逻辑没问题，但可以根据实际业务压力调整数量，避免线程过多导致上下文切换开销。

额外补充

如果你的需求真的是对象池（复用BatchRecord实例），那应该用ConcurrentBag(Of BatchRecord)来实现，它支持线程安全的添加和取出。但从你的日志来看，你是在处理不同的BatchId数据，所以生产消费队列场景更适合用BlockingCollection。

这样修改后，每个BatchRecord只会被一个线程取走处理，不会再出现多个线程拿到同一个元素的情况了。

内容的提问来源于stack exchange，提问作者EJD