如何优雅暂停异步方法并行执行以处理HTTP 429响应？

针对你的场景，优先推荐使用Polly库来处理429限流与重试逻辑——这是.NET生态中成熟的稳定性保障方案，无需自行实现底层暂停/重试逻辑，天然支持取消令牌与并行场景。如果坚持自定义实现，也可以使用支持CancellationToken的现代版PauseToken。

方案一：使用Polly（推荐） #

Polly可以自动处理429响应、遵循RetryAfter头、支持并发限流，且所有操作都能响应外部取消令牌。

步骤1：安装依赖 #

Install-Package Polly
Install-Package Polly.Extensions.Http

步骤2：配置HttpClient与Polly策略 #

在依赖注入容器中配置带有Polly策略的HttpClient：

using Polly;
using Polly.Extensions.Http;

// 在DI配置中添加
services.AddHttpClient<MyHttpClient>()
    .AddPolicyHandler(Get429HandlingPolicy());

static IAsyncPolicy<HttpResponseMessage> Get429HandlingPolicy()
{
    // 处理429的重试策略，自动遵循RetryAfter头
    var retryPolicy = HttpPolicyExtensions
        .HandleTransientHttpError()
        .OrResult(msg => msg.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
        .WaitAndRetryAsync(
            retryCount: int.MaxValue, // 可根据需求调整最大重试次数
            sleepDurationProvider: (_, response, _) =>
            {
                if (response.Result.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
                {
                    // 从响应头获取延迟时间，默认1秒
                    var retryAfter = response.Result.Headers.RetryAfter;
                    return retryAfter?.Date - DateTimeOffset.UtcNow ?? TimeSpan.FromSeconds(1);
                }
                // 其他错误使用指数退避
                return TimeSpan.FromSeconds(Math.Pow(2, _));
            });

    // 可选：添加并发限流，避免一次性发起过多请求触发429
    var rateLimitPolicy = Policy.RateLimitAsync(
        rateLimit: 100, // 每秒允许的请求数
        perSecond: 1);

    // 组合策略：先限流，再重试
    return rateLimitPolicy.WrapAsync(retryPolicy);
}

步骤3：简化MyHttpClient实现 #

无需维护自定义暂停逻辑，直接使用注入的HttpClient：

public class MyHttpClient
{
    private readonly HttpClient _client;

    public MyHttpClient(HttpClient client)
    {
        _client = client;
    }

    public async Task<HttpResponseMessage> PostAsJsonAsync<TValue>(string? requestUri, TValue value, CancellationToken cancellationToken = default)
    {
        return await _client.PostAsJsonAsync(requestUri, value, cancellationToken).ConfigureAwait(false);
    }
}

调用方式保持不变 #

Polly会自动处理所有429场景，你的调用代码无需修改：

public async Task SendToAPIAsync<T>(IEnumerable<T> items, CancellationToken cancellationToken = default)
{
    var tasks = items.Select(item => 
        _client.PostAsJsonAsync(ApiUrl, item, cancellationToken));

    await Task.WhenAll(tasks).ConfigureAwait(false);
}

Polly方案优势 #

• 成熟可靠，经过大量生产环境验证，无需自行维护底层逻辑
• 天然支持CancellationToken，所有等待/重试操作都会响应取消
• 自动解析并遵循RetryAfter头，无需手动计算延迟
• 可轻松扩展并发限流、指数退避等策略，优化请求效率
• 线程安全，并行场景下无状态冲突问题

方案二：自定义支持CancellationToken的PauseToken #

如果必须自行实现暂停逻辑，可使用以下线程安全、支持取消的现代版本：

public class PauseTokenSource
{
    private readonly object _lock = new();
    private bool _isPaused;
    private CancellationTokenSource? _pauseCts;

    public PauseToken Token => new(this);

    public bool IsPaused
    {
        get => _isPaused;
        set
        {
            lock (_lock)
            {
                if (_isPaused == value) return;

                _isPaused = value;
                if (!_isPaused)
                {
                    _pauseCts?.Cancel();
                    _pauseCts?.Dispose();
                    _pauseCts = null;
                }
            }
        }
    }

    internal async Task WaitWhilePausedAsync(CancellationToken cancellationToken)
    {
        while (IsPaused)
        {
            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            _pauseCts = linkedCts;

            try
            {
                await Task.Delay(Timeout.Infinite, linkedCts.Token).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                // 外部令牌取消时抛出异常，内部暂停结束则继续循环
                if (cancellationToken.IsCancellationRequested)
                    throw;
            }
        }
    }
}

public readonly struct PauseToken
{
    private readonly PauseTokenSource? _source;

    internal PauseToken(PauseTokenSource source)
    {
        _source = source;
    }

    public bool IsPaused => _source?.IsPaused ?? false;

    public Task WaitWhilePausedAsync(CancellationToken cancellationToken = default)
    {
        return _source?.WaitWhilePausedAsync(cancellationToken) ?? Task.CompletedTask;
    }
}

调整你的PostAsJsonAsync方法 #

public async Task<HttpResponseMessage> PostAsJsonAsync<TValue>(string? requestUri, TValue value, CancellationToken cancellationToken = default)
{
    while (true)
    {
        await PauseToken.WaitWhilePausedAsync(cancellationToken);

        var result = await _client.PostAsJsonAsync(requestUri, value, cancellationToken).ConfigureAwait(false);

        if (result.StatusCode != System.Net.HttpStatusCode.TooManyRequests)
            return result;

        // 设置暂停状态
        PauseSource.IsPaused = true;
        try
        {
            var delay = result.Headers.RetryAfter?.Date - DateTimeOffset.UtcNow ?? TimeSpan.FromSeconds(1);
            await Task.Delay(delay, cancellationToken);
        }
        finally
        {
            PauseSource.IsPaused = false;
        }
    }
}

自定义实现优势 #

• 线程安全，使用锁保护状态变更
• 完全支持CancellationToken，外部取消时立即终止等待
• 暂停结束时自动唤醒所有等待任务
• 避免旧版PauseToken的潜在内存泄漏问题

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