Vấn đề thực tế: Sự cố Out-of-Memory lúc 2:14 sáng
Điện thoại của tôi rung lên trên đầu giường lúc 2:14 sáng. PagerDuty thông báo microservice xử lý thanh toán chính đã bị sập do lỗi OutOfMemoryException chưa được bắt. Khi mở Grafana, biểu đồ resident memory dựng đứng—tăng vọt từ 320 MB lên 6.2 GB chỉ trong vài phút. Mức sử dụng CPU dao động sát ngưỡng 100%, và tình trạng thread pool starvation đang kéo sập toàn bộ các HTTP endpoint liên quan.
Nguyên nhân gốc rễ rất đơn giản: một đợt flash sale marketing lúc nửa đêm đã dội vào API của chúng tôi 25.000 yêu cầu giao dịch mỗi giây. Các controller trong ASP.NET Core tiếp nhận mọi payload gửi đến và lập tức đẩy vào một ConcurrentQueue<PaymentEvent> không giới hạn (unbounded), với kỳ vọng rằng một background task sẽ kịp ghi dữ liệu xuống PostgreSQL.
Thế nhưng, PostgreSQL chỉ đạt giới hạn tối đa 4.200 lượt ghi mỗi giây. Trong vòng tám phút, hơn 9,9 triệu đối tượng chưa được xử lý đã làm tràn ngập bộ nhớ heap. Điều này kích hoạt các đợt tạm dừng liên tục của Gen-2 Garbage Collection trước khi Kubernetes buộc phải kill pod.
Các kỹ sư thường nghĩ ngay đến RabbitMQ, AWS SQS hoặc Apache Kafka ngay khi gặp sự cố kiểu này. Tuy nhiên, việc bổ sung một distributed broker kéo theo chi phí thực tế: gánh nặng vận hành, độ trễ network round-trip, chi phí tuần tự hóa (serialization) payload và hóa đơn hạ tầng tăng hàng tháng.
Khi luồng xử lý của bạn chỉ diễn ra trong cùng một tiến trình (in-process)—chẳng hạn như đệm log event, tổng hợp audit trail hoặc ghi batch xuống database—bạn không nhất thiết phải dùng message broker. Thứ bạn cần là in-process backpressure. Đó chính là lý do System.Threading.Channels ra đời.
Khái niệm cốt lõi: Tại sao Channel<T> vượt trội hơn các Collection truyền thống
Trước khi .NET giới thiệu System.Threading.Channels, các lập trình viên thường phải lựa chọn giữa ConcurrentQueue<T> và BlockingCollection<T>. Dưới áp lực tải thực tế ở môi trường production, cả hai đều bộc lộ những hạn chế lớn:
- ConcurrentQueue<T> không có cơ chế giới hạn dung lượng (bounded) tích hợp sẵn và thiếu cơ chế bất đồng bộ (async wait primitive). Phía consumer phải spin hoặc sleep trong vòng lặp, gây lãng phí chu kỳ CPU trong lúc chờ việc mới.
- BlockingCollection<T> hỗ trợ giới hạn dung lượng cố định, nhưng API của nó lại block thread đồng bộ thông qua
Take(). Việc giữ các worker thread ở trạng thái chờ khi lưu lượng tăng đột biến sẽ nhanh chóng làm cạn kiệt .NET ThreadPool và khiến độ trễ của các HTTP request tăng vọt.
Channel<T> giải quyết triệt để cả hai vấn đề trên. Nó cung cấp cơ chế giao tiếp thread-safe, hoàn toàn bất đồng bộ và được phân tách rõ ràng thành hai contract:
ChannelWriter<T>: Phía producer đẩy dữ liệu vào bằngWriteAsync()hoặc hàm non-blockingTryWrite().ChannelReader<T>: Phía consumer lấy dữ liệu ra quaWaitToReadAsync()vàReadAllAsync().
Sức mạnh thực sự ở đây là khả năng điều tiết lưu lượng có giới hạn (bounded flow control). Khi một bounded channel bị đầy, WriteAsync() không hề block OS thread hiện tại. Thay vào đó, nó nhường quyền thực thi (yield) thông qua cơ chế task continuation tiêu chuẩn cho đến khi consumer giải phóng bớt dung lượng trống.
Một khi đã từng chứng kiến cảnh hàng đợi mất kiểm soát làm sập cả một node, bạn sẽ hiểu rằng backpressure không phải là tùy chọn. Đó là ranh giới sống còn giúp các dịch vụ thông lượng cao trụ vững.
Thực hành: Xây dựng In-Memory Pipeline bền bỉ
Hãy cùng thiết kế lại engine tiếp nhận thanh toán. Chúng ta sẽ cấu hình một bounded channel, viết một controller async producer và tiêu thụ hàng đợi bằng một BackgroundService.
Bước 1: Cấu hình các tùy chọn cho Channel
Trước tiên, khởi tạo bounded channel và cấu hình hành vi của nó khi bị đầy.
using System.Threading.Channels;
public static class ChannelFactory
{
public static Channel<PaymentEvent> CreatePaymentChannel(int capacity = 10_000)
{
var options = new BoundedChannelOptions(capacity)
{
// Áp dụng backpressure bằng cách tạm dừng các writer bất đồng bộ
FullMode = BoundedChannelFullMode.Wait,
// Nhiều web request thread đẩy dữ liệu đồng thời
SingleWriter = false,
// Chỉ có một background worker tiêu thụ hàng đợi
SingleReader = true,
// Tránh deadlock do synchronous continuation trên writing thread
AllowSynchronousContinuations = false
};
return Channel.CreateBounded<PaymentEvent>(options);
}
}
public record PaymentEvent(Guid TransactionId, decimal Amount, DateTime CreatedAt);Thiết lập SingleReader = true báo cho channel biết để bỏ qua các cơ chế khóa đồng bộ hóa giữa nhiều consumer. Thay đổi nhỏ này giúp loại bỏ atomic contention trong vòng lặp tiêu thụ dữ liệu và giảm tình trạng CPU cache-line bouncing.
Bước 2: Triển khai Producer (API Controller)
Inject ChannelWriter<PaymentEvent> vào API controller. Khi có một giao dịch thanh toán gửi đến, hãy đẩy nó vào bộ đệm một cách bất đồng bộ.
using System.Threading.Channels;
using Microsoft.AspNetCore.Mvc;
[ApiController]
[Route("api/[controller]")]
public class PaymentsController : ControllerBase
{
private readonly ChannelWriter<PaymentEvent> _writer;
private readonly ILogger<PaymentsController> _logger;
public PaymentsController(ChannelWriter<PaymentEvent> writer, ILogger<PaymentsController> logger)
{
_writer = writer;
_logger = logger;
}
[HttpPost]
public async Task<IActionResult> ProcessPayment([FromBody] PaymentRequest request, CancellationToken ct)
{
var paymentEvent = new PaymentEvent(Guid.NewGuid(), request.Amount, DateTime.UtcNow);
// Đợi bất đồng bộ nếu bộ đệm 10.000 phần tử đã đầy
await _writer.WriteAsync(paymentEvent, ct);
return Accepted(new { paymentEvent.TransactionId, Status = "Đã xếp hàng" });
}
}
public record PaymentRequest(decimal Amount);Hãy xem điều gì xảy ra khi số lượng giao dịch chờ xử lý chạm mốc 10.000. WriteAsync sẽ trả request thread về lại cho Kestrel thay vì chiếm giữ nó. Các HTTP request gửi đến chỉ đơn giản là đợi đến lượt mà không làm phình bộ nhớ hay gây cạn kiệt worker thread.
Bước 3: Triển khai Consumer Background Worker
Bây giờ hãy xây dựng một BackgroundService để lấy các phần tử từ reader và chuyển tiếp chúng xuống hạ tầng xử lý phía sau.
using System.Threading.Channels;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
public class PaymentProcessorService : BackgroundService
{
private readonly ChannelReader<PaymentEvent> _reader;
private readonly ILogger<PaymentProcessorService> _logger;
public PaymentProcessorService(ChannelReader<PaymentEvent> reader, ILogger<PaymentProcessorService> logger)
{
_reader = reader;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("Worker xử lý thanh toán đã khởi động.");
try
{
// ReadAllAsync trả về các phần tử ngay khi chúng đến và hoàn tất khi channel đóng
await foreach (PaymentEvent payment in _reader.ReadAllAsync(stoppingToken))
{
await ProcessPaymentAsync(payment, stoppingToken);
}
}
catch (OperationCanceledException)
{
_logger.LogInformation("Dịch vụ xử lý thanh toán đang dừng một cách an toàn.");
}
}
private async Task ProcessPaymentAsync(PaymentEvent payment, CancellationToken ct)
{
// Mô phỏng độ trễ khi lưu vào cơ sở dữ liệu
await Task.Delay(2, ct);
_logger.LogDebug("Giao dịch {Id} đã được xử lý.", payment.TransactionId);
}
}Bước 4: Đăng ký dịch vụ với Dependency Injection
Đăng ký channel dưới dạng singleton trong Program.cs, tách riêng reader và writer để đảm bảo ranh giới kiến trúc rõ ràng.
var builder = WebApplication.CreateBuilder(args);
var paymentChannel = ChannelFactory.CreatePaymentChannel(capacity: 10_000);
builder.Services.AddSingleton(paymentChannel.Reader);
builder.Services.AddSingleton(paymentChannel.Writer);
builder.Services.AddHostedService<PaymentProcessorService>();
builder.Services.AddControllers();
var app = builder.Build();
app.MapControllers();
app.Run();Checklist cho môi trường Production và những điểm mấu chốt
Việc chuyển từ một hàng đợi in-memory lỏng lẻo sang Channel<T> đã chấm dứt hoàn toàn những cảnh báo nửa đêm của chúng tôi. Trong đợt cao điểm bán hàng tiếp theo, mức sử dụng bộ nhớ duy trì ổn định ở 185 MB, ngay cả khi lưu lượng truy cập vọt qua mốc 28.000 request mỗi giây.
Hãy ghi nhớ bốn quy tắc sau trước khi đưa channel lên môi trường production:
- Luôn thiết lập dung lượng giới hạn (bounded capacity): Tuyệt đối không sử dụng
Channel.CreateUnbounded<T>phía sau các public endpoint. Bộ đệm không giới hạn chỉ đánh đổi thông lượng ngắn hạn lấy việc sập tiến trình trong tương lai. - Chọn FullMode phù hợp: Sử dụng
BoundedChannelFullMode.Waitkhi không được phép làm mất dữ liệu. Đối với telemetry không quá quan trọng, hãy chuyển sangDropOldesthoặcDropNewestđể ưu tiên dữ liệu mới hơn dữ liệu cũ. - Giải phóng hàng đợi khi graceful shutdown: Gọi
writer.Complete()bên trong các stopping hook của vòng đời ứng dụng. Điều này cho phépReadAllAsync()xử lý hết các phần tử còn lại trong bộ đệm trước khi runtime dừng hẳn. - Tận dụng các cờ single-actor: Bật
SingleReaderhoặcSingleWriterbất cứ khi nào chỉ có duy nhất một worker xử lý phía đó của channel. Runtime sẽ loại bỏ các lock nội bộ, giúp tăng thông lượng xử lý rõ rệt.
