支持返回结果的并发ValueTask<T>队列实现方案问询

原有的基于Channel的ASP.NET Core后台任务队列仅支持无返回值的ValueTask工作项，要扩展为支持ValueTask<T>并允许调用方获取执行结果，可通过以下方案实现：

核心实现思路 #

通过TaskCompletionSource 打通后台任务执行结果与调用方的等待逻辑：

• 提交任务时，创建TaskCompletionSource<T>并与工作项绑定，将包装后的无返回值工作项放入队列
• 调用方通过TaskCompletionSource<T>.Task等待任务执行结果
• 后台执行任务时，将结果或异常传递给TaskCompletionSource<T>，完成调用方的等待

1. 定义带返回值的队列接口 #

public interface IBackgroundTaskQueue<T>
{
    // 提交带返回值的工作项，返回可等待结果的Task<T>
    ValueTask<Task<T>> QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask<T>> workItem);
    // 取出队列中的工作项（后台服务调用）
    ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken);
}

2. 实现带返回值的队列 #

public class BackgroundTaskQueue<T> : IBackgroundTaskQueue<T>
{
    private readonly Channel<Func<CancellationToken, ValueTask>> _queue;

    public BackgroundTaskQueue(int capacity)
    {
        var options = new BoundedChannelOptions(capacity)
        {
            FullMode = BoundedChannelFullMode.Wait
        };
        _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
    }

    public async ValueTask<Task<T>> QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask<T>> workItem)
    {
        if (workItem == null)
            throw new ArgumentNullException(nameof(workItem));

        var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);

        // 将带返回值的工作项包装为无返回值的ValueTask，内部处理结果传递
        async ValueTask WrappedWorkItem(CancellationToken ct)
        {
            try
            {
                var result = await workItem(ct);
                tcs.TrySetResult(result);
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                tcs.TrySetCanceled(ct);
            }
            catch (Exception ex)
            {
                tcs.TrySetException(ex);
            }
        }

        await _queue.Writer.WriteAsync(WrappedWorkItem);
        return tcs.Task;
    }

    public async ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken)
    {
        return await _queue.Reader.ReadAsync(cancellationToken);
    }
}

3. 调整后台执行服务 #

原有的后台服务无需大幅修改，只需依赖泛型队列接口即可：

public class VisualizationService : BackgroundService
{
    private readonly IBackgroundTaskQueue<object> _taskQueue;
    private readonly ILogger<VisualizationService> _logger;

    public VisualizationService(IBackgroundTaskQueue<object> taskQueue, ILogger<VisualizationService> logger)
    {
        _taskQueue = taskQueue;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("后台任务服务已启动");

        while (!stoppingToken.IsCancellationRequested)
        {
            var workItem = await _taskQueue.DequeueAsync(stoppingToken);

            try
            {
                await workItem(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "执行后台任务时发生异常");
            }
        }

        _logger.LogInformation("后台任务服务已停止");
    }
}

4. 业务侧使用示例 #

提交带返回值的任务并获取结果：

public class TaskController : ControllerBase
{
    private readonly IBackgroundTaskQueue<string> _taskQueue;

    public TaskController(IBackgroundTaskQueue<string> taskQueue)
    {
        _taskQueue = taskQueue;
    }

    [HttpPost("start-task")]
    public async Task<IActionResult> StartBackgroundTask()
    {
        // 提交带返回值的后台任务
        var resultTask = await _taskQueue.QueueBackgroundWorkItemAsync(async ct =>
        {
            // 模拟耗时业务操作
            await Task.Delay(3000, ct);
            return $"任务执行完成，时间：{DateTime.Now:HH:mm:ss}";
        });

        // 可选：等待结果返回，或返回任务ID让客户端后续查询
        var result = await resultTask;
        return Ok(new { Success = true, Result = result });
    }
}

5. 依赖注入配置 #

在Program.cs中注册队列和后台服务：

// 注册带返回值的队列（示例为string类型，可根据业务替换为其他类型）
builder.Services.AddSingleton<IBackgroundTaskQueue<string>>(sp => new BackgroundTaskQueue<string>(capacity: 15));
// 注册后台执行服务
builder.Services.AddHostedService<VisualizationService>();

扩展说明 #

• 若需支持多种返回类型，可注册多个不同泛型参数的队列实例，或使用object作为通用返回类型
• 如果需要实现“异步提交+后续查询结果”的模式，可扩展队列：为每个任务生成唯一ID，将TaskCompletionSource<T>存储在内存字典中，提供通过ID查询结果的方法
• 原无返回值队列可与带返回值队列共存，根据业务场景按需使用

内容的提问来源于stack exchange，提问作者Евгений