Chuyện bắt đầu từ cái endpoint bị timeout

Cách đây không lâu mình có một API upload ảnh: nhận file, ghi metadata, rồi resize ảnh và push notification. Làm hết trong request thì mất 8-10 giây, client timeout sạch. Cách nhanh nhất là đẩy việc nặng ra background, và hồi đó mình chọn Task.Run.

Mọi thứ ổn trên máy dev. Lên production chạy 4 pod thì mới thấy cảnh: task biến mất lúc nào không biết, deploy xong là mất sạch việc đang xử lý dở, không log, không dấu vết.

Sau đó mình chuyển sang System.Threading.Channels. Nó không phải viên đạn bạc, nhưng cho nhu cầu "hàng đợi trong process, có backpressure, shutdown sạch" thì gọn hơn BlockingCollection và an toàn hơn Task.Run treo lơ lửng.

Channel<T> là gì và không là gì

Nói ngắn gọn: Channel<T> là hàng đợi FIFO bất đồng bộ nằm trong RAM của process.

  • Bên ghi dùng WriteAsync, bên đọc dùng ReadAsync hoặc ReadAllAsync.
  • Không persistent: process chết là việc trong hàng đợi mất hết.
  • Không cross-process: mỗi instance có hàng đợi riêng.
  • Có hai loại: bounded (giới hạn sức chứa) và unbounded (không giới hạn).

Chính chữ "bounded" mới là phần đáng nói. Unbounded nghe sướng vì không bao giờ chặn producer, nhưng producer nhanh hơn consumer thì hàng đợi phình dần cho tới khi ăn hết RAM. Một đợt cao điểm là đủ để hiểu vì sao mình không bao giờ dùng unbounded cho đường nhận request nữa.

Code: producer, worker, và backpressure

public sealed class ImageJobQueue : BackgroundService
{
    private readonly Channel<ImageJob> _channel =
        Channel.CreateBounded<ImageJob>(new BoundedChannelOptions(capacity: 500)
        {
            FullMode = BoundedChannelFullMode.Wait, // đầy thì producer phải chờ
            SingleReader = true,
            SingleWriter = false
        });

    public ValueTask EnqueueAsync(ImageJob job, CancellationToken ct)
        => _channel.Writer.WriteAsync(job, ct);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var job in _channel.Reader.ReadAllAsync(stoppingToken))
        {
            try
            {
                await ProcessAsync(job, stoppingToken);
            }
            catch (Exception ex)
            {
                // một job lỗi không được phép giết worker
                _logger.LogError(ex, "Job {JobId} xử lý thất bại", job.Id);
            }
        }
    }
}

Vài điểm mình thấy quan trọng nhất:

  • FullMode = Wait chính là backpressure. Hàng đợi đầy thì WriteAsync chờ, nghĩa là request mới bị chậm lại — đúng ý, vì hệ thống đang quá tải thật.
  • ReadAllAsync đọc cho tới khi channel hoàn tất, khỏi phải tự viết vòng while (await reader.WaitToReadAsync()).
  • Cái try/catch bên trong vòng lặp không phải cho có. Một job ném exception làm văng worker, sau đó không ai đọc hàng đợi nữa, hàng đợi đầy dần rồi WriteAsync của API treo vĩnh viễn. Lỗi này cực kỳ khó thấy khi test local với đúng một job.

Shutdown sao cho không mất việc

BackgroundService khi dừng sẽ cancel stoppingToken, vòng ReadAllAsync thoát, job đang xử lý dở bị bỏ ngang. Nếu muốn xử lý hết việc rồi mới tắt:

public override async Task StopAsync(CancellationToken ct)
{
    _channel.Writer.TryComplete(); // không nhận job mới nữa
    await base.StopAsync(TimeSpan.FromSeconds(20)); // chờ worker đọc hết
}

Ở ASP.NET Core, host mặc định cho 30 giây để shutdown (HostOptions.ShutdownTimeout). Nếu job của bạn có thể chạy lâu hơn, chỉnh timeout đó cho khớp, không thì process bị kill giữa lúc đang xử lý và bạn quay lại đúng vấn đề ban đầu.

Ba lỗi mình từng dính

  • Quên gọi TryComplete() lúc shutdown. Job trong hàng đợi biến mất im lặng, log trống trơn, và phải đọc code mới đoán ra.
  • Chọn unbounded vì "cho nhanh". Cao điểm hàng đợi phình lên hơn 200 nghìn item, RAM nhảy từ 400MB lên gần 3GB. Giờ mặc định của mình là bounded, kèm một metric đếm số item đang chờ để biết hàng đợi có đang đuối không.
  • Test một instance thì ngon, lên nhiều pod thì mỗi pod một hàng đợi riêng. Job nằm ở pod nào chỉ pod đó biết. Nếu cần đảm bảo mỗi job chạy đúng một lần, hoặc job phải sống sót qua restart, thì phải dùng Redis Stream, RabbitMQ hay SQS — không phải Channel.

Khi nào nên dùng Channel

  • Việc nền nhẹ, mất cũng chấp nhận được: ghi log, warm cache, gửi email không quan trọng.
  • Gom nhiều request lại xử lý theo lô. Một pattern mình dùng khá nhiều: đẩy item vô channel, worker flush mỗi 100ms hoặc khi đủ 50 item, thay vì ghi từng cái một.
  • Giới hạn mức xử lý đồng thời, thay vì mở 500 task cùng lúc rồi tự hỏi sao database nghẽn.

Còn khi cần retry bền vững, nhiều service cùng đọc, hoặc job phải chạy tiếp sau khi pod restart — đó là việc của message broker, đừng ép Channel làm chuyện nó không làm được.

📋 Phụ lục thuật ngữ

  • Channel<T>: hàng đợi FIFO bất đồng bộ trong bộ nhớ, thuộc System.Threading.Channels, dùng để chuyển dữ liệu giữa các task mà không cần khóa.
  • Bounded channel: hàng đợi có giới hạn sức chứa. Khi đầy, hành vi do FullMode quyết định.
  • Unbounded channel: không giới hạn sức chứa, rủi ro phình RAM nếu producer nhanh hơn consumer.
  • Backpressure: cơ chế tự làm chậm bên ghi khi hệ thống không theo kịp, thay vì cứ nhận rồi ôm hết vào bộ nhớ.
  • FullMode: quy định điều gì xảy ra khi bounded channel đầy — Wait, DropOldest, DropNewest hay DropWrite.
  • BackgroundService: lớp cơ sở của .NET cho tác vụ chạy nền theo vòng đời host, có ExecuteAsync và StopAsync.
  • Drain: đọc nốt toàn bộ item còn lại trong hàng đợi trước khi tắt hẳn.
  • Graceful shutdown: tắt theo trình tự — ngừng nhận việc mới, xử lý hết việc đang có, rồi mới thoát.

Chọn hàng đợi trong process hay hàng đợi bên ngoài không phải câu hỏi về công nghệ, mà là câu hỏi về việc mất một job thì bạn có sao không. Trả lời được câu đó là biết nên dùng gì.

Xem các bài viết khác