Channels в .NET (System.Threading.Channels) — для чего?

// Неограниченный: запись всегда синхронно успешна
var unbounded = Channel.CreateUnbounded<Job>(new UnboundedChannelOptions
{
    SingleReader = true,                    // оптимизации под одного читателя
    SingleWriter = false
});

// Ограниченный: backpressure
var bounded = Channel.CreateBounded<Job>(new BoundedChannelOptions(capacity: 1_000)
{
    FullMode = BoundedChannelFullMode.Wait, // WriteAsync ждёт места
    SingleReader = false,
    AllowSynchronousContinuations = false   // безопаснее: продолжения не выполняются в потоке писателя
});

// С уведомлением об отброшенных элементах (.NET 6+)
var dropping = Channel.CreateBounded<Metric>(new BoundedChannelOptions(10_000) { FullMode = BoundedChannelFullMode.DropOldest },
    itemDropped: m => droppedCounter.Add(1));

public sealed class EmailQueue
{
    private readonly Channel<EmailJob> _channel = Channel.CreateBounded<EmailJob>(new BoundedChannelOptions(5_000)
    {
        FullMode = BoundedChannelFullMode.Wait
    });

    public ChannelWriter<EmailJob> Writer => _channel.Writer;
    public ChannelReader<EmailJob> Reader => _channel.Reader;
}

builder.Services.AddSingleton<EmailQueue>();
builder.Services.AddHostedService<EmailWorker>();

app.MapPost("/newsletter/subscribe", async (SubscribeRequest req, EmailQueue queue, CancellationToken ct) =>
{
    await queue.Writer.WriteAsync(new EmailJob(req.Email, "welcome"), ct);   // быстро, ответ клиенту не ждёт отправки письма
    return Results.Accepted();
});

public sealed class EmailWorker(EmailQueue queue, IServiceScopeFactory scopes, ILogger<EmailWorker> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // 4 параллельных обработчика читают один канал
        await Parallel.ForEachAsync(queue.Reader.ReadAllAsync(stoppingToken),
            new ParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = stoppingToken },
            async (job, ct) =>
            {
                try
                {
                    await using var scope = scopes.CreateAsyncScope();
                    await scope.ServiceProvider.GetRequiredService<IEmailSender>().SendAsync(job, ct);
                }
                catch (Exception ex) when (ex is not OperationCanceledException)
                {
                    log.LogError(ex, "Не удалось отправить письмо {Email}", job.Email);   // не роняем воркер
                }
            });
    }
}

static ChannelReader<TOut> Stage<TIn, TOut>(ChannelReader<TIn> input, int workers, Func<TIn, CancellationToken, ValueTask<TOut>> transform, CancellationToken ct)
{
    var output = Channel.CreateBounded<TOut>(1_000);
    var tasks = Enumerable.Range(0, workers).Select(_ => Task.Run(async () =>
    {
        await foreach (var item in input.ReadAllAsync(ct))
            await output.Writer.WriteAsync(await transform(item, ct), ct);
    }, ct)).ToArray();

    _ = Task.WhenAll(tasks).ContinueWith(t => output.Writer.Complete(t.Exception), TaskScheduler.Default);  // завершить и передать ошибку дальше
    return output.Reader;
}

async IAsyncEnumerable<List<T>> ReadBatchesAsync<T>(ChannelReader<T> reader, int size, TimeSpan maxWait, [EnumeratorCancellation] CancellationToken ct = default)
{
    var batch = new List<T>(size);
    while (await reader.WaitToReadAsync(ct))
    {
        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
        timeout.CancelAfter(maxWait);
        try
        {
            while (batch.Count < size)
            {
                if (reader.TryRead(out var item)) { batch.Add(item); continue; }
                if (!await reader.WaitToReadAsync(timeout.Token)) break;
            }
        }
        catch (OperationCanceledException) when (!ct.IsCancellationRequested) { /* истекло maxWait — отдать неполный пакет */ }

        if (batch.Count > 0) { yield return batch; batch = new List<T>(size); }
    }
}