Design pattern Microservices

Outbox Pattern trong ASP.NET Core: Đảm Bảo Tính Nhất Quán Khi Gửi Sự Kiện

Thương Phạm

01/02/2025

3 Min Read

44 Views

0 Comments

Table Of Content

1. Giới thiệu về Outbox Pattern
2. Cách triển khai Outbox Pattern trong ASP.NET Core
Bước 1: Tạo Bảng Outbox trong Cơ Sở Dữ Liệu
Bước 2: Tạo Entity trong EF Core
Bước 3: Lưu Sự Kiện vào Outbox Khi Thêm Dữ Liệu
Bước 4: Dịch Vụ Background Worker Để Gửi Sự Kiện
Bước 5: Đăng Ký Background Worker Trong ASP.NET Core
3. Lợi Ích của Outbox Pattern
4. Kết Luận

1. Giới thiệu về Outbox Pattern

Trong các hệ thống phân tán hoặc kiến trúc microservices, việc đảm bảo tính nhất quán của dữ liệu giữa các dịch vụ là một thách thức lớn. Nếu một ứng dụng cần ghi dữ liệu vào cơ sở dữ liệu và đồng thời gửi sự kiện (event) đến hàng đợi tin nhắn như Kafka, RabbitMQ, hoặc Azure Service Bus, có khả năng xảy ra lỗi khi một trong hai thao tác bị thất bại.

Ví dụ, nếu ứng dụng ghi dữ liệu vào SQL Server thành công nhưng bị lỗi khi gửi tin nhắn đến RabbitMQ, hệ thống có thể mất sự kiện quan trọng, dẫn đến sự không nhất quán.

Outbox Pattern giải quyết vấn đề này bằng cách:

Ghi sự kiện vào bảng Outbox trong cùng giao dịch với dữ liệu chính.
Một dịch vụ nền (background process) sau đó đọc và gửi sự kiện từ Outbox đến message queue.
Sau khi gửi thành công, sự kiện được đánh dấu là “đã xử lý” để tránh gửi trùng lặp.

2. Cách triển khai Outbox Pattern trong ASP.NET Core

Dưới đây là cách triển khai Outbox Pattern trong một ứng dụng ASP.NET Core sử dụng EF Core và SQL Server.

Bước 1: Tạo Bảng Outbox trong Cơ Sở Dữ Liệu

Trước tiên, chúng ta cần tạo một bảng để lưu trữ các sự kiện chưa được gửi.

CREATE TABLE OutboxMessages (
    Id UNIQUEIDENTIFIER PRIMARY KEY,
    CreatedAt DATETIME NOT NULL DEFAULT GETUTCDATE(),
    ProcessedAt DATETIME NULL,
    EventType NVARCHAR(255) NOT NULL,
    Payload NVARCHAR(MAX) NOT NULL
);

CREATE TABLE OutboxMessages (

Id UNIQUEIDENTIFIER PRIMARY KEY,

CreatedAt DATETIME NOT NULL DEFAULT GETUTCDATE(),

ProcessedAt DATETIME NULL,

EventType NVARCHAR(255) NOT NULL,

Payload NVARCHAR(MAX) NOT NULL

);

Id: Định danh duy nhất của sự kiện.
CreatedAt: Thời điểm sự kiện được tạo.
ProcessedAt: Thời điểm sự kiện được xử lý (nếu null, sự kiện chưa được gửi).
EventType: Loại sự kiện.
Payload: Dữ liệu JSON của sự kiện.

Bước 2: Tạo Entity trong EF Core

Tạo một entity tương ứng trong ASP.NET Core.

public class OutboxMessage
{
    public Guid Id { get; set; }
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
    public DateTime? ProcessedAt { get; set; }
    public string EventType { get; set; } = string.Empty;
    public string Payload { get; set; } = string.Empty;
}

public class OutboxMessage

{

public Guid Id { get; set; }

public DateTime CreatedAt { get; set; } = DateTime.UtcNow;

public DateTime? ProcessedAt { get; set; }

public string EventType { get; set; } = string.Empty;

public string Payload { get; set; } = string.Empty;

}

Thêm vào DbContext để EF Core quản lý:

public class AppDbContext : DbContext
{
    public DbSet&lt;OutboxMessage&gt; OutboxMessages { get; set; }

    public AppDbContext(DbContextOptions&lt;AppDbContext&gt; options) : base(options) { }
}

public class AppDbContext : DbContext

{

public DbSet<OutboxMessage> OutboxMessages { get; set; }

public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

}

Bước 3: Lưu Sự Kiện vào Outbox Khi Thêm Dữ Liệu

Khi lưu dữ liệu vào database, ta đồng thời lưu sự kiện vào bảng OutboxMessages.

public async Task AddOrderAsync(Order order)
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();
    
    try
    {
        // Lưu dữ liệu chính
        _dbContext.Orders.Add(order);
        
        // Lưu sự kiện vào Outbox
        var outboxMessage = new OutboxMessage
        {
            Id = Guid.NewGuid(),
            EventType = "OrderCreated",
            Payload = JsonSerializer.Serialize(order)
        };
        _dbContext.OutboxMessages.Add(outboxMessage);
        
        await _dbContext.SaveChangesAsync();
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}

public async Task AddOrderAsync(Order order)

{

using var transaction = await _dbContext.Database.BeginTransactionAsync();

try

{

// Lưu dữ liệu chính

_dbContext.Orders.Add(order);

// Lưu sự kiện vào Outbox

var outboxMessage = new OutboxMessage

{

Id = Guid.NewGuid(),

EventType = "OrderCreated",

Payload = JsonSerializer.Serialize(order)

};

_dbContext.OutboxMessages.Add(outboxMessage);

await _dbContext.SaveChangesAsync();

await transaction.CommitAsync();

}

catch

{

await transaction.RollbackAsync();

throw;

}

Gói gọn tất cả thao tác trong một giao dịch (transaction) để đảm bảo dữ liệu nhất quán.
Nếu có lỗi xảy ra, tất cả thay đổi sẽ bị rollback.

Bước 4: Dịch Vụ Background Worker Để Gửi Sự Kiện

Tạo một dịch vụ nền (background service) để đọc từ bảng OutboxMessages và gửi đến hàng đợi tin nhắn.

public class OutboxProcessor : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;

    public OutboxProcessor(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            using var scope = _serviceProvider.CreateScope();
            var dbContext = scope.ServiceProvider.GetRequiredService&lt;AppDbContext&gt;();

            var messages = await dbContext.OutboxMessages
                .Where(m =&gt; m.ProcessedAt == null)
                .OrderBy(m =&gt; m.CreatedAt)
                .ToListAsync(stoppingToken);

            foreach (var message in messages)
            {
                try
                {
                    // Giả sử chúng ta gửi tin nhắn qua RabbitMQ
                    await SendMessageToQueueAsync(message);
                    
                    // Đánh dấu đã xử lý
                    message.ProcessedAt = DateTime.UtcNow;
                    dbContext.Update(message);
                }
                catch (Exception ex)
                {
                    // Log lỗi nhưng không đánh dấu đã xử lý để thử lại lần sau
                    Console.WriteLine($"Lỗi gửi sự kiện: {ex.Message}");
                }
            }

            await dbContext.SaveChangesAsync(stoppingToken);
            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); // Lặp lại sau 5 giây
        }
    }

    private Task SendMessageToQueueAsync(OutboxMessage message)
    {
        Console.WriteLine($"Gửi sự kiện {message.EventType}: {message.Payload}");
        // Thực hiện gửi đến message queue ở đây (RabbitMQ, Kafka, etc.)
        return Task.CompletedTask;
    }
}

public class OutboxProcessor : BackgroundService

{

private readonly IServiceProvider _serviceProvider;

public OutboxProcessor(IServiceProvider serviceProvider)

{

_serviceProvider = serviceProvider;

}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)

{

while (!stoppingToken.IsCancellationRequested)

{

using var scope = _serviceProvider.CreateScope();

var dbContext = scope.ServiceProvider.GetRequiredService<AppDbContext>();

var messages = await dbContext.OutboxMessages

.Where(m => m.ProcessedAt == null)

.OrderBy(m => m.CreatedAt)

.ToListAsync(stoppingToken);

foreach (var message in messages)

{

try

{

// Giả sử chúng ta gửi tin nhắn qua RabbitMQ

await SendMessageToQueueAsync(message);

// Đánh dấu đã xử lý

message.ProcessedAt = DateTime.UtcNow;

dbContext.Update(message);

}

catch (Exception ex)

{

// Log lỗi nhưng không đánh dấu đã xử lý để thử lại lần sau

Console.WriteLine($"Lỗi gửi sự kiện: {ex.Message}");

}

await dbContext.SaveChangesAsync(stoppingToken);

await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); // Lặp lại sau 5 giây

}

private Task SendMessageToQueueAsync(OutboxMessage message)

{

Console.WriteLine($"Gửi sự kiện {message.EventType}: {message.Payload}");

// Thực hiện gửi đến message queue ở đây (RabbitMQ, Kafka, etc.)

return Task.CompletedTask;

}

Bước 5: Đăng Ký Background Worker Trong ASP.NET Core

Thêm dịch vụ này vào Program.cs để chạy cùng ứng dụng.

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddDbContext&lt;AppDbContext&gt;(options =&gt;
    options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));

builder.Services.AddHostedService&lt;OutboxProcessor&gt;();

var app = builder.Build();
app.Run();

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddDbContext<AppDbContext>(options =>

options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));

builder.Services.AddHostedService<OutboxProcessor>();

var app = builder.Build();

app.Run();

3. Lợi Ích của Outbox Pattern

Đảm bảo tính nhất quán: Sự kiện chỉ được gửi đi sau khi dữ liệu được lưu vào database.
Khả năng phục hồi (Resiliency): Nếu việc gửi tin nhắn thất bại, background worker có thể thử lại sau.
Không cần sử dụng Distributed Transactions: Tránh phức tạp khi sử dụng 2PC (Two-Phase Commit).
Hỗ trợ mở rộng: Dễ dàng mở rộng với nhiều hàng đợi tin nhắn khác nhau.

4. Kết Luận

Outbox Pattern là một giải pháp mạnh mẽ giúp đảm bảo tính nhất quán dữ liệu trong các hệ thống ASP.NET Core sử dụng kiến trúc microservices hoặc event-driven architecture. Việc triển khai pattern này giúp giảm thiểu lỗi khi gửi sự kiện, đảm bảo hệ thống hoạt động ổn định ngay cả khi có sự cố xảy ra.

Bạn có thể áp dụng Outbox Pattern khi làm việc với các hệ thống như Kafka, RabbitMQ, Azure Service Bus để đảm bảo dữ liệu luôn chính xác và an toàn.

Last Update: 02/02/2025