如何在Visual Basic中实现线程安全的泛型Queue(Of Byte)

问题描述 #

需要创建一个线程安全的Byte类型队列：一个线程从串口接收数据并存入队列，主线程读取队列数据并处理。微软.NET文档建议使用Queue.Synchronized()，但该方法不适用于泛型Queue(Of T)。

尝试以下代码时出现错误：

Public Class Form1

#Region "Variables"
    Private RxDataUnsafe As Queue(Of Byte) = New Queue(Of Byte)
    Private RXData As Queue(Of Byte) = Queue(Of Byte).Synchronized(RxDataUnsafe)
#End Region

错误提示：Synchronized is not a member of Queue(Of Byte)

若使用非泛型Queue可以正常运行，但用非泛型存储单个字节存在不必要的装箱拆箱开销，代码如下：

Public Class Form1

#Region "Variables"
    Private RxDataUnsafe As Queue = New Queue
    Private RXData As Queue = Queue.Synchronized(RxDataUnsafe)
#End Region

寻求推荐的线程安全泛型Queue(Of T)实现方式。

推荐实现方式 #

1. 使用ConcurrentQueue(Of T)（官方推荐） #

.NET Framework 4.0及以上版本提供了System.Collections.Concurrent命名空间下的ConcurrentQueue(Of T)，这是原生支持的线程安全泛型队列，内部已处理线程同步逻辑，无需手动加锁：

Imports System.Collections.Concurrent

Public Class Form1
    Private RXData As ConcurrentQueue(Of Byte) = New ConcurrentQueue(Of Byte)

    ' 串口接收线程添加数据
    Sub SerialPortReceiveThread()
        Dim receivedByte As Byte = ' 从串口读取的字节
        RXData.Enqueue(receivedByte)
    End Sub

    ' 主线程读取处理数据
    Sub MainThreadProcess()
        Dim byteToProcess As Byte
        While RXData.TryDequeue(byteToProcess)
            ' 处理byteToProcess
        End While
    End Sub
End Class

Enqueue和TryDequeue方法均为线程安全，适合多线程读写场景，性能优于手动加锁的泛型队列。

2. 手动加锁封装泛型Queue(Of Byte) #

若项目基于.NET Framework 4.0之前的版本，可通过SyncLock手动封装Queue(Of Byte)实现线程安全：

Public Class ThreadSafeByteQueue
    Private ReadOnly _queue As Queue(Of Byte) = New Queue(Of Byte)
    Private ReadOnly _lockObj As Object = New Object()

    Public Sub Enqueue(ByVal item As Byte)
        SyncLock _lockObj
            _queue.Enqueue(item)
        End SyncLock
    End Sub

    Public Function TryDequeue(ByRef item As Byte) As Boolean
        SyncLock _lockObj
            If _queue.Count = 0 Then
                item = 0
                Return False
            End If
            item = _queue.Dequeue()
            Return True
        End SyncLock
    End Function

    Public ReadOnly Property Count As Integer
        Get
            SyncLock _lockObj
                Return _queue.Count
            End SyncLock
        End Get
    End Property
End Class

使用时直接实例化该自定义队列，所有操作通过SyncLock保证线程安全。

3. 使用BlockingCollection(Of T)（生产者-消费者场景适配） #

如果需要实现阻塞式的生产者-消费者模式（比如主线程等待队列有数据再处理），可以使用BlockingCollection(Of T)，它默认封装ConcurrentQueue(Of T)，支持阻塞取数：

Imports System.Collections.Concurrent

Public Class Form1
    Private RXData As BlockingCollection(Of Byte) = New BlockingCollection(Of Byte)

    ' 串口接收线程（生产者）
    Sub SerialPortReceiveThread()
        While True
            Dim receivedByte As Byte = ' 从串口读取字节
            RXData.Add(receivedByte)
            ' 如需停止，调用RXData.CompleteAdding()触发结束
        End While
    End Sub

    ' 主线程（消费者）
    Sub MainThreadProcess()
        ' GetConsumingEnumerable会阻塞直到有数据或CompleteAdding被调用
        For Each byteToProcess In RXData.GetConsumingEnumerable()
            ' 处理byteToProcess
        Next
    End Sub
End Class

这种方式无需手动轮询队列，适合需要阻塞等待数据的场景。

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