.NET Backend: Zero → Senior

Giao diện

6.8 - Channels và Producer-Consumer ​

Tóm tắt: System.Threading.Channels là hàng đợi trong bộ nhớ, không khoá, có hỗ trợ async — thứ bạn cần khi một phần code sinh việc nhanh hơn phần kia xử lý được. Quyết định quan trọng nhất là bounded hay unbounded: unbounded chịu mọi tải cho tới lúc hết RAM, bounded thì đẩy áp lực ngược về phía producer để hệ thống chậm lại thay vì chết. Hai cái bẫy hay gặp nhất: dùng IHostedService.StartAsync làm consumer (nó chặn khởi động của cả ứng dụng), và quên rằng channel nằm trong RAM nên restart là mất sạch.

Mục tiêu bài học ​

Sau bài này bạn có thể:

  • Giải thích channel giải quyết vấn đề gì mà Task.Run không giải quyết được.
  • Chọn đúng giữa bounded và unbounded, và đúng FullMode.
  • Viết consumer bằng BackgroundService với scope DI đúng cách.
  • Đóng channel gọn gàng khi ứng dụng shutdown.
  • Biết khi nào phải bỏ channel và dùng message broker thật.

Cần biết trước: bài 6.4 (CancellationToken) và bài 6.6 (vì sao fire-and-forget mất việc). Hosted service là một service chạy nền suốt vòng đời ứng dụng ASP.NET Core, được host khởi động lúc ứng dụng bật và dừng lúc ứng dụng tắt; BackgroundService là lớp cơ sở có sẵn để viết loại service này. Bài cũng dùng IServiceScopeFactory để tạo scope DI riêng; lifetime Singleton/Scoped được dạy kỹ ở bài 7.4.

Nội dung bài học ​

6.8.1 — Vấn đề: producer nhanh hơn consumer ​

Endpoint nhận webhook mất 5ms; xử lý mỗi webhook mất 400ms. Xử lý ngay trong request thì client chờ 400ms và bạn chịu tải bằng tốc độ xử lý. Fire-and-forget thì mất việc khi shutdown (bài 6.6).

Channel tách hai bên ra:

Producer trả lời ngay; consumer xử lý theo nhịp của nó; và khi hàng đợi đầy, producer bị làm chậm lại thay vì hệ thống sụp.

6.8.2 — Bounded và unbounded ​

csharp
// UNBOUNDED — không giới hạn. Producer không bao giờ bị chặn.
var unbounded = Channel.CreateUnbounded<Job>();

// BOUNDED — tối đa 1000. Đầy thì WriteAsync chờ.
var bounded = Channel.CreateBounded<Job>(new BoundedChannelOptions(1000)
{
    FullMode = BoundedChannelFullMode.Wait
});

Unbounded có một vấn đề duy nhất nhưng chí mạng: nếu producer bền bỉ nhanh hơn consumer, hàng đợi lớn dần cho tới khi tiến trình hết bộ nhớ. Và nó chết vào lúc tải cao nhất, tức đúng lúc tệ nhất.

Với bounded, bốn lựa chọn khi đầy:

FullModeHành viDùng khi
WaitWriteAsync chờ đến khi có chỗMặc định — không được mất việc
DropWriteBỏ item mớiTelemetry, metric — mới hay cũ đều được
DropOldestBỏ item cũ nhấtDữ liệu realtime, chỉ cần cái mới nhất
DropNewestBỏ item mới nhất trong hàng đợiHiếm dùng

Wait là thứ tạo ra backpressure: producer chậm lại, request tồn đọng, load balancer thấy và ngừng gửi thêm. Hệ thống suy giảm dần thay vì sập đột ngột.

Có hai tuỳ chọn tối ưu đáng bật khi bạn chắc chắn về mô hình:

csharp
new BoundedChannelOptions(1000)
{
    FullMode = BoundedChannelFullMode.Wait,
    SingleReader = true,    // chỉ một consumer
    SingleWriter = false,   // nhiều request cùng ghi
}

SingleReader = true cho phép runtime bỏ bớt đồng bộ hoá nội bộ. Khai báo sai thì hỏng âm thầm — chỉ bật khi đúng.

6.8.3 — Producer và consumer ​

csharp
public interface IJobQueue
{
    ValueTask EnqueueAsync(Job job, CancellationToken ct = default);
    IAsyncEnumerable<Job> ReadAllAsync(CancellationToken ct);
    void Complete();
}

public sealed class JobQueue : IJobQueue
{
    private readonly Channel<Job> _channel = Channel.CreateBounded<Job>(
        new BoundedChannelOptions(1000)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = true,
        });

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

    public IAsyncEnumerable<Job> ReadAllAsync(CancellationToken ct)
        => _channel.Reader.ReadAllAsync(ct);

    public void Complete() => _channel.Writer.Complete();
}

WriteAsync trả ValueTask chứ không phải Task — vì với channel chưa đầy nó hoàn thành đồng bộ và không cấp phát gì. Đây đúng là trường hợp ValueTask sinh ra để phục vụ (bài 6.2).

Có cả bản không chờ, dùng khi bạn không muốn producer bị chặn:

csharp
if (!_channel.Writer.TryWrite(job))
    _logger.LogWarning("Hàng đợi đầy, bỏ qua job {Id}", job.Id);

6.8.4 — Consumer phải là BackgroundService, không phải IHostedService.StartAsync ​

Đây là lỗi hay gặp nhất trong bài này:

csharp
// SAI — StartAsync không bao giờ trả về, ứng dụng không bao giờ khởi động xong
public async Task StartAsync(CancellationToken ct)
{
    await foreach (var job in _queue.ReadAllAsync(ct))
        await ProcessAsync(job, ct);
}

Host gọi StartAsync của từng hosted service tuần tự và chờ từng cái xong trước khi mở cổng HTTP. Vòng lặp trên chạy mãi, nên ứng dụng treo ở bước khởi động — không có log lỗi, chỉ là không bao giờ sẵn sàng.

csharp
public sealed class JobProcessor : BackgroundService
{
    private readonly IJobQueue _queue;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<JobProcessor> _logger;

    public JobProcessor(IJobQueue queue, IServiceScopeFactory scopeFactory, ILogger<JobProcessor> logger)
        => (_queue, _scopeFactory, _logger) = (queue, scopeFactory, logger);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var job in _queue.ReadAllAsync(stoppingToken))
        {
            try
            {
                await using var scope = _scopeFactory.CreateAsyncScope();
                var handler = scope.ServiceProvider.GetRequiredService<IJobHandler>();
                await handler.HandleAsync(job, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Job {Id} thất bại", job.Id);
                // KHÔNG throw — một job hỏng không được làm chết consumer
            }
        }
    }
}

Ba điểm bắt buộc:

  1. ExecuteAsync, không phải StartAsync. BackgroundService.StartAsync khởi động ExecuteAsync rồi trả về ngay, không chờ nó xong. Từ .NET 10, toàn bộ ExecuteAsync chạy trên thread nền nên không chặn khởi động; trên .NET 8 và 9, phần code trước await đầu tiên vẫn chạy trên luồng khởi động và chặn các service khác (breaking change của .NET 10).
  2. try/catch bao quanh từng job. Exception thoát ra khỏi ExecuteAsync thì consumer dừng hẳn. Với cấu hình mặc định (từ .NET 6), host còn dừng cả ứng dụng: một job hỏng kéo sập toàn bộ API (bài tập 3 đo điều này).
  3. Tạo scope riêng cho mỗi job. BackgroundService là singleton; lấy thẳng DbContext (scoped) vào constructor là lỗi vòng đời. Xem Module 7 — Dependency Injection.

Đăng ký:

csharp
builder.Services.AddSingleton<IJobQueue, JobQueue>();
builder.Services.AddHostedService<JobProcessor>();

6.8.5 — Nhiều consumer song song ​

csharp
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    var workers = Enumerable.Range(0, 4).Select(_ => ConsumeAsync(stoppingToken)).ToList();
    await Task.WhenAll(workers);
}

Nhớ đặt SingleReader = false khi làm vậy. Channel tự chia việc — mỗi item chỉ đến một consumer.

Đây là cách tăng thông lượng mà vẫn giữ trần tài nguyên: bốn consumer nghĩa là nhiều nhất bốn job chạy cùng lúc, bất kể hàng đợi dài bao nhiêu.

6.8.6 — Shutdown gọn gàng ​

Khi ứng dụng tắt, host gọi StopAsync của mọi hosted service. BackgroundService.StopAsync mặc định huỷ stoppingToken ngay lập tức rồi chờ ExecuteAsync kết thúc. Consumer ở mục 6.8.4 đọc bằng ReadAllAsync(stoppingToken), nên vòng lặp dừng ngay và mọi job còn trong hàng đợi bị bỏ.

Cách viết hay gặp trên mạng không sửa được việc này:

csharp
// KHÔNG drain được: base.StopAsync huỷ stoppingToken ngay, ReadAllAsync(stoppingToken) dừng luôn
public override async Task StopAsync(CancellationToken ct)
{
    _queue.Complete();
    await base.StopAsync(ct);
}

Muốn xử lý nốt việc còn tồn ("drain"), consumer phải đọc không theo stoppingToken, và StopAsync phải chờ vòng lặp xong trước khi gọi base.StopAsync:

csharp
public sealed class DrainingConsumer(JobQueue queue, ILogger<DrainingConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Chỉ dừng khi channel đã Complete() và rỗng, không dừng theo stoppingToken
        await foreach (var job in queue.ReadAllAsync(CancellationToken.None))
        {
            try
            {
                await ProcessAsync(job, CancellationToken.None);   // job đang chạy được làm cho xong
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Job {Id} thất bại", job.Id);
            }
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        queue.Complete();                                   // 1. không nhận việc mới
        if (ExecuteTask is { } running)
            await running.WaitAsync(cancellationToken);     // 2. chờ xử lý hết, nhưng không quá hạn của host
        await base.StopAsync(cancellationToken);            // 3. lúc này vòng lặp đã xong
    }
}

ExecuteTask là Task của ExecuteAsync mà BackgroundService giữ sẵn. cancellationToken của StopAsync bị huỷ khi hết thời hạn tắt của host, nên bước 2 không chờ quá mức đó. Hết hạn thì WaitAsync ném OperationCanceledException, và host ghi log rằng việc tắt bị cắt ngang.

Kết quả đo trên .NET 10: xếp 10 job (mỗi job 200 ms) rồi gọi StopApplication() ngay. Bản Complete() + base.StopAsync xử lý 0 job; bản DrainingConsumer xử lý đủ 10 job trong khoảng 2,1 giây rồi mới tắt.

Thời hạn tắt. HostOptions.ShutdownTimeout mặc định là 30 giây (từ .NET 6; trước đó là 5 giây). Job dài hơn thì tăng lên:

csharp
builder.Services.Configure<HostOptions>(o =>
    o.ShutdownTimeout = TimeSpan.FromSeconds(60));

Khi chạy trong Kubernetes, pod chỉ có terminationGracePeriodSeconds (mặc định 30 giây) từ lúc nhận tín hiệu tắt tới lúc bị giết hẳn. Đặt ShutdownTimeout dài hơn con số đó thì vô ích: tiến trình bị giết trước khi kịp drain.

6.8.7 — Khi nào channel là sai lựa chọn ​

Channel sống trong bộ nhớ của một tiến trình. Hệ quả:

  • Restart hoặc crash là mất sạch hàng đợi. Không persistence, không replay.
  • Không chia sẻ giữa các instance. Ba pod Kubernetes là ba hàng đợi riêng biệt.
  • Không có retry, không có dead-letter — bạn phải tự viết.
DùngKhi
ChannelViệc trong tiến trình, mất cũng chấp nhận được: log, metric, cache warm-up, gửi email không quan trọng
Hangfire / QuartzCần persistence và retry, vẫn một ứng dụng, có sẵn database
RabbitMQ / Kafka / Azure Service BusNhiều service, nhiều instance, không được mất message

Câu hỏi quyết định rất đơn giản: "mất một item khi server restart thì có sao không?" Nếu có sao, đừng dùng channel. Xem Module 14 — Caching và Background Jobs và Module 17 — Distributed Systems.

6.8.8 — Rà lại code của bạn ​

Danh sách rà soát Channels

  • Channel dùng CreateBounded, không phải CreateUnbounded.
  • FullMode được chọn có chủ đích, không để mặc định vì tiện.
  • Consumer kế thừa BackgroundService và ghi đè ExecuteAsync.
  • Mỗi job được xử lý trong try/catch riêng, exception không thoát ra ExecuteAsync.
  • Mỗi job tạo scope DI riêng qua IServiceScopeFactory.
  • StopAsync gọi Writer.Complete() trước khi chờ.
  • ShutdownTimeout đủ dài cho job dài nhất.
  • SingleReader/SingleWriter khớp với số consumer/producer thực tế.
  • Đã trả lời được: mất một item khi restart thì có sao không?

Bài tập áp dụng ​

Bài 1 — Thấy áp lực ngược ​

Tạo bounded channel sức chứa 10 với FullMode.Wait, producer ghi 1000 mục không nghỉ, consumer chờ 50 mili-giây mỗi mục. Đo thời gian mỗi lần WriteAsync.

Tiêu chí hoàn thành: bạn thấy rõ producer bị kéo về đúng nhịp của consumer, và nêu được điều gì xảy ra nếu dùng unbounded thay vào đó.

Gợi ý và lời giải — Bài 1

Gợi ý. WriteAsync trên một channel đầy sẽ chờ cho tới khi có chỗ trống. Thời gian chờ đó chính là áp lực ngược, và bạn đo được nó.

Lời giải.

csharp
using System.Diagnostics;
using System.Threading.Channels;

var bounded = args.Length == 0 || args[0] != "unbounded";
var ch = bounded
    ? Channel.CreateBounded<int>(new BoundedChannelOptions(10) { FullMode = BoundedChannelFullMode.Wait })
    : Channel.CreateUnbounded<int>();

int consumed = 0;
// Consumer — chậm
var consumer = Task.Run(async () =>
{
    await foreach (var x in ch.Reader.ReadAllAsync())
    {
        await Task.Delay(50);
        Interlocked.Increment(ref consumed);
    }
});

// Producer — nhanh, đo thời gian mỗi lần ghi
var total = Stopwatch.StartNew();
var sw = new Stopwatch();
for (int i = 0; i < 1000; i++)
{
    sw.Restart();
    await ch.Writer.WriteAsync(i);
    sw.Stop();
    if (i < 12 || i % 250 == 0 || i == 999)
        Console.WriteLine($"ghi {i,4}: {sw.Elapsed.TotalMilliseconds,6:F1} ms");
}
Console.WriteLine($"producer ghi xong 1000 mục sau {total.ElapsedMilliseconds} ms, consumer mới xử lý {consumed} mục");
ch.Writer.Complete();
await consumer;
Console.WriteLine($"consumer xong sau {total.ElapsedMilliseconds} ms");

Chạy dotnet run -c Release cho bản bounded và dotnet run -c Release -- unbounded cho bản unbounded.

Kết quả chạy trên .NET 10 (Windows):

text
ghi    0:    1,1 ms      <- channel còn trống
ghi    1:    0,0 ms
...
ghi   10:    0,0 ms      <- channel vừa đầy (10 mục chờ + 1 mục consumer đang xử lý)
ghi   11:   47,6 ms      <- BẮT ĐẦU CHỜ
ghi  250:   62,2 ms
ghi  500:   62,3 ms
ghi  750:   63,9 ms
producer ghi xong 1000 mục sau 61022 ms, consumer mới xử lý 989 mục

Mười một lần ghi đầu gần như tức thì: 10 chỗ trong channel cộng 1 mục consumer đã lấy ra. Từ đó, mỗi lần ghi phải chờ bằng đúng thời gian consumer xử lý một mục: producer đã bị kéo về nhịp của consumer. Con số là khoảng 62 ms chứ không phải 50 ms vì trên Windows, Task.Delay làm tròn lên theo bước hẹn giờ của hệ điều hành (mặc định khoảng 15,6 ms, và 4 × 15,6 ≈ 62).

Đây chính là áp lực ngược. Hệ thống tự điều chỉnh: bên nhanh phải chờ bên chậm, không cần ai cấu hình gì thêm.

Nếu dùng unbounded thay vào đó:

csharp
var ch = Channel.CreateUnbounded<int>();
text
ghi    0:    1,0 ms
ghi  999:    0,0 ms      <- KHÔNG BAO GIỜ chờ
producer ghi xong 1000 mục sau 10 ms, consumer mới xử lý 0 mục

Producer ghi xong 1000 mục trong 10 mili-giây, consumer chưa kịp xử lý mục nào. Cả 1000 mục nằm trong bộ nhớ, và consumer cần thêm 60 giây để xử lý hết.

Với 1000 mục thì không sao. Với một producer chạy liên tục nhanh hơn consumer, hàng đợi tăng vô hạn cho tới khi tiến trình hết bộ nhớ. Và triệu chứng rất khó chẩn đoán: bộ nhớ tăng đều, không có rò rỉ nào theo nghĩa thông thường, mọi thứ "hoạt động bình thường" cho tới lúc sập.

Bốn chế độ của bounded channel:

FullModeHành vi khi đầyDùng khi
WaitProducer chờMặc định — không được mất dữ liệu
DropOldestBỏ mục cũ nhấtDữ liệu đo lường, chỉ cần giá trị mới nhất
DropNewestBỏ mục mới nhất đang nằm trong hàng để nhường chỗ cho mục đang ghiHiếm dùng
DropWriteBỏ lặng lẽ chính mục đang ghi, WriteAsync trả về ngayLog không quan trọng

Chọn sức chứa bao nhiêu. Nó là bộ đệm hấp thụ các đợt tăng đột biến ngắn. Quá nhỏ thì producer chờ liên tục ngay cả với biến động bình thường; quá lớn thì độ trễ tăng và bộ nhớ tốn. Điểm khởi đầu hợp lý: đủ chứa lượng công việc của vài giây ở nhịp bình thường.

Vì sao áp lực ngược quan trọng hơn vẻ ngoài. Nó là cách hệ thống nói cho bạn biết rằng consumer không theo kịp. Với unbounded, tín hiệu đó bị che giấu cho tới khi quá muộn. Nhiều sự cố hết bộ nhớ trong hệ thống hàng đợi có nguyên nhân gốc chỉ là một dòng CreateUnbounded.

Bài 2 — Tái hiện treo khởi động ​

Viết consumer bằng IHostedService.StartAsync với vòng lặp await foreach, chạy ứng dụng và xác nhận nó không bao giờ nhận request. Chuyển sang BackgroundService và xác nhận nó khởi động bình thường.

Tiêu chí hoàn thành: bạn giải thích được khác biệt giữa hai cách, và nêu được khi nào IHostedService vẫn là lựa chọn đúng.

Gợi ý và lời giải — Bài 2

Gợi ý. Host gọi StartAsync của từng dịch vụ theo thứ tự và chờ từng cái xong trước khi chuyển sang cái tiếp theo. Vậy chuyện gì xảy ra nếu một StartAsync không bao giờ trả về?

Lời giải — bản treo:

csharp
public class EmailConsumer : IHostedService
{
    private readonly EmailQueue _q;

    public async Task StartAsync(CancellationToken ct)
    {
        await foreach (var job in _q.ReadAllAsync(ct))     // KHÔNG BAO GIỜ kết thúc
            await ProcessAsync(job, ct);
    }

    public Task StopAsync(CancellationToken ct) => Task.CompletedTask;
}

Chạy ứng dụng trên .NET 10: cửa sổ console không in ra dòng log nào của Microsoft.Hosting.Lifetime, không có Now listening on: ..., và curl vào ứng dụng thì hết thời gian chờ mà không có phản hồi.

Máy chủ web không bao giờ khởi động, vì host khởi động các hosted service theo thứ tự đăng ký, chờ StartAsync của từng cái trả về, rồi mới tới máy chủ web.

Bản đúng:

csharp
public class EmailConsumer(EmailQueue q) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var job in q.ReadAllAsync(stoppingToken))
            await ProcessAsync(job, stoppingToken);
    }
}
text
info: Microsoft.Hosting.Lifetime[0]
      Now listening on: http://localhost:5000

Khác biệt. BackgroundService cài đặt sẵn IHostedService, và trong StartAsync nó khởi động ExecuteAsync rồi trả về ngay mà không chờ:

csharp
// Xấp xỉ cài đặt bên trong BackgroundService trên .NET 10
public virtual Task StartAsync(CancellationToken cancellationToken)
{
    _stoppingCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    _executeTask = Task.Run(() => ExecuteAsync(_stoppingCts.Token), _stoppingCts.Token);  // không await
    return Task.CompletedTask;
}

Task.Run ở đây là thay đổi của .NET 10. Đã thử: đặt Thread.Sleep(3000) ngay đầu ExecuteAsync, trước mọi await; trên .NET 10 dòng Now listening on vẫn hiện ngay, còn trên .NET 8/9 ứng dụng sẽ chờ đủ 3 giây mới khởi động xong.

Nó vẫn giữ tham chiếu tới task, nên khi ứng dụng tắt, StopAsync huỷ token và chờ task kết thúc gọn gàng. Đây chính là fire-and-forget làm đúng cách — khác với _ = MethodAsync() ở bài 6.6, nơi không ai giữ task.

Khi nào IHostedService vẫn đúng. Khi bạn cần làm một việc có điểm kết thúc trong lúc khởi động hoặc tắt:

csharp
public class MigrationRunner(IServiceScopeFactory sf) : IHostedService
{
    public async Task StartAsync(CancellationToken ct)
    {
        using var scope = sf.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<CrmDbContext>();
        await db.Database.MigrateAsync(ct);      // CÓ kết thúc — và ta MUỐN chờ
    }

    public Task StopAsync(CancellationToken ct) => Task.CompletedTask;
}

Ở đây việc chặn khởi động là có chủ ý: ứng dụng không nên nhận request trước khi database đã sẵn sàng.

Bảng chọn:

Nhu cầuDùng
Vòng lặp chạy suốt vòng đời ứng dụngBackgroundService
Việc một lần lúc khởi động, phải xong trước khi nhận requestIHostedService
Việc một lần lúc tắtIHostedService, phần StopAsync
Việc chạy theo lịchBackgroundService kèm PeriodicTimer

Một chi tiết về PeriodicTimer. Với tác vụ định kỳ, nó tốt hơn Task.Delay trong vòng lặp vì nó không bị trôi thời gian:

csharp
using var timer = new PeriodicTimer(TimeSpan.FromMinutes(5));
while (await timer.WaitForNextTickAsync(stoppingToken))
    await ProcessAsync(stoppingToken);

Task.Delay(5 phút) sau mỗi lần xử lý nghĩa là chu kỳ thật bằng 5 phút cộng thời gian xử lý, và sai lệch tích luỹ dần.

Bài 3 — Giết consumer bằng một ngoại lệ ​

Bỏ try/catch quanh thân vòng lặp, đẩy vào một công việc gây ngoại lệ, rồi đẩy tiếp 10 công việc hợp lệ. Quan sát ứng dụng với cấu hình mặc định, rồi với BackgroundServiceExceptionBehavior.Ignore.

Tiêu chí hoàn thành: bạn mô tả được hai hành vi (cả ứng dụng dừng, và consumer chết im lặng trong khi ứng dụng vẫn chạy), và nêu được cách phát hiện trường hợp thứ hai.

Gợi ý và lời giải — Bài 3

Gợi ý. Khi ExecuteAsync ném ra ngoài, BackgroundService không tự khởi động lại nó. Phần còn lại phụ thuộc HostOptions.BackgroundServiceExceptionBehavior: mặc định là StopHost.

Lời giải — bản dễ vỡ:

csharp
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    await foreach (var job in _q.ReadAllAsync(stoppingToken))
        await ProcessAsync(job, stoppingToken);      // ném -> thoát vòng lặp -> consumer chết
}

Thử:

csharp
await q.EnqueueAsync(new Job("FAIL"), ct);                   // gây ngoại lệ
for (int i = 1; i <= 10; i++)
    await q.EnqueueAsync(new Job($"valid-{i}"), ct);

Kết quả với cấu hình mặc định, chạy trên .NET 10: job FAIL ném ngoại lệ, và cả ứng dụng dừng chỉ vài mili-giây sau đó:

text
fail: Microsoft.Extensions.Hosting.Internal.Host[9]
      BackgroundService failed
      System.InvalidOperationException: Job FAIL hỏng
crit: Microsoft.Extensions.Hosting.Internal.Host[10]
      The HostOptions.BackgroundServiceExceptionBehavior is configured to StopHost.
      A BackgroundService has thrown an unhandled exception, and the IHost instance is stopping. ...
info: Microsoft.Hosting.Lifetime[0]
      Application is shutting down...

Một job hỏng kéo sập cả API. Đó là hành vi mặc định từ .NET 6, và nó hợp lý: lỗi lộ ra ngay thay vì âm thầm.

Kết quả với Ignore. Nhiều dự án đổi hành vi này để "ứng dụng không bị tắt":

csharp
builder.Services.Configure<HostOptions>(o =>
    o.BackgroundServiceExceptionBehavior = BackgroundServiceExceptionBehavior.Ignore);

Chạy lại: log chỉ có một dòng fail: ... BackgroundService failed, rồi ứng dụng vẫn phục vụ HTTP bình thường (đã thử curl vẫn nhận phản hồi), health check vẫn xanh. Mười job hợp lệ nằm mãi trong hàng đợi, không ai xử lý.

Đây là loại sự cố tệ nhất: một phần hệ thống chết trong khi mọi chỉ số bề mặt đều bình thường.

Bản đúng:

csharp
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    await foreach (var job in _q.ReadAllAsync(stoppingToken))
    {
        try
        {
            await ProcessAsync(job, stoppingToken);
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            break;                                   // tắt ứng dụng — thoát gọn gàng
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Job {JobId} thất bại", job.Id);
            // KHÔNG ném lại — một job hỏng không được giết cả consumer
        }
    }
}

Hai khối catch có vai trò khác hẳn nhau:

  • Khối đầu bắt tín hiệu tắt và thoát gọn gàng.
  • Khối sau bắt lỗi của một công việc và đi tiếp.

Thiếu khối đầu thì log đầy ngoại lệ mỗi lần triển khai. Thiếu khối sau thì một công việc hỏng giết cả consumer.

Dù cấu hình là gì, kết luận giống nhau: consumer phải tự bắt lỗi của từng job. Với StopHost thì thiếu try/catch làm sập cả ứng dụng; với Ignore thì làm consumer chết im lặng.

Ba cách phát hiện consumer đã chết:

  1. Health check dựa trên nhịp tim. Consumer cập nhật một mốc thời gian sau mỗi vòng; health check (bài 8.9) báo không khoẻ nếu mốc đó quá cũ:

    csharp
    builder.Services.AddHealthChecks()
    .AddCheck<EmailConsumerHealthCheck>("email-consumer", tags: ["ready"]);

  2. Cảnh báo theo độ dài hàng đợi. Hàng đợi chỉ tăng mà không giảm là dấu hiệu rõ ràng consumer không còn làm việc.

  3. Đếm số công việc đã xử lý. Xuất một chỉ số và cảnh báo khi nó ngừng tăng trong khi hàng đợi vẫn có việc.

Nâng cấp đáng làm: xử lý công việc lỗi. Thay vì chỉ ghi log rồi bỏ, hãy có chiến lược thử lại. Nhưng có một cái bẫy chết người:

csharp
// SAI — consumer tự await WriteAsync vào channel bounded mà chính nó là reader duy nhất
catch (Exception ex)
{
    if (job.RetryCount < 3)
        await _q.EnqueueAsync(job with { RetryCount = job.RetryCount + 1 }, stoppingToken);
}

Khi hàng đầy, EnqueueAsync chờ có chỗ trống. Nhưng người duy nhất có thể tạo chỗ trống là chính consumer, và nó đang đứng chờ ở dòng đó: deadlock vĩnh viễn. Đã thử với channel sức chứa 5, mọi job đều lỗi và được ghi lại: sau lượt xử lý đầu tiên, consumer kẹt ở WriteAsync mãi, hàng đợi đứng yên ở 5 mục.

Quy tắc: không bao giờ await WriteAsync vào channel bounded mà chính mình là reader. Ba cách đúng:

csharp
catch (Exception ex)
{
    _logger.LogError(ex, "Job {JobId} thất bại lần {Attempt}", job.Id, job.RetryCount);

    var retry = job with { RetryCount = job.RetryCount + 1 };
    if (job.RetryCount >= 3 || !_q.TryEnqueue(retry))   // TryWrite: đầy thì trả false ngay, không chờ
        await _deadLetter.SaveAsync(job, ex, stoppingToken);   // hết lượt hoặc hàng đầy: vào hàng thư chết
}

TryEnqueue ở đây bọc _channel.Writer.TryWrite(job). Hai cách khác: đưa job lỗi vào một channel retry riêng (không giới hạn, có độ trễ trước khi thử lại), hoặc thử lại ngay trong vòng xử lý bằng thư viện như Polly.

Mẫu hàng đợi thư chết này được Module 17 trình bày đầy đủ, kèm cả chiến lược chờ tăng dần giữa các lần thử lại.

Tự kiểm tra ​

Channel giải quyết vấn đề gì?

Nó tách producer khỏi consumer khi một bên sinh việc nhanh hơn bên kia xử lý được. Producer trả lời ngay, consumer xử lý theo nhịp của nó, và khi hàng đợi đầy thì producer bị làm chậm lại thay vì hệ thống sụp. Đây là hàng đợi trong bộ nhớ, không cần lock và hỗ trợ async.

Nên chọn bounded hay unbounded?

Gần như luôn là bounded. Unbounded không chặn producer nên nếu producer bền bỉ nhanh hơn consumer thì hàng đợi lớn dần cho tới khi tiến trình hết bộ nhớ, và nó chết đúng vào lúc tải cao nhất. Bounded với FullMode Wait tạo ra backpressure nên hệ thống suy giảm dần thay vì sập đột ngột.

Vì sao không được viết consumer trong IHostedService.StartAsync?

Vì host gọi StartAsync của từng hosted service tuần tự và chờ từng cái xong trước khi mở cổng HTTP. Một vòng lặp await foreach chạy mãi sẽ làm ứng dụng treo ở bước khởi động, không có log lỗi, chỉ là không bao giờ sẵn sàng. Đúng là kế thừa BackgroundService và ghi đè ExecuteAsync.

Vì sao mỗi job cần scope DI riêng?

Vì BackgroundService là singleton, còn DbContext và phần lớn service nghiệp vụ là scoped. Lấy thẳng chúng vào constructor là lỗi vòng đời và sẽ dùng chung một instance cho mọi job. Cách đúng là tiêm IServiceScopeFactory rồi tạo scope mới cho từng job.

Vì sao phải try/catch quanh từng job?

Vì exception thoát ra khỏi ExecuteAsync làm consumer dừng hẳn. Với cấu hình mặc định StopHost (từ .NET 6), cả ứng dụng dừng theo: một job hỏng kéo sập toàn bộ API. Nếu dự án đặt BackgroundServiceExceptionBehavior.Ignore thì tệ theo kiểu khác: hàng đợi vẫn nhận việc nhưng không còn ai xử lý, ứng dụng vẫn chạy bình thường nên không ai phát hiện ra.

Khi nào không nên dùng Channel?

Khi không được phép mất việc. Channel sống trong bộ nhớ của một tiến trình nên restart hay crash là mất sạch, không chia sẻ được giữa các instance, và không có retry hay dead-letter. Câu hỏi quyết định là mất một item khi server restart thì có sao không; nếu có sao thì dùng Hangfire, RabbitMQ hoặc Kafka.

Kết luận ​

Ba điều đáng nhớ nhất:

  1. Bounded channel là mặc định. Backpressure giúp hệ thống chậm lại thay vì chết vì hết bộ nhớ.
  2. Consumer là BackgroundService.ExecuteAsync, có try/catch quanh từng job và scope DI riêng cho từng job.
  3. Channel không có persistence. Mất việc khi restart là chấp nhận được thì dùng; không thì cần broker thật.

Tham khảo ​

Điều hướng ​

Phát hành theo giấy phép MIT.

Nội dung gốc của TienNHM