Bí kíp "Sống Sót" Cùng Outbox Pattern và SQL Server: Đảm Bảo Không Lỡ Bất Kỳ Tin Nhắn Nào!

Lê Lân

19/06/2025

Triển Khai Outbox Pattern Với SQL Server và Brighter: Đảm Bảo Tính Nhất Quán Giao Dịch

Mở Đầu

Outbox Pattern là một giải pháp phổ biến để đảm bảo tính nhất quán giao dịch giữa việc cập nhật cơ sở dữ liệu và việc gửi tin nhắn đến hệ thống nhắn tin như RabbitMQ. Trong bài viết này, chúng ta sẽ đi sâu vào việc triển khai Outbox Pattern sử dụng Microsoft SQL Server cùng thư viện Brighter để xử lý các tình huống thực tế khi tạo đơn hàng và gửi các sự kiện liên quan một cách an toàn, tránh các tin nhắn "bị bỏ sót" hoặc "bị gửi nhầm".

Trong bối cảnh các ứng dụng phân tán hiện nay, việc kết hợp giữa thay đổi dữ liệu và phát các sự kiện ra các hệ thống bên ngoài luôn là thách thức. Việc này nếu không được xử lý đúng cách có thể dẫn tới tình trạng không đồng bộ giữa dữ liệu và thông điệp, từ đó ảnh hưởng tiêu cực đến tính toàn vẹn của hệ thống. Bài viết sẽ trình bày chi tiết cách thực hiện Outbox Pattern với SQL Server, cùng cách cấu hình Brighter để đảm bảo các tin nhắn chỉ được gửi ra khi thao tác dữ liệu thành công.

Bạn sẽ được giới thiệu về:

Tổng quan dự án mẫu tạo đơn hàng với các sự kiện OrderPlaced và OrderPaid

Cách tạo bảng Outbox trong SQL Server

Cách sử dụng Brighter và các thư viện liên quan để triển khai Outbox Pattern

Cấu hình và quản lý transaction nhằm đảm bảo tính nguyên tử giữa cập nhật dữ liệu và gửi tin nhắn

Cách xử lý lỗi và tái gửi thông điệp chưa được gửi thành công

Tổng Quan Dự Án Mẫu

Ý Tưởng Dự Án

Dự án mẫu mô phỏng quá trình gửi một lệnh tạo đơn hàng. Khi đơn hàng được tạo thành công, hệ thống sẽ phát ra hai sự kiện:

OrderPlaced

OrderPaid

Nếu trong quá trình xử lý phát sinh lỗi, hệ thống sẽ không gửi bất kỳ tin nhắn nào nhằm đảm bảo tính toàn vẹn dữ liệu và thông điệp.

Yêu Cầu Phần Mềm

.NET 8+

Podman hoặc Docker để chạy container SQL Server và RabbitMQ

Các package NuGet từ Brighter như:

Paramore.Brighter.Extensions.DependencyInjection

Paramore.Brighter.Extensions.Hosting

Paramore.Brighter.MessagingGateway.RMQ

Paramore.Brighter.Outbox.MsSql

Paramore.Brighter.ServiceActivator.Extensions.DependencyInjection

Paramore.Brighter.ServiceActivator.Extensions.Hosting

Các Message Sử Dụng

Message Type	Mô Tả
CreateNewOrder	Command dùng để tạo đơn hàng với giá trị ( `Value` )
OrderPlaced	Event báo đơn hàng đã được đặt
OrderPaid	Event báo đơn hàng đã hoàn tất thanh toán

public class CreateNewOrder() : Command(Guid.NewGuid()) {
    public decimal Value { get; set; }
}

public class OrderPlaced() : Event(Guid.NewGuid()) {
    public string OrderId { get; set; } = string.Empty;
    public decimal Value { get; set; }
}

public class OrderPaid() : Event(Guid.NewGuid()) {
    public string OrderId { get; set; } = string.Empty;
}

Cách Triển Khai Chi Tiết Outbox Pattern

1. Message Mappers

Chỉ những tin nhắn sự kiện OrderPlaced và OrderPaid sẽ được xuất bản lên RabbitMQ, do đó cần viết mapper để chuyển đổi giữa object và message sử dụng JSON serialization:

public class OrderPlacedMapper : IAmAMessageMapper<OrderPlaced> {
    public Message MapToMessage(OrderPlaced request) {
        var header = new MessageHeader();
        header.Id = request.Id;
        header.TimeStamp = DateTime.UtcNow;
        header.Topic = "order-placed";
        header.MessageType = MessageType.MT_EVENT;

        var body = new MessageBody(JsonSerializer.Serialize(request));
        return new Message(header, body);
    }

    public OrderPlaced MapToRequest(Message message) {
        return JsonSerializer.Deserialize<OrderPlaced>(message.Body.Bytes)!;
    }
}

public class OrderPaidMapper : IAmAMessageMapper<OrderPaid> {
    public Message MapToMessage(OrderPaid request) {
        var header = new MessageHeader();
        header.Id = request.Id;
        header.TimeStamp = DateTime.UtcNow;
        header.Topic = "order-paid";
        header.MessageType = MessageType.MT_EVENT;

        var body = new MessageBody(JsonSerializer.Serialize(request));
        return new Message(header, body);
    }

    public OrderPaid MapToRequest(Message message) {
        return JsonSerializer.Deserialize<OrderPaid>(message.Body.Bytes)!;
    }
}

2. Request Handlers

Đối với hai sự kiện, handler sẽ ghi lại log thông tin nhận được:

public class OrderPlaceHandler(ILogger<OrderPlaceHandler> logger) : RequestHandlerAsync<OrderPlaced> {
    public override Task<OrderPlaced> HandleAsync(OrderPlaced command, CancellationToken cancellationToken = default) {
        logger.LogInformation("{OrderId} placed with value {OrderValue}", command.OrderId, command.Value);
        return base.HandleAsync(command, cancellationToken);
    }
}

public class OrderPaidHandler(ILogger<OrderPaidHandler> logger) : RequestHandlerAsync<OrderPaid> {
    public override Task<OrderPaid> HandleAsync(OrderPaid command, CancellationToken cancellationToken = default) {
        logger.LogInformation("{OrderId} paid", command.OrderId);
        return base.HandleAsync(command, cancellationToken);
    }
}

3. CreateNewOrder Handler

Handler lệnh CreateNewOrder mô phỏng xử lý nghiệp vụ:

Delay 10ms để mô phỏng quá trình tạo đơn.

Gửi OrderPlaced vào outbox.

Nếu giá trị Value chia hết cho 3 sẽ ném exception mô phỏng lỗi nghiệp vụ.

Nếu không lỗi thì gửi OrderPaid .

Toàn bộ hoạt động nằm trong một transaction dùng IUnitOfWork .

public class CreateNewOrderHandler(
    IAmACommandProcessor commandProcessor,
    IUnitOfWork unitOfWork,
    ILogger<CreateNewOrderHandler> logger) : RequestHandlerAsync<CreateNewOrder> {

    public override async Task<CreateNewOrder> HandleAsync(CreateNewOrder command, CancellationToken cancellationToken = default) {
        await unitOfWork.BeginTransactionAsync(cancellationToken);
        try {
            string id = Guid.NewGuid().ToString();
            logger.LogInformation("Creating a new order: {OrderId}", id);
            await Task.Delay(10, cancellationToken);

            _ = await commandProcessor.DepositPostAsync(new OrderPlaced { OrderId = id, Value = command.Value }, cancellationToken: cancellationToken);

            if (command.Value % 3 == 0) {
                throw new InvalidOperationException("invalid value");
            }

            _ = await commandProcessor.DepositPostAsync(new OrderPaid { OrderId = id }, cancellationToken: cancellationToken);
            await unitOfWork.CommitAsync(cancellationToken);

            return await base.HandleAsync(command, cancellationToken);
        }
        catch {
            logger.LogError("Invalid data");
            await unitOfWork.RollbackAsync(cancellationToken);
            throw;
        }
    }
}

Điểm mấu chốt: IUnitOfWork chia sẻ transaction SQL để đảm bảo tính nguyên tử giữa lưu đơn hàng và ghi tin nhắn vào Outbox. Tin nhắn chỉ được gửi nếu transaction commit thành công.

Cấu Hình Microsoft SQL Server Cho Outbox

1. Tạo Bảng OutboxMessages

Xác nhận bảng OutboxMessages tồn tại để lưu trữ tin nhắn gửi ra.

IF OBJECT_ID('OutboxMessages', 'U') IS NULL
BEGIN
    CREATE TABLE [OutboxMessages] (
        [Id] [BIGINT] NOT NULL IDENTITY,
        [MessageId] UNIQUEIDENTIFIER NOT NULL,
        [Topic] NVARCHAR(255) NULL,
        [MessageType] NVARCHAR(32) NULL,
        [Timestamp] DATETIME NULL,
        [CorrelationId] UNIQUEIDENTIFIER NULL,
        [ReplyTo] NVARCHAR(255) NULL,
        [ContentType] NVARCHAR(128) NULL,
        [Dispatched] DATETIME NULL,
        [HeaderBag] NTEXT NULL,
        [Body] NTEXT NULL,
        PRIMARY KEY ([Id])
    );
END

2. Cấu Hình Dependency Injection (DI)

services
    .AddServiceActivator(opt => {
        // Cài đặt đăng ký subscription nếu cần
    })
    .UseMsSqlOutbox(new MsSqlConfiguration(ConnectionString, "OutboxMessages"), typeof(SqlConnectionProvider), ServiceLifetime.Scoped)
    .UseMsSqlTransactionConnectionProvider(typeof(SqlConnectionProvider))
    .UseOutboxSweeper(opt => opt.BatchSize = 10);

UseMsSqlOutbox : Liên kết Outbox với SQL Server.

UseOutboxSweeper : Cấu hình dịch vụ kiểm tra và tái gửi tin nhắn chưa gửi thành công.

3. Quản Lý Transaction

Để đảm bảo nguyên tử giữa logic nghiệp vụ và phát tin nhắn, implement IMsSqlTransactionConnectionProvider và IUnitOfWork , sử dụng cùng một transaction SQL.

a. SqlConnectionProvider

public class SqlConnectionProvider(SqlUnitOfWork sqlConnection) : IMsSqlTransactionConnectionProvider {
    private readonly SqlUnitOfWork _sqlConnection = sqlConnection;

    public SqlConnection GetConnection() => _sqlConnection.Connection;

    public Task<SqlConnection> GetConnectionAsync(CancellationToken cancellationToken = default) 
        => Task.FromResult(_sqlConnection.Connection);

    public SqlTransaction? GetTransaction() => _sqlConnection.Transaction;

    public bool HasOpenTransaction => _sqlConnection.Transaction != null;

    public bool IsSharedConnection => true;
}

b. Interface IUnitOfWork

public interface IUnitOfWork {
    Task BeginTransactionAsync(CancellationToken cancellationToken, IsolationLevel isolationLevel = IsolationLevel.Serializable);
    Task CommitAsync(CancellationToken cancellationToken);
    Task RollbackAsync(CancellationToken cancellationToken);
}

c. Implement SqlUnitOfWork

public class SqlUnitOfWork(MsSqlConfiguration configuration) : IUnitOfWork {
    public SqlConnection Connection { get; } = new(configuration.ConnectionString);
    public SqlTransaction? Transaction { get; private set; }

    public async Task BeginTransactionAsync(CancellationToken cancellationToken, IsolationLevel isolationLevel = IsolationLevel.Serializable) {
        if (Transaction == null) {
            if (Connection.State != ConnectionState.Open) {
                await Connection.OpenAsync(cancellationToken);
            }
            Transaction = Connection.BeginTransaction(isolationLevel);
        }
    }

    public async Task CommitAsync(CancellationToken cancellationToken) {
        if (Transaction != null) {
            await Transaction.CommitAsync(cancellationToken);
        }
    }

    public async Task RollbackAsync(CancellationToken cancellationToken) {
        if (Transaction != null) {
            await Transaction.RollbackAsync(cancellationToken);
        }
    }

    public async Task<SqlCommand> CreateSqlCommandAsync(string sql, SqlParameter[] parameters, CancellationToken cancellationToken) {
        if (Connection.State != ConnectionState.Open) {
            await Connection.OpenAsync(cancellationToken);
        }

        SqlCommand command = Connection.CreateCommand();
        if (Transaction != null) {
            command.Transaction = Transaction;
        }

        command.CommandText = sql;
        if (parameters.Length > 0) {
            command.Parameters.AddRange(parameters);
        }

        return command;
    }
}

d. Đăng Ký Service Trong DI

services
    .AddScoped<SqlUnitOfWork, SqlUnitOfWork>()
    .TryAddScoped<IUnitOfWork>(provider => provider.GetRequiredService<SqlUnitOfWork>());

Rất quan trọng rằng transaction dùng chung cho cả lưu dữ liệu và ghi tin nhắn vào bảng OutboxMessages nhằm đảm bảo tính nhất quán dữ liệu.

Kết Luận

Qua bài viết này, chúng ta đã:

Tìm hiểu cách sử dụng Outbox Pattern với SQL Server và Brighter để đảm bảo tính nguyên tử giữa lưu dữ liệu nghiệp vụ và gửi sự kiện.

Triển khai mô hình xử lý đơn hàng với các sự kiện OrderPlaced và OrderPaid được lưu trong bảng outbox rồi mới gửi ra RabbitMQ.

Cấu hình transaction dùng chung qua IUnitOfWork và IMsSqlTransactionConnectionProvider để tránh tình trạng tin nhắn bị gửi mà dữ liệu không được lưu hoặc ngược lại.

Giới thiệu dịch vụ Outbox Sweeper đảm bảo kết nối lại và gửi lại tin nhắn khi broker tạm thời mất kết nối.

Nhấn mạnh vai trò của kiến trúc phân tán decoupled giúp tăng độ bền vững, dễ mở rộng hệ thống.

Nếu bạn quan tâm đến việc triển khai microservices hay các hệ thống event-driven, Outbox Pattern là kỹ thuật không thể thiếu. Hãy thử áp dụng mẫu này cho dự án của bạn để nâng cao độ tin cậy trong xử lý tin nhắn và dữ liệu.

Tham Khảo

Brighter with RabbitMQ và Outbox Pattern - Dev.to

Paramore.Brighter NuGet Packages

Mã nguồn mẫu Outbox với MSSQL trên Github

Microsoft Docs - SQL Server Transaction Management: https://learn.microsoft.com/en-us/sql/t-sql/language-elements/transactions-transact-sql

RabbitMQ Official Documentation: https://www.rabbitmq.com/documentation.html

Patterns of Enterprise Application Architecture - Martin Fowler

June 15, 2024