Channels trong C# | Adrian Bailador | .NET

Tìm hiểu cách xây dựng ứng dụng producer-consumer hiệu năng cao, an toàn luồng (thread-safe) bằng System.Threading.Channels. Khám phá bounded và unbounded channels, các chiến lược backpressure và các mẫu triển khai thực tế.

Các ứng dụng hiện đại thường xuyên cần xử lý dữ liệu đồng thời, trong đó một hoặc nhiều producer tạo dữ liệu trong khi consumer xử lý dữ liệu đó. Namespace System.Threading.Channels cung cấp một cơ chế mạnh mẽ, an toàn luồng để triển khai các mẫu producer-consumer này mà không cần đến sự phức tạp của việc đồng bộ hóa thủ công.

Channels là gì?

Channels là một lớp trừu tượng cấp cao để truyền dữ liệu giữa producer và consumer một cách bất đồng bộ. Hãy hình dung channel như một hàng đợi an toàn luồng cho phép nhiều luồng gửi và nhận thông điệp mà không cần khóa ngoài hoặc cơ chế đồng bộ hóa.

Các thành phần chính của một channel bao gồm:

• Writer: Đẩy dữ liệu vào channel
• Reader: Kéo dữ liệu ra khỏi channel
• Channel: Ống dẫn quản lý luồng dữ liệu giữa writer và reader

Khác với các bộ sưu tập đồng thời truyền thống, channels được thiết kế riêng cho các mẫu async/await và cung cấp hỗ trợ tích hợp cho việc báo hiệu hoàn thành (completion signalling) và quản lý backpressure.

Khi nào bạn nên sử dụng Channels?

Channels lý tưởng cho các tình huống:

• Bạn cần tách rời (decouple) producer khỏi consumer
• Nhiều luồng hoặc task cần giao tiếp an toàn
• Bạn muốn triển khai các pipeline xử lý nền
• Bạn cần backpressure để ngăn chặn quá tải hệ thống
• Bạn cần một giải pháp thay thế trong bộ nhớ (in-memory) cho hàng đợi thông điệp trong một ứng dụng duy nhất

Channels đặc biệt phù hợp để stream các sự kiện, xử lý các tác vụ nền và xây dựng các pipeline dữ liệu thông lượng cao.

Các loại Channel: Bounded vs Unbounded

Lớp tĩnh Channel cung cấp các phương thức factory để tạo hai loại channel:

Unbounded Channels

Unbounded channel không có giới hạn dung lượng và chấp nhận nhiều thông điệp nhất có thể trong phạm vi bộ nhớ cho phép:

var channel = Channel.CreateUnbounded<string>();

Unbounded channels đơn giản hơn khi sử dụng nhưng tiềm ẩn rủi ro: nếu producer tạo dữ liệu nhanh hơn consumer xử lý, mức sử dụng bộ nhớ có thể tăng không giới hạn.

Bounded Channels

Bounded channel có dung lượng tối đa cố định. Khi đạt đến dung lượng, channel sẽ áp dụng backpressure để kiểm soát luồng dữ liệu:

var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait
});

Bounded channels cung cấp khả năng quản lý tài nguyên tốt hơn và thường được ưu tiên sử dụng trong các tình huống production.

Hiểu về BoundedChannelFullMode

Khi một bounded channel đạt đến dung lượng, thuộc tính FullMode quyết định điều gì sẽ xảy ra:

• Wait: Writer chờ cho đến khi có chỗ trống (mặc định, an toàn nhất)
• DropWrite: Phần tử đang được ghi bị loại bỏ; channel vẫn đầy
• DropOldest: Phần tử cũ nhất trong channel bị loại bỏ để tạo chỗ trống
• DropNewest: Phần tử mới được thêm gần đây nhất bị loại bỏ để tạo chỗ trống

Việc chọn chế độ phù hợp phụ thuộc vào trường hợp sử dụng của bạn:

• Wait: Tốt nhất cho các tác vụ nền khi mọi thông điệp đều quan trọng
• DropOldest: Hữu ích khi chỉ dữ liệu mới nhất là quan trọng (ví dụ: đo từ xa thời gian thực)
• DropNewest: Hiếm khi được sử dụng, nhưng có thể bảo toàn dữ liệu lịch sử
• DropWrite: Khi việc mất một số thông điệp mới là chấp nhận được

Bạn cũng có thể đăng ký một callback để quan sát các phần tử bị loại bỏ:

var channel = Channel.CreateBounded(
    new BoundedChannelOptions(10)
    {
        FullMode = BoundedChannelFullMode.DropOldest
    },
    droppedItem => Console.WriteLine($"Dropped: {droppedItem}")
);

Mẫu Producer-Consumer cơ bản

Dưới đây là một ví dụ cơ bản minh họa mẫu producer-consumer:

using System.Threading.Channels;

var channel = Channel.CreateUnbounded<int>();

// Producer task
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 10; i++)
    {
        await channel.Writer.WriteAsync(i);
        Console.WriteLine($"Produced: {i}");
        await Task.Delay(100);
    }
    channel.Writer.Complete();
});

// Consumer task
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Consumed: {item}");
    }
});

await Task.WhenAll(producer, consumer);

Các điểm chính cần lưu ý:

1. WriteAsync thêm các phần tử vào channel một cách bất đồng bộ
2. ReadAllAsync cung cấp một enumerable bất đồng bộ để tiêu thụ các phần tử
3. Complete() báo hiệu rằng sẽ không còn phần tử nào được ghi nữa
4. Cả hai thao tác đều không chặn (non-blocking) và an toàn luồng

API Writer và Reader

Các phương thức Writer

• WriteAsync(T item): Ghi một phần tử bất đồng bộ; chờ nếu channel đầy
• TryWrite(T item): Cố gắng ghi đồng bộ; trả về false nếu không thể
• WaitToWriteAsync(): Chờ cho đến khi có thể ghi
• Complete(): Đánh dấu channel là hoàn thành; không cho phép ghi thêm
• TryComplete(): Cố gắng đánh dấu hoàn thành; trả về false nếu đã hoàn thành

Các phương thức Reader

• ReadAsync(): Đọc một phần tử bất đồng bộ; chờ nếu channel rỗng
• TryRead(out T item): Cố gắng đọc đồng bộ; trả về false nếu rỗng
• WaitToReadAsync(): Chờ cho đến khi có phần tử hoặc channel hoàn thành
• ReadAllAsync(): Trả về một enumerable bất đồng bộ của tất cả các phần tử
• Completion: Một Task hoàn thành khi channel được đánh dấu hoàn thành

Tích hợp Channels với ASP.NET Core

Channels hoạt động tuyệt vời với dependency injection và background services của ASP.NET Core. Dưới đây là một triển khai thực tế:

Đăng ký Channel

builder.Services.AddSingleton(Channel.CreateBounded<WorkItem>(
    new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.Wait,
        SingleReader = true,
        SingleWriter = false
    }));

Background Service Consumer

public class WorkItemProcessor : BackgroundService
{
    private readonly Channel<WorkItem> _channel;
    private readonly ILogger<WorkItemProcessor> _logger;

    public WorkItemProcessor(
        Channel<WorkItem> channel,
        ILogger<WorkItemProcessor> logger)
    {
        _channel = channel;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Work item processor starting");

        await foreach (var workItem in _channel.Reader.ReadAllAsync(stoppingToken))
        {
            try
            {
                await ProcessWorkItemAsync(workItem, stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing work item {Id}", workItem.Id);
            }
        }
    }

    private async Task ProcessWorkItemAsync(WorkItem item, CancellationToken ct)
    {
        _logger.LogInformation("Processing work item {Id}", item.Id);
        await Task.Delay(100, ct); // Mô phỏng công việc
        _logger.LogInformation("Completed work item {Id}", item.Id);
    }
}

API Endpoint Producer

app.MapPost("/api/work", async (
    WorkItem workItem,
    Channel<WorkItem> channel) =>
{
    await channel.Writer.WriteAsync(workItem);
    return Results.Accepted();
});

Tùy chọn Channel để tinh chỉnh hiệu suất

Cả bounded và unbounded channels đều hỗ trợ các tùy chọn có thể cải thiện hiệu suất trong các tình huống cụ thể:

var options = new BoundedChannelOptions(100)
{
    // Đặt true khi chỉ có một task đọc từ channel
    SingleReader = true,
    
    // Đặt true khi chỉ có một task ghi vào channel
    SingleWriter = false,
    
    // Cho phép các continuation đồng bộ để cải thiện hiệu suất
    // nhưng có thể gây stack dives trong một số tình huống
    AllowSynchronousContinuations = false,
    
    // Hành vi khi channel đầy
    FullMode = BoundedChannelFullMode.Wait
};

Việc thiết lập SingleReader và SingleWriter phù hợp cho phép triển khai channel sử dụng các đường mã được tối ưu hóa, cải thiện thông lượng.

Các phương pháp hay nhất (Best Practices)

Luôn sử dụng các phương thức Async

Channels được thiết kế cho mã bất đồng bộ. Luôn sử dụng await khi ghi hoặc đọc để tránh chặn các luồng:

// Tốt
await channel.Writer.WriteAsync(data, cancellationToken);
await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken))
{
    // Xử lý phần tử
}

// Tránh trừ khi bạn có lý do cụ thể
channel.Writer.TryWrite(data);  // Đồng bộ, có thể mất dữ liệu

Luôn sử dụng Cancellation Tokens

Truyền cancellation tokens để cho phép tắt máy một cách duyên dáng:

await channel.Writer.WriteAsync(item, cancellationToken);
await channel.Reader.ReadAsync(cancellationToken);

Báo hiệu hoàn thành

Luôn gọi Complete() trên writer khi đã sản xuất xong:

try
{
    // Sản xuất các phần tử
}
finally
{
    channel.Writer.Complete();
}

Xử lý ngoại lệ đúng cách

Sử dụng TryComplete(Exception) để truyền lỗi đến consumer:

try
{
    await ProduceItemsAsync(channel.Writer, cancellationToken);
    channel.Writer.Complete();
}
catch (Exception ex)
{
    channel.Writer.TryComplete(ex);
}

Ưu tiên Bounded Channels trong Production

Trừ khi bạn có lý do cụ thể cho unbounded channels, hãy sử dụng bounded channels để ngăn ngừa các vấn đề về bộ nhớ:

// Cấu hình sẵn sàng cho production
var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(capacity)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = true
});

Channels so với các phương pháp khác

• Channels: Các tình huống producer-consumer nhanh, xử lý nền, streaming dữ liệu
• TPL Dataflow: Các pipeline phức tạp có phân nhánh, nối kết và biến đổi
• ConcurrentQueue: Các tình huống đơn giản không cần hỗ trợ async hoặc báo hiệu hoàn thành
• BlockingCollection: Mã kế thừa hoặc khi cần ngữ nghĩa chặn (blocking semantics)
• Message Queues: Giao tiếp liên tiến trình hoặc phân tán

Channels mang lại điểm cân bằng lý tưởng giữa sự đơn giản và sức mạnh, cung cấp hiệu suất tuyệt vời cho các tình huống producer-consumer trong bộ nhớ đồng thời duy trì một API sạch sẽ và thân thiện với async.

Kết luận

System.Threading.Channels cung cấp một nền tảng vững chắc để xây dựng các ứng dụng đồng thời trong .NET. Bằng cách hiểu sự khác biệt giữa bounded và unbounded channels, chọn các hành vi full mode phù hợp và tuân theo các phương pháp hay nhất về lập trình bất đồng bộ, bạn có thể tạo ra các hệ thống producer-consumer hiệu quả và có khả năng mở rộng.

Các điểm mấu chốt cần ghi nhớ:

• Sử dụng bounded channels với dung lượng phù hợp cho khối lượng công việc production
• Chọn FullMode dựa trên việc có chấp nhận mất thông điệp hay không
• Luôn sử dụng các phương thức async và cancellation tokens
• Báo hiệu hoàn thành khi kết thúc quá trình sản xuất
• Cân nhắc các tùy chọn SingleReader/SingleWriter để tối ưu hóa hiệu suất

Hãy bắt đầu đưa channels vào ứng dụng của bạn ngay hôm nay để đơn giản hóa việc lập trình đồng thời và cải thiện thông lượng cũng như khả năng phản hồi của ứng dụng.