C# 异步编程深水区详解Task、ValueTask、线程池饥饿与背压设计

using System.Threading.Channels;
public sealed record ExportJob(Guid JobId, long UserId, DateTime CreatedAt);
public sealed class ExportQueue
{
private readonly Channel<ExportJob> _channel = Channel.CreateBounded<ExportJob>(
new BoundedChannelOptions(500)
{
FullMode = BoundedChannelFullMode.DropWrite,
SingleWriter = false,
SingleReader = false
});
public bool TryEnqueue(ExportJob job) => _channel.Writer.TryWrite(job);
public IAsyncEnumerable<ExportJob> ReadAllAsync(CancellationToken ct) => _channel.Reader.ReadAllAsync(ct);
}
public sealed class ExportWorker : BackgroundService
{
private readonly ExportQueue _queue;
private readonly ILogger<ExportWorker> _logger;
private readonly SemaphoreSlim _concurrency = new(4);
public ExportWorker(ExportQueue queue, ILogger<ExportWorker> logger)
{
_queue = queue;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var job in _queue.ReadAllAsync(stoppingToken))
{
_ = ProcessOneAsync(job, stoppingToken);
}
}
private async Task ProcessOneAsync(ExportJob job, CancellationToken ct)
{
await _concurrency.WaitAsync(ct);
try
{
await Task.Delay(200, ct); // 模拟 I/O
_logger.LogInformation("export done {JobId}", job.JobId);
}
catch (OperationCanceledException)
{
// 正常退出
}
finally
{
_concurrency.Release();
}
}
}

app.MapPost("/api/exports", (ExportQueue queue, long userId) =>
{
var job = new ExportJob(Guid.NewGuid(), userId, DateTime.UtcNow);
return queue.TryEnqueue(job)
? Results.Accepted($"/api/exports/{job.JobId}", new { job.JobId })
: Results.StatusCode(StatusCodes.Status429TooManyRequests);
});