如何为VB.NET异步任务添加25分钟超时取消机制？

由于ProcessingTask是内部类型无法修改，无法直接将CancellationToken传入其异步方法，我们可以通过Task.WhenAny结合CancellationTokenSource实现超时控制，同时在超时后记录错误并尝试取消已启动的任务（若ProcessingJob支持取消操作）。

1. 初始化超时取消令牌 #

首先创建CancellationTokenSource并设置25分钟（1500000毫秒）的超时：

Dim cts As New CancellationTokenSource()
cts.CancelAfter(1500000) ' 25分钟超时

2. 为任务初始化阶段（setupGPJob）添加超时控制 #

原代码直接等待所有初始化任务完成，我们需要将每个初始化任务与超时任务绑定，检测是否超时：

objErrorLogging.LogActivity("Starting processing jobs")

Dim jobListSetup As New List(Of Task(Of ProcessingJob)) From {setupGPJob(ConfigurationManager.AppSettings("GPService"))}
If doForecast Then jobListSetup.Add(setupGPJob(ConfigurationManager.AppSettings("CastService")))

' 为每个初始化任务绑定超时检测
Dim setupTasksWithTimeout As New List(Of Task(Of ProcessingJob))()
For Each setupTask In jobListSetup
    Dim timeoutTask = Task.Delay(Timeout.Infinite, cts.Token).ContinueWith(Function(t)
                                                                                objErrorLogging.LogError(_CLASS_NAME, "SetupJobTimeout", "任务初始化阶段超时", New OperationCanceledException("Setup job timed out after 25 minutes"))
                                                                                Return Nothing
                                                                            End Function, TaskContinuationOptions.OnlyOnCanceled)
    setupTasksWithTimeout.Add(Task.WhenAny(setupTask, timeoutTask).Unwrap())
Next

' 等待所有初始化任务完成或超时
Dim jobList() As ProcessingJob = Await Task.WhenAll(setupTasksWithTimeout)
' 过滤超时返回的空任务
jobList = jobList.Where(Function(j) j IsNot Nothing).ToArray()
If jobList.Length = 0 Then
    objErrorLogging.LogActivity("所有任务初始化超时，终止流程")
    cts.Dispose()
    Return
End If

3. 为任务执行阶段（GetResultAsync）添加超时控制 #

同样对每个任务的执行过程绑定超时检测：

' 为每个任务的执行阶段绑定超时控制
Dim jobResultsList As New List(Of Task(Of ProcessingResult))()
For Each listedJob As ProcessingJob In jobList
    Dim resultTask = listedJob.GetResultAsync()
    Dim timeoutTask = Task.Delay(Timeout.Infinite, cts.Token).ContinueWith(Function(t)
                                                                                objErrorLogging.LogError(_CLASS_NAME, "JobExecutionTimeout", $"任务执行超时: {listedJob.ToString()}", New OperationCanceledException("Job execution timed out after 25 minutes"))
                                                                                ' 若ProcessingJob提供Cancel方法，可在此调用以终止任务
                                                                                ' listedJob.Cancel()
                                                                                Return Nothing
                                                                            End Function, TaskContinuationOptions.OnlyOnCanceled)
    jobResultsList.Add(Task.WhenAny(resultTask, timeoutTask).Unwrap())
Next

' 等待所有执行任务完成或超时
Await Task.WhenAll(jobResultsList)

完整修改后代码 #

objErrorLogging.LogActivity("Starting processing jobs")

' 初始化25分钟超时的取消令牌
Dim cts As New CancellationTokenSource()
cts.CancelAfter(1500000) ' 25分钟 = 25*60*1000=1500000ms

Dim jobListSetup As New List(Of Task(Of ProcessingJob)) From {setupGPJob(ConfigurationManager.AppSettings("GPService"))}
If doForecast Then jobListSetup.Add(setupGPJob(ConfigurationManager.AppSettings("CastService")))

' 处理初始化阶段的超时
Dim setupTasksWithTimeout As New List(Of Task(Of ProcessingJob))()
For Each setupTask In jobListSetup
    Dim timeoutTask = Task.Delay(Timeout.Infinite, cts.Token).ContinueWith(Function(t)
                                                                                objErrorLogging.LogError(_CLASS_NAME, "SetupJobTimeout", "任务初始化阶段超时", New OperationCanceledException("Setup job timed out after 25 minutes"))
                                                                                Return Nothing
                                                                            End Function, TaskContinuationOptions.OnlyOnCanceled)
    setupTasksWithTimeout.Add(Task.WhenAny(setupTask, timeoutTask).Unwrap())
Next

Dim jobList() As ProcessingJob = Await Task.WhenAll(setupTasksWithTimeout)
jobList = jobList.Where(Function(j) j IsNot Nothing).ToArray()
If jobList.Length = 0 Then
    objErrorLogging.LogActivity("所有任务初始化超时，终止流程")
    cts.Dispose()
    Return
End If

' 处理任务执行阶段的超时
Dim jobResultsList As New List(Of Task(Of ProcessingResult))()
For Each listedJob As ProcessingJob In jobList
    Dim resultTask = listedJob.GetResultAsync()
    Dim timeoutTask = Task.Delay(Timeout.Infinite, cts.Token).ContinueWith(Function(t)
                                                                                objErrorLogging.LogError(_CLASS_NAME, "JobExecutionTimeout", $"任务执行超时: {listedJob.ToString()}", New OperationCanceledException("Job execution timed out after 25 minutes"))
                                                                                ' 若ProcessingJob提供Cancel方法，取消任务
                                                                                ' listedJob.Cancel()
                                                                                Return Nothing
                                                                            End Function, TaskContinuationOptions.OnlyOnCanceled)
    jobResultsList.Add(Task.WhenAny(resultTask, timeoutTask).Unwrap())
Next

Await Task.WhenAll(jobResultsList)

' 释放取消令牌资源
cts.Dispose()

关键说明 #

• 因无法修改内部ProcessingTask，采用Task.WhenAny将目标任务与受取消令牌控制的Task.Delay绑定，超时触发时Task.Delay因取消令牌完成，实现超时检测。
• 超时后会记录对应错误日志，若ProcessingJob提供Cancel方法，可在超时回调中调用以终止任务。
• 最后需释放CancellationTokenSource资源，避免内存泄漏。

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