Chuyển tới nội dung chính

17.3 — 2. Hai nỗi sợ cốt lõi: mất tin & gửi trùng

Tóm tắt

Hai nỗi sợ này là hai mặt của một đồng xu, và bạn không thể loại bỏ cả hai. Message bị mất khi bạn ghi database rồi publish broker trong hai bước riêng rẽ — dual write problem, và nó không sửa được bằng try/catch. Message bị trùng vì broker đảm bảo at-least-once: nó thà giao hai lần còn hơn mất. Điều quan trọng phải hiểu ngay: "exactly-once delivery" không tồn tại ở tầng vận chuyển — đó là giới hạn lý thuyết, không phải thiếu sót của phần mềm. Cái bạn có thể đạt được là exactly-once processing: chấp nhận message tới nhiều lần và làm consumer idempotent. Và điểm hay bị bỏ sót: bảng chống trùng chỉ thật sự an toàn khi dựa vào unique constraint của database, không phải một câu if kiểm tra trước.

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

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

  • Giải thích dual write problem và vì sao try/catch không cứu được.
  • Nói đúng vì sao exactly-once delivery không tồn tại.
  • Cài đặt consumer idempotent bằng inbox hoặc khoá nghiệp vụ.
  • Xử lý đúng race condition khi hai consumer nhận cùng một message.
  • Chọn giữa at-most-once và at-least-once theo nghiệp vụ.

Nội dung bài học​

17.3.1 — Mất tin: dual write problem​

// SAI — hai hệ thống, không có transaction chung
public async Task ConvertLeadAsync(LeadId id, CancellationToken ct)
{
    lead.Convert(DateTime.UtcNow);
    await _db.SaveChangesAsync(ct);                    // (1) DB commit OK

    // <-- CRASH ở đây: DB đã đổi, broker không biết gì. MẤT VĨNH VIỄN.

    await _bus.PublishAsync(new LeadConvertedEvent(...), ct);   // (2)
}

Giữa (1) và (2) có một khoảng thời gian mà process có thể chết: deploy, OOM kill, mất điện, hoặc chỉ là GC pause dài gặp đúng lúc health check timeout. Khi đó lead đã "Won" trong database nhưng Billing không bao giờ biết.

Đổi thứ tự cũng không cứu:

// CŨNG SAI — theo hướng ngược lại
await _bus.PublishAsync(new LeadConvertedEvent(...), ct);   // (1) publish OK
// <-- CRASH: Billing tạo subscription cho một lead CHƯA hề được convert
await _db.SaveChangesAsync(ct);                            // (2)

Giờ bạn mất tính nhất quán theo hướng ngược lại — còn tệ hơn, vì Billing đã tính tiền một hợp đồng không tồn tại.

Try/catch không sửa được:

// VẪN SAI — chỉ thu hẹp cửa sổ, không đóng được
try
{
    await _bus.PublishAsync(evt, ct);
}
catch
{
    // rollback DB? SaveChanges đã COMMIT rồi, không rollback được.
    // ghi log rồi retry? process chết trước khi retry thì sao?
}

Vấn đề gốc: hai hệ thống lưu trữ khác nhau không chia sẻ transaction. Không có đoạn code nào trong tiến trình sửa được điều đó, vì chính tiến trình là thứ có thể chết.

Có đúng hai cách giải:

CáchCơ chếThực tế
Distributed transaction (2PC)Coordinator khoá cả hai bên tới khi commit xongHầu như không dùng: chậm, coordinator là điểm chết đơn, phần lớn broker hiện đại không hỗ trợ
Outbox patternGhi message vào cùng database trong cùng transaction, worker đọc ra và publish sauCách chuẩn — xem bài 17.4

17.3.2 — Gửi trùng: vì sao exactly-once không tồn tại​

Tình huống làm sinh ra message trùng, đơn giản đến mức khó tin:

1. Broker giao message cho consumer
2. Consumer xử lý xong  (đã ghi DB!)
3. Consumer gửi ACK về broker
4. Mạng đứt TRƯỚC khi ACK tới nơi
5. Broker không thấy ACK -> giao LẠI message
6. Consumer xử lý LẦN HAI cùng một message

Broker không có cách nào phân biệt "consumer chết trước khi xử lý" với "consumer xử lý xong nhưng ACK mất". Nó phải chọn:

  • At-most-once: ACK trước khi xử lý → không bao giờ trùng, nhưng có thể mất.
  • At-least-once: ACK sau khi xử lý → không bao giờ mất, nhưng có thể trùng.

Không có lựa chọn thứ ba. Đây là giới hạn lý thuyết của hệ phân tán, không phải thiếu sót của RabbitMQ hay Kafka.

Phần lớn hệ thống chọn at-least-once, vì mất dữ liệu tệ hơn xử lý lại. Khi đó trách nhiệm chuyển sang consumer: phải an toàn khi lặp.

Còn "exactly-once semantics" của Kafka? Nó có thật nhưng phạm vi hẹp: chỉ trong luồng Kafka-to-Kafka với transaction và idempotent producer. Ngay khi consumer ghi ra database ngoài, đảm bảo đó không còn áp dụng. Bạn vẫn cần idempotency.

17.3.3 — Làm consumer idempotent​

Cách 1 — khoá nghiệp vụ tự nhiên (tốt nhất khi có).

// Idempotent nho UNIQUE INDEX tren (CustomerId, Period)
public async Task HandleAsync(LeadConvertedEvent evt, CancellationToken ct)
{
    var subscription = Subscription.Create(evt.CustomerId, evt.Period);
    _db.Subscriptions.Add(subscription);

    try
    {
        await _db.SaveChangesAsync(ct);
    }
    catch (DbUpdateException ex) when (ex.IsUniqueViolation())
    {
        // Đã xử lý rồi -> coi như thành công, ACK bình thường
        _logger.LogInformation("Subscription da ton tai cho {CustomerId}", evt.CustomerId);
    }
}

Không cần bảng phụ. Database tự bảo vệ, và nó đúng kể cả khi hai consumer chạy song song.

Cách 2 — bảng inbox (khi không có khoá tự nhiên).

public sealed class InboxMessage
{
    public Guid MessageId { get; set; }          // PRIMARY KEY
    public string Type { get; set; } = null!;
    public DateTime ProcessedUtc { get; set; }
}
public async Task HandleAsync(LeadConvertedEvent evt, CancellationToken ct)
{
    await using var tx = await _db.Database.BeginTransactionAsync(ct);

    _db.InboxMessages.Add(new InboxMessage
    {
        MessageId = evt.EventId,
        Type = nameof(LeadConvertedEvent),
        ProcessedUtc = DateTime.UtcNow
    });

    try
    {
        // Ghi inbox TRƯỚC để "giành chỗ" message này
        await _db.SaveChangesAsync(ct);
    }
    catch (DbUpdateException ex) when (ex.IsUniqueViolation())
    {
        return;                                  // đã xử lý, bỏ qua
    }

    await _billing.CreateDraftSubscriptionAsync(evt.CustomerId, ct);
    await _db.SaveChangesAsync(ct);

    await tx.CommitAsync(ct);                    // inbox + nghiệp vụ commit CÙNG NHAU
}

Điểm mấu chốt: inbox và thay đổi nghiệp vụ phải nằm trong cùng transaction, cùng database. Nếu ghi inbox commit riêng còn nghiệp vụ thất bại, message bị đánh dấu "đã xử lý" trong khi thực ra chưa — và retry sẽ bỏ qua nó.

17.3.4 — Cái bẫy: kiểm tra trước rồi mới ghi​

// SAI — race condition kinh dien
if (await _db.InboxMessages.AnyAsync(m => m.MessageId == evt.EventId, ct))
    return;

await ProcessAsync(evt, ct);
_db.InboxMessages.Add(new InboxMessage { MessageId = evt.EventId });
await _db.SaveChangesAsync(ct);

Hai consumer chạy song song (điều bình thường — bạn scale ra nhiều instance):

Consumer A: AnyAsync -> false
Consumer B: AnyAsync -> false        <-- cả hai đều thấy "chưa xử lý"
Consumer A: ProcessAsync             <-- tao subscription lan 1
Consumer B: ProcessAsync             <-- tao subscription lan 2

Khoảng thời gian giữa "kiểm tra" và "ghi" là nơi lỗi sống. Cách duy nhất chắc chắn: để database quyết định bằng unique constraint và bắt lỗi vi phạm, đúng như hai ví dụ ở mục trên. Đó là một thao tác nguyên tử; câu if thì không.

Kiểm tra IsUniqueViolation phụ thuộc provider:

public static bool IsUniqueViolation(this DbUpdateException ex) => ex.InnerException switch
{
    SqlException s => s.Number is 2601 or 2627,
    PostgresException p => p.SqlState == "23505",
    _ => false
};

17.3.5 — Dọn bảng inbox​

Bảng inbox lớn vô hạn nếu không dọn. Giữ dữ liệu đủ lâu để phủ hết mọi lần retry có thể xảy ra — thường 7 ngày là dư:

DELETE TOP (5000) FROM InboxMessages
WHERE ProcessedUtc < DATEADD(day, -7, SYSUTCDATETIME());

Xoá theo lô để không khoá bảng lâu, và chạy định kỳ bằng job nền (bài 14.7).

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

Danh sách rà soát chống mất tin và trùng tin

  • •Không có chỗ nào ghi database rồi publish broker trong hai bước riêng rẽ.
  • •Đã dùng outbox thay vì try/catch quanh publish.
  • •Mỗi consumer đều an toàn khi nhận cùng message hai lần.
  • •Chống trùng dựa vào unique constraint, không dựa vào câu if kiểm tra trước.
  • •Ghi inbox và thay đổi nghiệp vụ nằm trong cùng một transaction.
  • •Đã kiểm thử bằng cách xử lý cùng một message hai lần liên tiếp.
  • •Bảng inbox có job dọn dữ liệu cũ, xoá theo lô.
  • •Không dựa vào exactly-once của broker khi consumer ghi ra database ngoài.

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

Bài 1 — Tái hiện mất tin​

Viết handler ghi database rồi publish, chèn Environment.FailFast giữa hai bước. Chạy, khởi động lại, xác nhận database đã đổi còn consumer không nhận được gì.

Tiêu chí hoàn thành: bạn nêu được vì sao không có thứ tự nào của hai thao tác giải quyết được vấn đề, và biết tên gọi của bài toán này.

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

Gợi ý. Hai hệ thống lưu trữ, không có transaction chung. Thử cả hai thứ tự và xem cái nào an toàn.

Lời giải — tái hiện:

public async Task<Result> ChotLeadAsync(LeadId id, CancellationToken ct)
{
    var lead = await _db.Leads.FirstAsync(l => l.Id == id, ct);
    var kq = lead.ChuyenSangWon(_user.ToNguoiDung(), _clock.GetUtcNow().UtcDateTime);
    if (!kq.ThanhCong) return kq;

    await _db.SaveChangesAsync(ct);                         // bước 1

    Environment.FailFast("Mô phỏng tiến trình chết");        // <- giữa hai bước

    await _bus.Publish(new LeadDaChotV1(...), ct);          // bước 2
    return Result.ThanhCong();
}
curl -X POST http://localhost:8080/leads/abc-123/chot
(tiến trình chết, không có phản hồi)
dotnet run                      # khởi động lại
SELECT Id, Status, ClosedUtc FROM Leads WHERE Id = 'abc-123';
Id         Status   ClosedUtc
abc-123    Won      2026-09-25 08:14:22
rabbitmqctl list_queues name messages
erp-tao-don-hang    0

Lead đã chốt. Message không bao giờ được gửi. Đơn hàng không bao giờ được tạo.

Và không có gì báo lỗi — từ mọi log, request đơn giản là không hoàn tất.

Thử thứ tự ngược lại:

await _bus.Publish(new LeadDaChotV1(...), ct);          // publish TRƯỚC
Environment.FailFast("Mô phỏng tiến trình chết");
await _db.SaveChangesAsync(ct);
Message ĐÃ được gửi
Lead KHÔNG được chốt trong database

-> Consumer nhận message, cố nạp lead 'abc-123'
-> Lead vẫn ở trạng thái New
-> Consumer tạo đơn hàng cho một lead chưa chốt
   (hoặc thất bại, retry, rồi vào dead-letter)

Vì sao không thứ tự nào giải quyết được:

Thứ tựNếu chết giữa chừng
Database trướcDữ liệu có, message mất — hệ quả không bao giờ xảy ra
Message trướcMessage có, dữ liệu mất — consumer xử lý thứ không tồn tại

Bạn chỉ đổi hình dạng của lỗi, không loại bỏ nó. Lý do căn bản:

Database và message broker là HAI hệ thống lưu trữ riêng biệt,
không có transaction chung.

Giữa hai thao tác LUÔN có một khoảnh khắc mà tiến trình có thể chết.

Tên gọi: dual write problem. Nó là một trường hợp của bài toán hai tướng quân — một kết quả đã được chứng minh là không giải được: hai bên giao tiếp qua kênh không tin cậy không thể đạt được sự đồng thuận chắc chắn.

Không có giao thức nào, dù phức tạp đến đâu, loại bỏ được khoảnh khắc đó.

Và distributed transaction (2PC) không phải câu trả lời:

MSDTC / XA:
    - RabbitMQ, Kafka, Redis không hỗ trợ
    - Khoá tài nguyên trên cả hai hệ thống suốt giao thức
    - Coordinator chết giữa chừng -> transaction treo, phải gỡ bằng tay
    - Không dùng được qua ranh giới cloud
    - Thông lượng giảm mạnh

Ngành đã rời bỏ 2PC cho loại bài toán này.

Câu trả lời: outbox pattern — biến hai thao tác trên hai hệ thống thành một thao tác trên một hệ thống:

public async Task<Result> ChotLeadAsync(LeadId id, CancellationToken ct)
{
    var lead = await _db.Leads.FirstAsync(l => l.Id == id, ct);
    var kq = lead.ChuyenSangWon(_user.ToNguoiDung(), _clock.GetUtcNow().UtcDateTime);
    if (!kq.ThanhCong) return kq;

    // Message ghi vào CÙNG database, CÙNG transaction
    _db.Outbox.Add(TinNhanOutbox.Tao(new LeadDaChotV1(...)));

    await _db.SaveChangesAsync(ct);      // MỘT điểm commit duy nhất
    return Result.ThanhCong();
}
Chết TRƯỚC SaveChanges:  không gì được ghi -> nhất quán
Chết SAU SaveChanges:    lead VÀ message đều được ghi -> nhất quán
                          dispatcher sẽ publish khi khởi động lại

Khoảnh khắc "chết giữa chừng" vẫn tồn tại, nhưng nó chuyển sang một chỗ vô hại:

Dispatcher publish thành công, rồi chết trước khi đánh dấu GuiLuc
-> khởi động lại, thấy message chưa đánh dấu, publish LẠI
-> message TRÙNG LẶP, không phải message MẤT

Outbox đổi "mất message" thành "message trùng" — và trùng thì xử lý được bằng consumer idempotent (bài 2 và 3), còn mất thì không xử lý được bằng gì cả.

Bảo đảm của outbox:  AT-LEAST-ONCE
Nghĩa vụ kèm theo:   consumer PHẢI idempotent

Kiểm chứng:

public async Task<Result> ChotLeadAsync(LeadId id, CancellationToken ct)
{
    // ...
    _db.Outbox.Add(TinNhanOutbox.Tao(new LeadDaChotV1(...)));
    await _db.SaveChangesAsync(ct);

    Environment.FailFast("Mô phỏng tiến trình chết");    // chết SAU commit

    return Result.ThanhCong();
}
SELECT Id, Status FROM Leads WHERE Id = 'abc-123';
SELECT Id, LoaiMessage, GuiLuc FROM Outbox WHERE GuiLuc IS NULL;
abc-123    Won

Id                                    LoaiMessage      GuiLuc
0192f8a3-...                          LeadDaChotV1     NULL      <- chờ gửi
dotnet run --project src/Crm.Worker
[OutboxDispatcher] Đã publish 1 message
rabbitmqctl list_queues name messages
erp-tao-don-hang    1

Message được gửi sau khi khởi động lại. Không mất gì.

Ba nơi khác có dual write mà ít người nhận ra:

// 1. Database + file storage
await _db.SaveChangesAsync(ct);
await _blob.UploadAsync(stream, ct);        // chết giữa -> bản ghi trỏ tới file không tồn tại

// 2. Database + cache
await _db.SaveChangesAsync(ct);
await _cache.RemoveAsync(key, ct);          // chết giữa -> cache giữ dữ liệu cũ tới hết TTL

// 3. Database + API bên ngoài
await _db.SaveChangesAsync(ct);
await _paymentApi.ChargeAsync(...);         // chết giữa -> đơn hàng có, chưa thu tiền

Trường hợp 2 ít nghiêm trọng nhất vì TTL là lưới an toàn (bài 14.4). Trường hợp 1 và 3 cần cùng cách xử lý như outbox: ghi ý định vào database trước, thực hiện sau, và đánh dấu khi xong.

Bài 2 — Tái hiện race condition khi khử trùng lặp​

Gọi handler dùng kiểu "kiểm tra trước rồi ghi" từ 10 task song song với cùng một EventId. Đếm số bản ghi được tạo.

Tiêu chí hoàn thành: bạn giải thích được vì sao kiểu "kiểm tra trước rồi ghi" sai, và nhận ra đây là cùng một mẫu lỗi đã gặp ở nhiều chỗ khác.

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

Gợi ý. Mười task cùng chạy SELECT. Tất cả thấy gì?

Lời giải — consumer có lỗi:

public class TaoDonHangConsumer : IConsumer<LeadDaChotV1>
{
    public async Task Consume(ConsumeContext<LeadDaChotV1> ctx)
    {
        var e = ctx.Message;

        // KIỂM TRA
        if (await _db.MessageDaXuLy.AnyAsync(m => m.MessageId == e.EventId))
        {
            _logger.LogInformation("Message {Id} đã xử lý, bỏ qua", e.EventId);
            return;
        }

        // rồi HÀNH ĐỘNG — có khe hở ở giữa
        var don = Order.TaoTuLead(e.LeadId, e.KhachHangId, Money.VND(e.GiaTri));
        _db.Orders.Add(don);
        _db.MessageDaXuLy.Add(new MessageDaXuLy { MessageId = e.EventId });

        await _db.SaveChangesAsync();
    }
}
var e = new LeadDaChotV1(EventId: Guid.CreateVersion7(), LeadId: leadId, GiaTri: 5_000_000);

await Task.WhenAll(Enumerable.Range(0, 10).Select(async _ =>
{
    await using var scope = _sp.CreateAsyncScope();
    var consumer = scope.ServiceProvider.GetRequiredService<TaoDonHangConsumer>();
    await consumer.Consume(TaoContext(e));
}));
SELECT COUNT(*) FROM Orders WHERE LeadId = @leadId;
SELECT COUNT(*) FROM MessageDaXuLy WHERE MessageId = @eventId;
Orders:          7
MessageDaXuLy:   7

Bảy đơn hàng cho một message. Và con số đổi mỗi lần chạy.

Vì sao kiểu "kiểm tra trước rồi ghi" sai:

t=0,000  Task 1..10 cùng chạy SELECT -> tất cả thấy "chưa xử lý"
t=0,012  Task 1..10 cùng chạy INSERT -> tất cả thành công

Kết quả của SELECT là một ảnh chụp đã cũ tại thời điểm INSERT. Giữa hai lệnh có một khe hở, và mười task chen vào được.

Đây là cùng một mẫu lỗi đã gặp ở nhiều chỗ khác — và nhận ra điều đó là phần giá trị nhất của bài:

Bối cảnhBàiTriệu chứng
Trừ tồn kho13.8Bán quá số lượng, tồn kho âm
Kiểm tra email trùng16.7Nhiều bản ghi cùng email
Idempotency key ở API14.9Nhiều đơn hàng cho một key
Job chạy một lần mỗi ngày14.11Job chạy nhiều lần
Khử trùng lặp messageBài nàyMessage xử lý nhiều lần

Mẫu chung: kiểm-tra-rồi-hành-động (check-then-act). Nó luôn sai khi có đồng thời, vì điều kiện được kiểm tra trên một ảnh chụp đã lỗi thời tại thời điểm hành động.

Và cách sửa cũng luôn giống nhau: đẩy phép kiểm tra xuống database, nơi nó nguyên tử.

Kiểm-tra-rồi-hành-động:  SELECT ... IF ... INSERT
                          -> hai lệnh, có khe hở

Nguyên tử:                INSERT + unique constraint
                          UPDATE ... WHERE <điều kiện>
                          -> một lệnh, không có khe hở

Vì sao nâng mức cô lập không phải câu trả lời thực dụng:

await using var tx = await _db.Database.BeginTransactionAsync(IsolationLevel.Serializable, ct);

Với Serializable, bạn đang khoá sự vắng mặt của một dòng, cần khoá phạm vi. Hậu quả:

Msg 1205: Transaction was deadlocked on lock resources

Bạn đổi "7 bản ghi trùng" lấy "hầu hết consumer nhận lỗi 1205" — không tốt hơn, và giờ bạn phải xử lý retry cho deadlock.

Thêm một biến thể sai thường gặp — khoá trong bộ nhớ:

private static readonly SemaphoreSlim _khoa = new(1, 1);

await _khoa.WaitAsync(ct);
try { /* kiểm tra rồi ghi */ }
finally { _khoa.Release(); }

Nó hoạt động với 10 task trong một tiến trình. Với ba instance consumer chạy song song, nó không khoá được gì — và đây chính là lỗi ở bài 14.11.

Cách sửa đúng ở bài 3.

Và một lưu ý về việc tái hiện: nếu bạn chạy thử mà chỉ thấy 1 bản ghi, có thể do MessageId đã có unique index mà bạn quên. Kiểm tra:

SELECT i.name, i.is_unique FROM sys.indexes i
JOIN sys.tables t ON t.object_id = i.object_id
WHERE t.name = 'MessageDaXuLy';

Race condition chỉ xuất hiện khi thật sự không có ràng buộc nào — và việc nó không xuất hiện trong trường hợp ngược lại chính là bằng chứng cho lời giải ở bài 3.

Bài 3 — Sửa bằng unique constraint​

Thêm primary key cho MessageId, đổi sang bắt DbUpdateException, chạy lại bài 2 và xác nhận chỉ còn đúng một bản ghi.

Tiêu chí hoàn thành: bạn nêu được vì sao unique constraint là lớp bảo đảm duy nhất không thể đi vòng qua, và biết chọn khoá idempotency đúng.

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

Gợi ý. Ai thực thi unique constraint — code của bạn hay database?

Lời giải:

public class MessageDaXuLy
{
    public Guid     MessageId  { get; set; }        // PRIMARY KEY
    public string   LoaiMessage { get; set; } = null!;
    public DateTime XuLyLuc    { get; set; }
    public string?  ChiTiet    { get; set; }
}
builder.Entity<MessageDaXuLy>(b =>
{
    b.HasKey(m => m.MessageId);
    b.Property(m => m.LoaiMessage).HasMaxLength(200).IsRequired();
    b.HasIndex(m => m.XuLyLuc);                      // cho việc dọn bảng
});
public class TaoDonHangConsumer : IConsumer<LeadDaChotV1>
{
    public async Task Consume(ConsumeContext<LeadDaChotV1> ctx)
    {
        var e = ctx.Message;

        // GHI TRƯỚC — để database quyết định
        _db.MessageDaXuLy.Add(new MessageDaXuLy
        {
            MessageId   = e.EventId,
            LoaiMessage = nameof(LeadDaChotV1),
            XuLyLuc     = _clock.GetUtcNow().UtcDateTime,
        });

        var don = Order.TaoTuLead(e.LeadId, e.KhachHangId, Money.VND(e.GiaTri));
        _db.Orders.Add(don);

        try
        {
            // Cả hai trong MỘT transaction
            await _db.SaveChangesAsync(ctx.CancellationToken);
        }
        catch (DbUpdateException ex) when (LaViPhamKhoaChinh(ex))
        {
            _logger.LogInformation("Message {Id} đã được xử lý, bỏ qua", e.EventId);
            return;                                   // ACK, không retry
        }
    }

    private static bool LaViPhamKhoaChinh(DbUpdateException ex)
        => ex.InnerException is SqlException { Number: 2601 or 2627 };
}
await Task.WhenAll(Enumerable.Range(0, 10).Select(/* ... */));
SELECT COUNT(*) FROM Orders WHERE LeadId = @leadId;
SELECT COUNT(*) FROM MessageDaXuLy WHERE MessageId = @eventId;
Orders:          1
MessageDaXuLy:   1

Chín consumer nhận DbUpdateException, ghi log, và ack. Một consumer thành công.

Vì sao unique constraint là lớp bảo đảm duy nhất không thể đi vòng qua:

LớpBắt đượcĐi vòng qua bằng
Kiểm tra trong codeTrường hợp tuần tựĐồng thời
Khoá trong bộ nhớĐồng thời trong một tiến trìnhNhiều instance
Khoá phân tán (Redis)Đồng thời nhiều instanceKhoá hết hạn, Redis failover, network partition
Unique constraintMọi trường hợp—

Lý do: phép kiểm tra được thực thi bởi chính database, ở cùng nơi và cùng thời điểm với việc ghi. Không có khe hở nào giữa "kiểm tra" và "ghi", vì chúng là một thao tác.

Ba cách mà các lớp khác thất bại trong im lặng:

Khoá Redis TTL 10 phút, xử lý mất 12 phút
    -> khoá tự mở ở phút 10 -> instance khác vào -> trùng lặp
    -> và không ai biết

Redis failover
    -> khoá mất -> nhiều instance cùng giữ khoá

Network partition
    -> hai phía cùng tin mình có khoá

Unique constraint không có khái niệm hết hạn, không phụ thuộc vào hạ tầng khác, và được bảo vệ bởi cùng cơ chế toàn vẹn của mọi dữ liệu nghiệp vụ khác.

Ba chi tiết quan trọng trong phần cài đặt:

1. Ghi bản ghi khử trùng lặp và dữ liệu nghiệp vụ trong CÙNG transaction.

// SAI — hai transaction riêng
_db.MessageDaXuLy.Add(...);
await _db.SaveChangesAsync(ct);        // transaction 1

_db.Orders.Add(don);
await _db.SaveChangesAsync(ct);        // transaction 2 — chết ở đây thì mất đơn hàng

Với hai transaction, bạn tạo ra một dual write mới — đúng vấn đề ở bài 1.

2. Bắt đúng mã lỗi. DbUpdateException bao gồm nhiều loại vi phạm:

// Quá rộng — nuốt cả vi phạm khoá ngoại, ràng buộc CHECK
catch (DbUpdateException) { return; }

// Đúng
catch (DbUpdateException ex) when (ex.InnerException is SqlException { Number: 2601 or 2627 })

Với PostgreSQL:

catch (DbUpdateException ex) when (ex.InnerException is PostgresException { SqlState: "23505" })

3. ACK message, đừng retry. Message trùng lặp không phải lỗi — nó là điều được mong đợi trong hệ thống at-least-once. Retry chỉ tạo thêm tải.

Chọn khoá idempotency đúng — đây là phần quan trọng nhất:

KhoáBảo vệ khỏiKhông bảo vệ khỏi
MessageIdCùng một message gửi lạiHai message khác nhau cho cùng một hành động
Định danh nghiệp vụCả hai—
Kịch bản: outbox dispatcher publish message, rồi chết trước khi đánh dấu
    -> khởi động lại, publish LẠI cùng EventId
    -> MessageId khử trùng được

Kịch bản: người dùng bấm "Chốt lead" hai lần
    -> hai request, hai EventId KHÁC NHAU, cùng LeadId
    -> MessageId KHÔNG khử trùng được -> hai đơn hàng

Dùng định danh nghiệp vụ khi có:

public class Order
{
    public OrderId Id { get; private set; }
    public LeadId  LeadId { get; private set; }      // unique — một lead, một đơn hàng
}
builder.Entity<Order>()
    .HasIndex(o => o.LeadId)
    .IsUnique()
    .HasFilter("[IsDeleted] = 0");
try
{
    _db.Orders.Add(Order.TaoTuLead(e.LeadId, e.KhachHangId, Money.VND(e.GiaTri)));
    await _db.SaveChangesAsync(ct);
}
catch (DbUpdateException ex) when (LaViPhamUnique(ex, "IX_Orders_LeadId"))
{
    _logger.LogInformation("Đơn hàng cho lead {LeadId} đã tồn tại", e.LeadId);
    return;
}

Với khoá nghiệp vụ, cả hai kịch bản đều được bảo vệ — và bạn không cần bảng MessageDaXuLy riêng.

Ba mức, theo thứ tự nên chọn:

1. Định danh nghiệp vụ có unique constraint tự nhiên     <- tốt nhất
   (một lead -> một đơn hàng; một tháng -> một hoá đơn)

2. Khoá tổ hợp từ nghiệp vụ
   $"tao-don:{leadId}"  hoặc  $"hoa-don:{khachHangId}:{thang:yyyy-MM}"

3. MessageId                                              <- chỉ khi không có 1 hoặc 2

Và nhớ dọn bảng khử trùng lặp:

DELETE TOP (5000) FROM MessageDaXuLy
WHERE XuLyLuc < DATEADD(DAY, -30, SYSUTCDATETIME());

30 ngày là mốc hợp lý: dài hơn mọi chuỗi retry và thời gian message có thể nằm trong dead-letter queue. Không dọn thì bảng lớn vô hạn, và truy vấn INSERT chậm dần vì index lớn dần.

Test cho mọi consumer:

[Fact]
public async Task Consumer_xu_ly_message_hai_lan_chi_tao_mot_ban_ghi()
{
    var e = new LeadDaChotV1(Guid.CreateVersion7(), leadId, khachHangId, 5_000_000, "VND");

    await _harness.Bus.Publish(e);
    await _harness.Bus.Publish(e);                       // GỬI LẠI cùng message
    await _harness.InactivityTask;

    (await DemDonHangAsync(leadId)).Should().Be(1);
}

Test này nên là bắt buộc cho mọi consumer. Nó rẻ, và nó biến một giả định ngầm — "consumer này idempotent" — thành một khẳng định được kiểm chứng ở mỗi lần chạy CI.

Tự kiểm tra​

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

Dual write problem là gì?

Là việc ghi vào hai hệ thống lưu trữ không chia sẻ transaction, ví dụ database và broker. Giữa hai thao tác luôn có một khoảng mà tiến trình có thể chết, nên một bên đã đổi còn bên kia thì chưa. Đổi thứ tự chỉ đổi hướng mất nhất quán chứ không sửa được.

Vì sao try/catch không sửa được dual write?

Vì SaveChanges đã commit nên không rollback được, và nếu ghi log để retry thì tiến trình vẫn có thể chết trước khi retry chạy. Vấn đề gốc là tiến trình chính là thứ có thể chết, nên không đoạn code nào bên trong nó giải quyết được.

Vì sao exactly-once delivery không tồn tại?

Vì broker không phân biệt được consumer chết trước khi xử lý với consumer xử lý xong nhưng ACK bị mất trên đường về. Nó buộc phải chọn ACK trước khi xử lý, tức có thể mất, hoặc ACK sau khi xử lý, tức có thể trùng. Không có lựa chọn thứ ba.

Exactly-once semantics của Kafka có mâu thuẫn với điều đó không?

Không, vì phạm vi của nó hẹp, chỉ áp dụng cho luồng Kafka sang Kafka với transaction và idempotent producer. Ngay khi consumer ghi ra một database ngoài thì đảm bảo đó không còn áp dụng và bạn vẫn cần idempotency.

Vì sao kiểm tra tồn tại trước rồi mới ghi là sai?

Vì giữa lúc kiểm tra và lúc ghi, một consumer khác có thể kiểm tra và cũng thấy chưa xử lý, nên cả hai cùng xử lý. Cách chắc chắn là để database quyết định bằng unique constraint rồi bắt lỗi vi phạm, vì đó là một thao tác nguyên tử.

Vì sao inbox và thay đổi nghiệp vụ phải cùng transaction?

Vì nếu ghi inbox commit riêng mà phần nghiệp vụ thất bại, message bị đánh dấu đã xử lý trong khi thực ra chưa, và lần retry sau sẽ bỏ qua nó. Khi đó bạn mất dữ liệu một cách im lặng.

Kết luận​

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

  1. Dual write không sửa được bằng code trong tiến trình. Dùng outbox.
  2. At-least-once là mặc định thực tế. Idempotency là trách nhiệm của consumer, không phải của broker.
  3. Chống trùng phải dựa vào unique constraint, vì câu if kiểm tra trước luôn có race condition.

Tham khảo​

Điều hướng​

Bài liên quan​