VB.NET多线程文件爬取与批量上传实现方案咨询

问题背景 #

开发了一款VB.NET应用，用于从网络爬取文件并上传至云服务，但批量上传效率极低，初步测试显示完成批量任务需耗时数天甚至数周。计划采用生产者-消费者模式优化：单线程负责爬取文件并加入任务队列，多个工作线程从队列中获取任务执行上传。

前期尝试遇到的问题 #

已基于示例代码实现了包含爬取线程、单个工作线程、监控线程的测试程序，但无法扩展为多工作线程共享队列——尝试将队列封装为独立类时，出现线程间通信异常。

优化实现与效果 #

经调整，改用ConcurrentQueue实现线程安全的任务队列，成功支撑多工作线程共享任务池，测试中已明显观察到上传性能提升。

改进前代码 #

' 改进前单工作线程版本代码
Imports System.Threading

Public Class FileUploadApp
    Private _taskQueue As Queue(Of String)
    Private _lockObj As New Object()
    Private _isRunning As Boolean = True

    Public Sub Start()
        ' 启动爬取线程
        Dim crawlThread As New Thread(AddressOf CrawlFiles)
        crawlThread.Start()

        ' 启动单个工作线程
        Dim uploadThread As New Thread(AddressOf UploadFiles)
        uploadThread.Start()

        ' 启动监控线程
        Dim monitorThread As New Thread(AddressOf MonitorQueue)
        monitorThread.Start()
    End Sub

    Private Sub CrawlFiles()
        While _isRunning
            ' 模拟爬取文件，获取文件路径
            Dim filePath As String = $"C:\TestFiles\File_{Guid.NewGuid()}.txt"
            SyncLock _lockObj
                _taskQueue.Enqueue(filePath)
            End SyncLock
            Thread.Sleep(100) ' 模拟爬取间隔
        End While
    End Sub

    Private Sub UploadFiles()
        While _isRunning
            Dim filePath As String = Nothing
            SyncLock _lockObj
                If _taskQueue.Count > 0 Then
                    filePath = _taskQueue.Dequeue()
                End If
            End SyncLock
            If filePath IsNot Nothing Then
                ' 模拟上传操作
                Console.WriteLine($"上传文件: {filePath}")
                Thread.Sleep(500) ' 模拟上传耗时
            Else
                Thread.Sleep(50) ' 无任务时等待
            End If
        End While
    End Sub

    Private Sub MonitorQueue()
        While _isRunning
            SyncLock _lockObj
                Console.WriteLine($"当前队列任务数: {_taskQueue.Count}")
            End SyncLock
            Thread.Sleep(1000)
        End While
    End Sub
End Class

改进后代码 #

' 改进后多工作线程版本（使用ConcurrentQueue）
Imports System.Threading
Imports System.Collections.Concurrent

Public Class OptimizedFileUploadApp
    Private _taskQueue As New ConcurrentQueue(Of String)()
    Private _isRunning As Boolean = True
    Private Const WorkerThreadCount As Integer = 5 ' 配置工作线程数量

    Public Sub Start()
        ' 启动爬取线程
        Dim crawlThread As New Thread(AddressOf CrawlFiles)
        crawlThread.Start()

        ' 启动多个工作线程
        For i As Integer = 1 To WorkerThreadCount
            Dim uploadThread As New Thread(AddressOf UploadFiles)
            uploadThread.Name = $"UploadWorker_{i}"
            uploadThread.Start()
        Next

        ' 启动监控线程
        Dim monitorThread As New Thread(AddressOf MonitorQueue)
        monitorThread.Start()
    End Sub

    Private Sub CrawlFiles()
        While _isRunning
            ' 模拟爬取文件，获取文件路径
            Dim filePath As String = $"C:\TestFiles\File_{Guid.NewGuid()}.txt"
            _taskQueue.Enqueue(filePath)
            Thread.Sleep(100) ' 模拟爬取间隔
        End While
    End Sub

    Private Sub UploadFiles()
        While _isRunning
            Dim filePath As String = Nothing
            If _taskQueue.TryDequeue(filePath) Then
                ' 模拟上传操作
                Console.WriteLine($"[{Thread.CurrentThread.Name}] 上传文件: {filePath}")
                Thread.Sleep(500) ' 模拟上传耗时
            Else
                Thread.Sleep(50) ' 无任务时等待
            End If
        End While
    End Sub

    Private Sub MonitorQueue()
        While _isRunning
            Console.WriteLine($"当前队列任务数: {_taskQueue.Count}")
            Thread.Sleep(1000)
        End While
    End Sub
End Class

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