Cách Xây Dựng Hàng Đợi Công Việc Nền Trong Bộ Nhớ Từ Đầu Trong ASP.NET Core

Trong các ứng dụng web, việc cung cấp API nhanh và phản hồi kịp thời là rất quan trọng đối với trải nghiệm người dùng. Tuy nhiên, một số thao tác như gửi email, tạo báo cáo hoặc thêm người dùng hàng loạt có thể tốn nhiều thời gian. Thực hiện các tác vụ này trong thời gian xử lý yêu cầu có thể chặn các luồng, dẫn đến thời gian phản hồi chậm.

Để giải quyết vấn đề này một cách hiệu quả, chúng ta nên chạy các tác vụ dài hạn trong một quy trình nền. Thay vì chờ yêu cầu hoàn thành, chúng ta có thể xếp hàng tác vụ, cung cấp phản hồi ngay lập tức cho người dùng và thực hiện công việc trong một quy trình nền riêng biệt.

Mặc dù có các thư viện như Hangfire hoặc message broker như RabbitMQ cho mục đích này, bạn không cần bất kỳ phụ thuộc bên ngoài nào. Trong bài viết này, chúng ta sẽ xây dựng một hàng đợi công việc nền trong bộ nhớ từ đầu trong ASP.NET Core chỉ sử dụng các tính năng tích hợp sẵn của .NET.

Mục lục

Các thành phần chính của hệ thống công việc nền

Trước khi triển khai mã, điều quan trọng là trước tiên phải hiểu các phần cơ bản tạo nên hệ thống công việc nền và cách chúng hoạt động để xử lý các tác vụ đã xếp hàng.

IHostedService và BackgroundService là gì?

IHostedService là một giao diện được sử dụng trong ASP.NET Core để triển khai các tác vụ nền chạy dài được quản lý bởi vòng đời ứng dụng. Khi ứng dụng của bạn khởi động, nó sẽ khởi động tất cả các ứng dụng IHostedService đã đăng ký. Khi ứng dụng dừng, nó sẽ hướng dẫn chúng dừng lại.

BackgroundService là một lớp cơ sở trừu tượng triển khai IHostedService. Nó cung cấp cho bạn một cách hiệu quả hơn để tạo một dịch vụ định thời bằng cách cung cấp một phương thức có thể ghi đè duy nhất ExecuteAsync(CancellationToken stoppingToken). Chúng ta sẽ sử dụng lớp này làm cơ sở cho bộ xử lý công việc của chúng ta.

Cấu trúc Producer/Consumer

Producer: Đây là phần của ứng dụng tạo công việc và thêm chúng vào hàng đợi. Trong ví dụ của chúng ta, producer sẽ là một API Controller.
Consumer: Đây là một dịch vụ nền liên tục giám sát hàng đợi, lấy công việc từ hàng đợi và thực thi chúng. Ứng dụng BackgroundService của chúng ta sẽ là consumer.
Queue: Đây là cấu trúc dữ liệu giữa producer và consumer chứa các công việc đang chờ được xử lý.

Tại sao lại là System.Threading.Channels?

Được giới thiệu trong .NET Core 3.0, System.Threading.Channels cung cấp một cấu trúc dữ liệu đồng bộ hóa được thiết kế cho các kịch bản producer-consumer không đồng bộ.

Thread Safety: Xử lý tất cả các thao tác khóa và đồng bộ hóa phức tạp bên trong, làm cho nó có thể sử dụng từ nhiều luồng (ví dụ: khi nhiều yêu cầu API đang thêm công việc đồng thời).
Async Native: Được thiết kế để sử dụng với async/await, ngăn chặn việc chặn luồng. Bạn có thể không đồng bộ chờ một mục có sẵn trong hàng đợi.
Performance: Được tối ưu hóa cho tốc độ và phân bổ bộ nhớ thấp. Nhìn chung vượt trội hơn các cấu trúc cũ như ConcurrentQueue cho các quy trình không đồng bộ vì nó loại bỏ nhu cầu polling.

Triển khai hàng đợi công việc nền

Bây giờ chúng ta đã hiểu kiến trúc, chúng ta có thể triển khai hàng đợi công việc nền.

Tạo dự án

Đầu tiên, tạo một dự án ASP.NET Core Web API mới bằng .NET CLI.

dotnet new sln -n BackgroundJobQueue dotnet new webapi -n BackgroundJobQueue.Api dotnet sln BackgroundJobQueue.sln add BackgroundJobQueue.Api/BackgroundJobQueue.Api.csproj

Định nghĩa giao diện hàng đợi công việc

Để dependency injection và khả năng kiểm thử, chúng ta sẽ bắt đầu bằng cách định nghĩa một giao diện cho hàng đợi của chúng ta. Giao diện này sẽ định nghĩa các phương thức để thêm công việc (EnqueueAsync) và xóa công việc (DequeueAsync).

Tạo một tệp mới có tên IBackgroundTaskQueue.cs.

namespace BackgroundJobQueue; public interface IBackgroundTaskQueue { // Thêm một mục công việc vào hàng đợi ValueTask EnqueueAsync(Func<CancellationToken, ValueTask> workItem); // Xóa và trả về một mục công việc từ hàng đợi ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken); }

Triển khai dịch vụ hàng đợi với Channels

Bây giờ, hãy triển khai giao diện IBackgroundTaskQueue bằng cách sử dụng lớp System.Threading.Channels. Lớp này sẽ quản lý kênh trong bộ nhớ.

Tạo một tệp mới có tên BackgroundTaskQueue.cs.

using System.Threading.Channels; namespace BackgroundJobQueue; public class BackgroundTaskQueue : IBackgroundTaskQueue { private readonly Channel<Func<CancellationToken, ValueTask>> _queue; public BackgroundTaskQueue(int capacity) { // BoundedChannelOptions xác định hành vi của kênh. var options = new BoundedChannelOptions(capacity) { // FullMode.Wait yêu cầu writer chờ không đồng bộ nếu hàng đợi đầy. FullMode = BoundedChannelFullMode.Wait }; _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options); } public async ValueTask EnqueueAsync(Func<CancellationToken, ValueTask> workItem) { if (workItem is null) { throw new ArgumentNullException(nameof(workItem)); } // Ghi một mục vào kênh. Nếu kênh đầy, await _queue.Writer.WriteAsync(workItem); } public async ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken) { // Đọc một mục từ kênh. Nếu kênh trống, var workItem = await _queue.Reader.ReadAsync(cancellationToken); return workItem; } }

Tạo dịch vụ Consumer (QueuedHostedService)

Đây là consumer của chúng ta. Nó là một BackgroundService liên tục xóa hàng đợi và thực thi công việc.

Tạo một tệp mới có tên QueuedHostedService.cs.

namespace BackgroundJobQueue; public class QueuedHostedService : BackgroundService { private readonly IBackgroundTaskQueue _taskQueue; private readonly ILogger<QueuedHostedService> _logger; public QueuedHostedService(IBackgroundTaskQueue taskQueue, ILogger<QueuedHostedService> logger) { _taskQueue = taskQueue; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("Queued Hosted Service is running."); while (!stoppingToken.IsCancellationRequested) { // Xóa một mục công việc khỏi hàng đợi var workItem = await _taskQueue.DequeueAsync(stoppingToken); try { // Thực thi mục công việc await workItem(stoppingToken); } catch (Exception ex) { _logger.LogError(ex, "Error occurred executing {WorkItem}.", nameof(workItem)); } } _logger.LogInformation("Queued Hosted Service is stopping."); } }

Khối try-catch ở đây rất quan trọng. Nó đảm bảo rằng nếu một công việc thất bại với một ngoại lệ, nó sẽ không làm sập toàn bộ dịch vụ nền.

Đăng ký các dịch vụ trong Program.cs

Tiếp theo chúng ta cần đăng ký hàng đợi và dịch vụ được lưu trữ trong container dependency injection bên trong Program.cs.

using BackgroundJobQueue; var builder = WebApplication.CreateBuilder(args); //... // Đăng ký hàng đợi công việc nền dưới dạng Singleton builder.Services.AddSingleton<IBackgroundTaskQueue>(_ => { // Bạn có thể cấu hình dung lượng của hàng đợi ở đây if (!int.TryParse(builder.Configuration["QueueCapacity"], out var queueCapacity)) { queueCapacity = 100; } return new BackgroundTaskQueue(queueCapacity); }); // Đăng ký dịch vụ được lưu trữ builder.Services.AddHostedService<QueuedHostedService>(); var app = builder.Build(); //... app.Run();

Tạo Producer (API Controller)

Cuối cùng, hãy tạo một producer. Đây sẽ là một API controller đơn giản với một endpoint xếp hàng một công việc nền mới.

Tạo một controller mới có tên JobsController.cs.

using Microsoft.AspNetCore.Mvc; namespace BackgroundJobQueue.Controllers; [ApiController] [Route("[controller]")] public class JobsController : ControllerBase { private readonly IBackgroundTaskQueue _queue; private readonly ILogger<JobsController> _logger; public JobsController(IBackgroundTaskQueue queue, ILogger<JobsController> logger) { _queue = queue; _logger = logger; } [HttpPost] public async Task<IActionResult> EnqueueJob() { // Xếp hàng một công việc mô phỏng tác vụ chạy dài await _queue.EnqueueAsync(async token => { // Mô phỏng tác vụ 5 giây var guid = Guid.NewGuid(); _logger.LogInformation("Job {Guid} started.", guid); await Task.Delay(TimeSpan.FromSeconds(5), token); _logger.LogInformation("Job {Guid} finished.", guid); }); return Ok("Job has been enqueued."); } }

Bây giờ, khi bạn chạy ứng dụng của mình và gửi yêu cầu POST đến /jobs, API sẽ phản hồi ngay lập tức với “Job has been enqueued.” trong khi tác vụ 5 giây chạy ở nền. Bạn sẽ thấy các thông báo nhật ký xuất hiện trong console của mình sau khoảng thời gian trễ.

Các cân nhắc quan trọng và chủ đề nâng cao

Shutdown

CancellationToken được truyền cho ExecuteAsync rất quan trọng. Máy chủ kích hoạt token này khi bạn dừng ứng dụng của mình. Điều kiện vòng lặp while (!stoppingToken.IsCancellationRequested) và truyền token cho DequeueAsync khiến dịch vụ dừng lắng nghe các mục mới và thoát.

Xử lý lỗi

Khối try-catch của chúng ta ngăn một công việc thất bại duy nhất làm sập consumer. Đối với các kịch bản sản xuất, bạn có thể xem xét một chiến lược xử lý lỗi nâng cao hơn, chẳng hạn như cơ chế thử lại. Các thư viện như Polly có thể được tích hợp ở đây để tự động thử lại các công việc thất bại với các chính sách như fallback.

Giới hạn của hàng đợi trong bộ nhớ

Nhược điểm lớn nhất của phương pháp này là hàng đợi là trong bộ nhớ. Nếu quy trình ứng dụng của bạn khởi động lại hoặc sập vì bất kỳ lý do gì, bất kỳ công việc nào đang chờ trong hàng đợi sẽ bị mất. Giải pháp này tốt nhất cho các tác vụ idempotent hoặc không quan trọng.

Khi nào bạn nên sử dụng thư viện bên ngoài?

Được xây dựng từ đầu, giải pháp này mạnh mẽ, nhưng điều quan trọng là phải biết khi nào nên chuyển sang các công cụ tiên tiến hơn.

Hangfire: Sử dụng Hangfire khi bạn cần tính liên tục công việc (lưu công việc vào cơ sở dữ liệu), thử lại tự động, bảng điều khiển quản lý, công việc được lên lịch và thực thi bị trì hoãn. Các tác vụ nền là giải pháp lý tưởng trong khi xử lý.
RabbitMQ: Khi xây dựng hệ thống phân tán (chẳng hạn như kiến trúc microservices), bạn có thể sử dụng message broker tùy chỉnh. Các broker này cung cấp hàng đợi có thể mở rộng cách ly các dịch vụ của bạn và đảm bảo phân phối tin nhắn.

Câu hỏi thường gặp

Q1: Điều gì xảy ra với các công việc trong hàng đợi nếu ứng dụng của tôi khởi động lại hoặc sập?

A: Vì đây là hàng đợi trong bộ nhớ, bất kỳ công việc nào chưa được xử lý sẽ bị mất khi ứng dụng đóng hoặc sập. Giải pháp này phù hợp nhất cho các tác vụ không quan trọng hoặc idempotent (tức là chúng có thể được chạy lại một cách an toàn mà không gây ra sự cố). Để đảm bảo tính liên tục kinh doanh, bạn nên sử dụng giải pháp lưu trữ công việc trong cơ sở dữ liệu hoặc message broker, chẳng hạn như Hangfire hoặc RabbitMQ.

Q2: Làm thế nào tôi có thể triển khai logic thử lại cho các công việc thất bại do ngoại lệ?

A: Khối try-catch trong QueuedHostedService ngăn toàn bộ dịch vụ nền bị sập, nhưng nó không tự động thử lại công việc thất bại. Bạn có thể tích hợp thư viện như Polly để thêm khả năng thử lại mạnh mẽ. Bạn có thể bao bọc lệnh gọi await workItem(stoppingToken); trong chính sách thử lại của Polly để tự động chạy lại công việc một số lần được cấu hình với độ trễ.

Q3: Điều này có hoạt động trong môi trường cân bằng tải với nhiều instance máy chủ không?

A: Không, triển khai này dành riêng cho instance. Hàng đợi chỉ tồn tại trong bộ nhớ của instance máy chủ nhận được yêu cầu API. Nếu bạn triển khai ứng dụng của mình trên nhiều máy chủ, một công việc được xếp hàng trên Máy chủ A sẽ chỉ được xử lý bởi Máy chủ A. Đối với hàng đợi dùng chung có thể được xử lý bởi bất kỳ máy chủ nào trong môi trường web, bạn nên sử dụng message broker tập trung, phân tán như RabbitMQ, Azure Service Bus.

Q4: Điều gì xảy ra nếu các công việc được thêm vào hàng đợi nhanh hơn dịch vụ nền có thể xử lý chúng?

A: Ứng dụng này sử dụng BoundedChannel với dung lượng cố định (ví dụ: 100 trong ví dụ của chúng ta). Khi hàng đợi đầy, việc đặt FullMode = BoundedChannelFullMode.Wait khiến producer (API controller) chờ không đồng bộ cho đến khi có chỗ trống trong hàng đợi. Điều này tạo ra “backpressure” ngăn ứng dụng của bạn hết bộ nhớ. Tuy nhiên, điều này cũng có nghĩa là nếu dịch vụ nền liên tục bị quá tải, endpoint API của bạn sẽ trở nên chậm hơn khi phản hồi vì nó chờ xếp hàng các công việc mới.

Q5: Làm thế nào tôi có thể giám sát số lượng công việc hiện tại trong hàng đợi?

A: Để giám sát cơ bản, bạn có thể tiêm IBackgroundTaskQueue vào một dịch vụ và định kỳ ghi lại số lượng hàng đợi. Cách tiếp cận nâng cao hơn là tạo một endpoint metrics tùy chỉnh (ví dụ: cho Prometheus). Bạn có thể tạo một controller tùy chỉnh truy cập thuộc tính Count của ChannelReader để báo cáo độ sâu hàng đợi hiện tại và đặt cảnh báo nếu hàng đợi phát triển quá dài.

Kết luận

Chúng ta đã xây dựng một hàng đợi công việc nền trong bộ nhớ hiệu suất cao trong ASP.NET Core chỉ sử dụng các tính năng framework tích hợp sẵn. Đây là một cách để tăng khả năng phản hồi của API và cải thiện trải nghiệm người dùng bằng cách vô hiệu hóa các tác vụ chạy dài.

Sử dụng IHostedService và System.Threading.Channels, chúng ta đã tạo ra một triển khai hiệu quả của mô hình producer-consumer. Mặc dù cách tiếp cận trong bộ nhớ này có một số hạn chế, nhưng nó là một công cụ mạnh mẽ cho nhiều kịch bản phổ biến. Khi nhu cầu của bạn phát triển, bạn có thể xem xét các công cụ phong phú tính năng hơn như Hangfire hoặc RabbitMQ, biết rằng bạn đã nắm vững các nguyên tắc cơ bản.

Cách Xây Dựng Hàng Đợi Công Việc Nền Trong Bộ Nhớ Từ Đầu Trong ASP.NET Core

Các thành phần chính của hệ thống công việc nền