17.9 — 8. MassTransit / NServiceBus / EasyNetQ

Tóm tắt

Ba thư viện này nằm ở ba mức trừu tượng khác nhau, và chọn sai mức tốn kém hơn chọn sai sản phẩm. MassTransit là lựa chọn mặc định của phần lớn dự án .NET: abstraction dày, có sẵn outbox, saga, retry hai tầng — nhưng hãy kiểm tra điều khoản giấy phép của phiên bản bạn định dùng trước khi cam kết, vì mô hình cấp phép của dự án đã thay đổi theo thời gian. NServiceBus mạnh, tài liệu vào loại tốt nhất trong hệ sinh thái, và thu phí thương mại một cách minh bạch từ đầu. EasyNetQ mỏng, chỉ RabbitMQ, không có saga hay outbox — nhưng đôi khi mỏng chính là thứ bạn cần. Điều quan trọng hơn cả ba lựa chọn: giữ code nghiệp vụ không biết gì về thư viện bạn chọn, để đổi thư viện chỉ phải viết lại lớp hạ tầng.

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

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

• So sánh ba thư viện theo mức trừu tượng, tính năng và chi phí.
• Biết kiểm tra giấy phép trước khi đưa thư viện vào dự án dài hạn.
• Cấu hình MassTransit với RabbitMQ ở mức đủ dùng cho production.
• Giữ code nghiệp vụ không phụ thuộc thư viện messaging.
• Nhận ra khi nào không cần thư viện nào cả.

Nội dung bài học

17.9.1 — Ba mức trừu tượng

	MassTransit	NServiceBus	EasyNetQ
Mức trừu tượng	Cao	Cao	Thấp
Broker hỗ trợ	RabbitMQ, Azure SB, SQS, Kafka	RabbitMQ, Azure SB, SQS, MSMQ	Chỉ RabbitMQ
Outbox tích hợp	Có	Có	Không
Saga / state machine	Có	Có	Không
Retry hai tầng	Có	Có	Cơ bản
Scheduling	Có	Có	Không
Chi phí	Kiểm tra theo phiên bản	Thương mại	Miễn phí (MIT)
Độ dốc học	Trung bình	Cao	Thấp
Tài liệu	Tốt	Rất tốt	Đủ dùng

Ba dòng đáng chú ý nhất: broker hỗ trợ (EasyNetQ khoá bạn vào RabbitMQ), saga (nếu cần điều phối quy trình nhiều bước thì EasyNetQ loại khỏi danh sách), và chi phí.

17.9.2 — Về giấy phép: kiểm tra trước khi cam kết

Đây là rủi ro ít người tính tới khi chọn thư viện: mô hình cấp phép của dự án mã nguồn mở có thể thay đổi. Một thư viện miễn phí hôm nay có thể yêu cầu trả phí ở phiên bản sau, và khi đó bạn đứng trước ba lựa chọn đều không dễ chịu: trả tiền, ở lại phiên bản cũ không còn được cập nhật, hoặc viết lại lớp hạ tầng messaging.

MassTransit là ví dụ cụ thể: mô hình cấp phép của nó đã có thay đổi giữa các phiên bản lớn. Hãy đọc trang giấy phép chính thức của đúng phiên bản bạn định dùng trước khi đưa vào dự án dài hạn, đừng dựa vào bài viết cũ hay thông tin truyền miệng.

Ba việc nên làm cho mọi thư viện nền tảng, không riêng messaging:

1. Đọc giấy phép của phiên bản cụ thể, không phải của repo nói chung.
2. Ước lượng chi phí thoát — nếu ngày mai phải bỏ thư viện này, bao nhiêu file phải sửa?
3. Giữ abstraction mỏng của riêng bạn để câu trả lời cho câu hỏi trên là "vài file", không phải "cả dự án".

Điểm 3 là thứ bạn kiểm soát được, và mục 17.9.5 nói cách làm.

17.9.3 — MassTransit ở mức đủ dùng

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<LeadConvertedConsumer>();

    // Outbox: message chi publish khi transaction commit
    x.AddEntityFrameworkOutbox<AppDbContext>(o =>
    {
        o.QueryDelay = TimeSpan.FromSeconds(5);
        o.UseSqlServer();
        o.UseBusOutbox();
    });

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("RabbitMq"));

        cfg.ReceiveEndpoint("lead-converted", e =>
        {
            e.PrefetchCount = 16;                        // gioi han message chua ACK
            e.UseMessageRetry(r => r.Immediate(3));      // loi chop nhoang
            e.UseScheduledRedelivery(r => r.Intervals(   // loi keo dai
                TimeSpan.FromMinutes(1),
                TimeSpan.FromMinutes(5),
                TimeSpan.FromMinutes(15)));

            e.ConfigureConsumer<LeadConvertedConsumer>(context);
        });
    });
});

Cấu hình này đã có ba thứ cần cho production: outbox chống mất tin (bài 17.4), retry hai tầng (bài 17.6), và prefetch để phân bổ tải đều giữa consumer (bài 17.4).

Saga là thứ khó tự viết nhất và là lý do chính đáng nhất để dùng thư viện:

public sealed class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State AwaitingPayment { get; private set; } = null!;
    public State Completed { get; private set; } = null!;

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Initially(
            When(OrderSubmitted)
                .Then(c => c.Saga.SubmittedAtUtc = DateTime.UtcNow)
                .TransitionTo(AwaitingPayment));

        During(AwaitingPayment,
            When(PaymentReceived)
                .TransitionTo(Completed)
                .Finalize(),
            When(PaymentTimeout)
                .Publish(c => new OrderCancelledEvent(c.Saga.CorrelationId))
                .Finalize());
    }
}

Trạng thái của saga được lưu bền vào database, nên nó sống sót qua restart. Tự viết đúng phần này — kể cả timeout, đồng thời và phục hồi — tốn nhiều công hơn người ta tưởng (saga pattern).

17.9.4 — EasyNetQ: khi mỏng là đủ

using var bus = RabbitHutch.CreateBus("host=localhost;username=guest;password=guest");

await bus.PubSub.PublishAsync(new LeadConvertedEvent(leadId, customerId));

await bus.PubSub.SubscribeAsync<LeadConvertedEvent>("billing", async evt =>
{
    await billingService.CreateSubscriptionAsync(evt.CustomerId);
});

Ba dòng, chạy được ngay, MIT, không có gì để hiểu thêm.

Đánh đổi: không outbox, không saga, không scheduling, không retry hai tầng — bạn tự viết hoặc không có. Với một service nhỏ chỉ publish vài loại event đơn giản, đó là đánh đổi hợp lý. Với hệ thống có quy trình nhiều bước, bạn sẽ dần tự viết lại những gì MassTransit đã có, chỉ tệ hơn.

17.9.5 — Giữ code nghiệp vụ độc lập

Đây là phần quan trọng nhất của bài, và nó áp dụng cho cả ba lựa chọn.

// SAI — Application layer phu thuoc truc tiep MassTransit
public sealed class ConvertLeadHandler(IPublishEndpoint publishEndpoint)   // <- MassTransit
{
    public async Task Handle(ConvertLeadCommand cmd, CancellationToken ct)
    {
        await publishEndpoint.Publish(new LeadConvertedEvent(...), ct);
    }
}

Mỗi handler là một chỗ phải sửa nếu đổi thư viện — và trong một dự án thật đó là hàng chục file.

// ĐÚNG — Application chỉ biết interface của CHÍNH MÌNH
// Crm.Application/Abstractions/IEventPublisher.cs
public interface IEventPublisher
{
    Task PublishAsync<TEvent>(TEvent @event, CancellationToken ct) where TEvent : class;
}

// Crm.Infrastructure/Messaging/MassTransitEventPublisher.cs
public sealed class MassTransitEventPublisher(IPublishEndpoint publishEndpoint) : IEventPublisher
{
    public Task PublishAsync<TEvent>(TEvent @event, CancellationToken ct) where TEvent : class
        => publishEndpoint.Publish(@event, ct);
}

Đây chính là dependency rule (bài 16.4) áp dụng cho messaging. Đổi sang EasyNetQ chỉ phải viết lại một file hạ tầng.

Hai lợi ích thực tế nữa, có ngay cả khi bạn không bao giờ đổi thư viện:

• Test không cần broker. Một FakeEventPublisher thu message vào List là đủ để assert.
• Application không bị ràng buộc vào vòng đời DI của thư viện, thứ hay gây lỗi "cannot consume scoped service" khó chẩn đoán (bài 7.5).

Giới hạn của lời khuyên này: đừng bọc tính năng nâng cao. Saga state machine của MassTransit không thể bọc sau một interface chung mà vẫn giữ giá trị — nếu bạn dùng saga, hãy chấp nhận phụ thuộc và cô lập nó trong một project riêng.

17.9.6 — Khi nào không cần thư viện nào

Nếu bạn chỉ publish vài loại event lên một broker duy nhất và đã tự viết outbox (bài 17.8), client gốc là đủ:

// RabbitMQ.Client trực tiếp — ít phụ thuộc, kiểm soát hoàn toàn
public sealed class RabbitMqEventPublisher(IConnection connection) : IEventPublisher
{
    public async Task PublishAsync<TEvent>(TEvent @event, CancellationToken ct) where TEvent : class
    {
        await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

        var body = JsonSerializer.SerializeToUtf8Bytes(@event);
        await channel.BasicPublishAsync(
            exchange: "crm-events",
            routingKey: typeof(TEvent).Name,
            body: body,
            cancellationToken: ct);
    }
}

Chọn cách này khi: một broker, không saga, không cần scheduling, và đội muốn ít phụ thuộc bên ngoài. Chọn thư viện khi: cần saga, cần retry/DLQ có sẵn, hoặc muốn đổi broker mà không sửa code nghiệp vụ.

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

Danh sách rà soát khi chọn thư viện messaging

• •Đã đọc giấy phép của đúng phiên bản định dùng, không dựa vào thông tin cũ.
• •Đã ước lượng chi phí thoát nếu phải bỏ thư viện này.
• •Chỉ dùng một thư viện messaging trong dự án.
• •Application layer chỉ phụ thuộc interface của chính mình.
• •Có fake publisher để test không cần broker.
• •Đã cấu hình prefetch, retry hai tầng và outbox nếu dùng MassTransit.
• •Nếu dùng saga, phần phụ thuộc được cô lập trong một project riêng.
• •Đã cân nhắc client gốc nếu nhu cầu thật sự đơn giản.

Bài tập áp dụng

Bài 1 — Đọc giấy phép

Mở trang giấy phép chính thức của MassTransit, NServiceBus và EasyNetQ, ghi lại điều khoản áp dụng cho phiên bản mới nhất và cho phiên bản bạn định dùng.

Tiêu chí hoàn thành: bạn ghi lại được giấy phép theo từng phiên bản, và hiểu vì sao giấy phép là một quyết định kiến trúc chứ không phải thủ tục pháp lý.

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

Gợi ý. Một thư viện đổi giấy phép ở phiên bản mới. Phiên bản cũ bạn đang dùng có bị ảnh hưởng không? Còn bản vá bảo mật thì sao?

Lời giải — kiểm tra thực tế:

# Xem giấy phép của gói đang dùng
dotnet list package --include-transitive | grep -i "masstransit\|nservicebus\|easynetq"

# Hoặc đọc metadata của gói đã tải
unzip -p ~/.nuget/packages/masstransit/*/masstransit.*.nupkg '*.nuspec' \
  | grep -E "<license|<version"

# Xem trên nuget.org
curl -s "https://api.nuget.org/v3/registration5-gz-semver2/masstransit/index.json" | jq

Ba thư viện, ba mô hình giấy phép khác nhau:

	Mô hình	Điều cần kiểm tra
MassTransit	Mã nguồn mở, đã công bố chuyển sang thương mại cho phiên bản tương lai	Phiên bản nào là ranh giới; điều khoản cho bản cũ
NServiceBus	Thương mại từ đầu, có bản miễn phí giới hạn	Giới hạn của bản miễn phí; giá theo endpoint hay theo máy chủ
EasyNetQ	Mã nguồn mở (MIT)	Mức độ bảo trì, số người đóng góp

Quan trọng: đừng tin bảng này. Điều khoản giấy phép thay đổi, và bảng trong một tài liệu học luôn có nguy cơ lỗi thời. Đọc trang chính thức tại thời điểm bạn quyết định:

MassTransit:   https://masstransit.io
NServiceBus:   https://particular.net/licensing
EasyNetQ:      https://github.com/EasyNetQ/EasyNetQ/blob/master/LICENSE

Ghi lại theo mẫu này:

## Giấy phép thư viện messaging — kiểm tra ngày 2026-09-25

| Thư viện | Phiên bản đang dùng | Giấy phép của phiên bản đó | Phiên bản mới nhất | Giấy phép |
|---|---|---|---|---|
| MassTransit | 8.2.x | <ghi từ nuspec> | <mới nhất> | <từ trang chính thức> |

### Câu hỏi đã xác nhận
- [ ] Bản vá bảo mật cho phiên bản cũ có được phát hành không, và trong bao lâu?
- [ ] Nếu nâng lên phiên bản có giấy phép thương mại, chi phí tính theo gì?
- [ ] Bộ phận pháp lý đã duyệt chưa?
- [ ] Có ràng buộc nào về việc phân phối phần mềm cho khách hàng không?

### Nguồn
- <đường dẫn trang giấy phép>, truy cập ngày ...

Vì sao giấy phép là quyết định kiến trúc:

1. Nó quyết định bạn có nâng cấp được không.

Phiên bản đang dùng: miễn phí
Phiên bản mới:       thương mại

-> Bản vá bảo mật chỉ có ở phiên bản mới?
   -> hoặc trả phí, hoặc chạy với lỗ hổng, hoặc di chuyển sang thư viện khác
-> Cả ba lựa chọn đều là quyết định lớn, và không có lựa chọn nào rẻ

2. Chi phí có thể tăng theo quy mô hệ thống.

Giá theo endpoint hoặc theo instance
-> chi phí tăng khi bạn scale
-> một quyết định kiến trúc (tách thêm dịch vụ) trở thành quyết định tài chính

3. Di chuyển sang thư viện khác không rẻ — trừ khi bạn đã tách abstraction (bài 2).

grep -rn "MassTransit\|IConsumer<\|ConsumeContext" --include="*.cs" src/ | wc -l

187

187 chỗ. Nếu phải đổi, đó là một dự án, không phải một nhiệm vụ.

4. Nó ảnh hưởng tới việc phân phối phần mềm. Nếu bạn bán phần mềm cài tại chỗ cho khách hàng, một số giấy phép yêu cầu mỗi bản cài phải có giấy phép riêng — điều này đổi hẳn mô hình kinh doanh.

Bốn câu hỏi phải trả lời trước khi chọn:

1. Giấy phép của phiên bản TÔI SẼ DÙNG là gì?
   (không phải phiên bản mới nhất)

2. Nếu tôi không nâng cấp, tôi có nhận được bản vá bảo mật không, trong bao lâu?

3. Nếu phải nâng cấp, chi phí tính theo gì và ước tính bao nhiêu ở quy mô của tôi?

4. Nếu phải đổi thư viện, mất bao lâu?
   -> câu trả lời phụ thuộc vào việc tôi đã tách abstraction chưa (bài 2)

Câu 4 là câu duy nhất bạn kiểm soát được. Ba câu đầu phụ thuộc vào quyết định của người khác; câu 4 phụ thuộc vào thiết kế của bạn.

Đưa việc kiểm tra giấy phép vào CI:

<PackageReference Include="NuGetDefense" Version="..." PrivateAssets="all" />

dotnet tool install -g dotnet-project-licenses
dotnet-project-licenses -i src/Crm.sln --allowed-license-types MIT,Apache-2.0,BSD-3-Clause \
  --output-directory artifacts --export-license-texts

LICENSE VIOLATION: MassTransit 9.0.0 — <giấy phép không nằm trong danh sách cho phép>

Kiểm tra này bắt được thời điểm một gói âm thầm đổi giấy phép ở một bản nâng cấp — điều đã xảy ra với nhiều thư viện .NET phổ biến trong vài năm qua, và thường không ai nhận ra cho tới khi bộ phận pháp lý hỏi.

Và một điểm đáng nói: giấy phép thương mại không phải điều xấu. Một thư viện có nguồn thu ổn định thường được bảo trì tốt hơn, có tài liệu tốt hơn, và có hỗ trợ khi bạn gặp sự cố lúc 2 giờ sáng.

Vấn đề không phải "miễn phí hay trả phí" mà là biết trước và quyết định có ý thức — thay vì phát hiện ra sau khi đã có 187 chỗ phụ thuộc.

Phương án thứ tư ít được nhắc: dùng thẳng thư viện client.

// RabbitMQ.Client — chính thức, MIT, không có lớp trừu tượng
using RabbitMQ.Client;

var factory = new ConnectionFactory { Uri = new Uri(chuoiKetNoi) };
await using var conn = await factory.CreateConnectionAsync(ct);
await using var channel = await conn.CreateChannelAsync(cancellationToken: ct);

// Confluent.Kafka — chính thức, Apache 2.0
using var producer = new ProducerBuilder<string, string>(config).Build();

Bạn tự viết retry, serialize, khử trùng lặp, và định tuyến — khoảng 300–500 dòng cho một hệ thống vừa. Đổi lại: không phụ thuộc giấy phép, không bị khoá vào framework, và bạn hiểu chính xác điều gì đang xảy ra.

Với một hệ thống có dưới 10 loại message và một broker, đây là lựa chọn hoàn toàn hợp lý — và nó là lựa chọn mà nhiều đội chọn sau khi đã trải qua một lần đổi giấy phép.

---

Bài 2 — Tách abstraction

Trong dự án của bạn, tìm mọi chỗ Application layer tham chiếu trực tiếp thư viện messaging. Đưa về một interface IEventPublisher và đếm số file phải sửa.

Tiêu chí hoàn thành: bạn đo được số file trước và sau, và nêu được ranh giới của việc tách — cái gì tách được, cái gì không.

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

Gợi ý. IPublishEndpoint tách được. IConsumer<T> thì sao?

Lời giải — đo hiện trạng:

grep -rln "MassTransit" --include="*.cs" src/ | sort | uniq -c
grep -rn "IPublishEndpoint\|ISendEndpoint\|IBus\b" --include="*.cs" src/ | wc -l
grep -rn "IConsumer<\|ConsumeContext<" --include="*.cs" src/ | wc -l

src/Crm.Application/   34 file
src/Crm.Infrastructure/ 12 file
src/Crm.Api/            3 file

IPublishEndpoint và bạn bè:  47 chỗ
IConsumer và ConsumeContext: 28 chỗ

Tách phía publish:

// Crm.Application/Abstractions/IEventPublisher.cs — KHÔNG tham chiếu MassTransit
public interface IEventPublisher
{
    Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class;
    Task PublishManyAsync<T>(IEnumerable<T> events, CancellationToken ct = default) where T : class;
}

// Crm.Infrastructure/Messaging/MassTransitEventPublisher.cs
public sealed class MassTransitEventPublisher(IPublishEndpoint endpoint) : IEventPublisher
{
    public Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class
        => endpoint.Publish(@event, ct);

    public Task PublishManyAsync<T>(IEnumerable<T> events, CancellationToken ct = default)
        where T : class
        => endpoint.PublishBatch(events, ct);
}

// Trước
public class ChotLeadHandler(IPublishEndpoint bus, CrmDbContext db) { }

// Sau
public class ChotLeadHandler(IEventPublisher publisher, CrmDbContext db) { }

Số file phải sửa: 34
Thời gian: khoảng 2 giờ (chủ yếu là tìm và thay)

grep -rn "MassTransit" --include="*.cs" src/Crm.Application | wc -l

0

Tách phía consume — khó hơn nhiều:

// Không tách hoàn toàn được — IConsumer<T> là điểm móc của framework
public class TaoSubscriptionConsumer : IConsumer<LeadConvertedV1>
{
    public async Task Consume(ConsumeContext<LeadConvertedV1> ctx) { }
}

Cách tốt nhất: giữ consumer mỏng, đẩy logic vào handler không phụ thuộc framework:

// Crm.Application — KHÔNG biết MassTransit
public interface IEventHandler<in TEvent> where TEvent : class
{
    Task HandleAsync(TEvent @event, CancellationToken ct);
}

public sealed class TaoSubscriptionHandler(CrmDbContext db, ILogger<TaoSubscriptionHandler> logger)
    : IEventHandler<LeadConvertedV1>
{
    public async Task HandleAsync(LeadConvertedV1 e, CancellationToken ct)
    {
        db.MessageDaXuLy.Add(new MessageDaXuLy { MessageId = e.EventId, XuLyLuc = DateTime.UtcNow });
        db.Subscriptions.Add(Subscription.Tao(e.CustomerId, e.GiaTri));

        try { await db.SaveChangesAsync(ct); }
        catch (DbUpdateException ex) when (LaViPhamKhoaChinh(ex))
        {
            logger.LogInformation("Message {Id} đã xử lý, bỏ qua", e.EventId);
        }
    }
}

// Crm.Infrastructure — lớp bọc MỎNG, generic, viết MỘT lần cho mọi event
public sealed class MassTransitConsumerAdapter<TEvent>(IEventHandler<TEvent> handler)
    : IConsumer<TEvent> where TEvent : class
{
    public Task Consume(ConsumeContext<TEvent> ctx)
        => handler.HandleAsync(ctx.Message, ctx.CancellationToken);
}

services.AddMassTransit(x =>
{
    x.AddConsumer<MassTransitConsumerAdapter<LeadConvertedV1>>();
    x.AddConsumer<MassTransitConsumerAdapter<DonHangDaTaoV1>>();

    x.UsingRabbitMq((ctx, cfg) => cfg.ConfigureEndpoints(ctx));
});

services.AddScoped<IEventHandler<LeadConvertedV1>, TaoSubscriptionHandler>();

Số file phải sửa: 28 consumer -> 28 handler + 1 adapter generic
Kết quả: Crm.Application hoàn toàn sạch

Ranh giới của việc tách — cái gì tách được, cái gì không:

Khái niệm	Tách được	Vì sao
Publish, Send	Có	Chỉ là một lời gọi phương thức
Logic xử lý message	Có	Qua IEventHandler<T>
Serialize	Có	Cấu hình ở Infrastructure
Retry policy	Một phần	Cấu hình được ở Infrastructure, nhưng ngữ nghĩa khác nhau giữa các thư viện
Consumer registration	Không	Là API của framework
Endpoint configuration	Không	Cú pháp và khái niệm khác hẳn nhau
Saga, state machine	Không	Mô hình lập trình riêng của từng thư viện
Request-reply	Khó	Ngữ nghĩa khác nhau đáng kể
Scheduling, delayed message	Không	Phụ thuộc broker và cách thư viện cài

Ba dòng cuối là ranh giới thật. Nếu bạn dùng saga của MassTransit hay scheduling của NServiceBus, việc đổi thư viện là viết lại, không phải thay một lớp adapter.

Tách được:   khoảng 80% code — logic nghiệp vụ, publish, consume
KHÔNG tách:  khoảng 20% — cấu hình, đăng ký, saga, scheduling
             -> nhưng 20% này nằm GỌN trong Infrastructure

Lợi ích thật của việc tách — và "đổi thư viện" không phải lợi ích chính:

1. Test không cần broker (bài 3) — đây là lợi ích được dùng hằng ngày.

2. Application layer không phụ thuộc hạ tầng — kiến trúc test ở bài 16.2 bảo vệ được ranh giới này:

[Fact]
public void Application_khong_duoc_phu_thuoc_thu_vien_messaging()
{
    var kq = Types.InAssembly(typeof(ChotLeadHandler).Assembly)
        .ShouldNot().HaveDependencyOnAny("MassTransit", "NServiceBus", "EasyNetQ",
                                         "RabbitMQ.Client", "Confluent.Kafka")
        .GetResult();

    kq.IsSuccessful.Should().BeTrue();
}

3. Nâng cấp phiên bản lớn dễ hơn. MassTransit v8 sang v9 đổi API ở một số chỗ — với abstraction, bạn sửa một file adapter thay vì 34 file.

4. Đổi thư viện khả thi — lợi ích ít được dùng nhất, nhưng cũng là lợi ích thật khi giấy phép đổi.

Và một cảnh báo: đừng tách quá tay.

// QUÁ TAY — dựng lại toàn bộ API của MassTransit
public interface IMessageBus
{
    Task PublishAsync<T>(T msg, CancellationToken ct);
    Task SendAsync<T>(T msg, Uri destination, CancellationToken ct);
    Task<TResponse> RequestAsync<TRequest, TResponse>(TRequest req, CancellationToken ct);
    Task ScheduleAsync<T>(T msg, DateTime scheduledTime, CancellationToken ct);
    ISagaBuilder<TState> CreateSaga<TState>();
    // ...
}

Đây là dấu hiệu quen thuộc: nếu abstraction của bạn có hình dạng giống hệt thư viện bên dưới, nó không phải abstraction — nó chỉ là một lớp chuyển tiếp thêm chi phí mà không mang lại gì. Cùng vấn đề với IRepository<T> generic ở bài 13.9.

Chỉ tách những gì Application layer thật sự dùng:

public interface IEventPublisher
{
    Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class;
}

Một phương thức. Nếu sau này cần SendAsync, thêm khi đó — không phải bây giờ.

---

Bài 3 — Fake publisher cho test

Cài FakeEventPublisher thu message vào List và dùng nó để viết một test cho use case, không dựng broker.

Tiêu chí hoàn thành: test chạy dưới 100 ms, và bạn nêu được hai loại test mà fake không thay thế được.

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

Gợi ý. Fake kiểm tra được "use case có publish đúng event không". Nó kiểm tra được gì về broker?

Lời giải:

public sealed class FakeEventPublisher : IEventPublisher
{
    private readonly List<object> _daPublish = [];

    public IReadOnlyList<object> DaPublish => _daPublish.AsReadOnly();

    public Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class
    {
        _daPublish.Add(@event);
        return Task.CompletedTask;
    }

    public Task PublishManyAsync<T>(IEnumerable<T> events, CancellationToken ct = default)
        where T : class
    {
        _daPublish.AddRange(events);
        return Task.CompletedTask;
    }

    public IEnumerable<T> LayTheoKieu<T>() => _daPublish.OfType<T>();
    public void Reset() => _daPublish.Clear();
}

[Fact]
public async Task Chot_lead_phat_dung_mot_event_LeadDaChot()
{
    await using var db = TaoSqliteInMemory();
    var lead = await TaoLeadAsync(db, giaTri: 5_000_000);
    var publisher = new FakeEventPublisher();
    var handler = new ChotLeadHandler(db, publisher, _user, _clock);

    var kq = await handler.Handle(new ChotLeadCommand(lead.Id), CancellationToken.None);

    kq.ThanhCong.Should().BeTrue();

    var events = publisher.LayTheoKieu<LeadDaChotV1>().ToList();
    events.Should().ContainSingle();
    events[0].LeadId.Should().Be(lead.Id.Value);
    events[0].GiaTri.Should().Be(5_000_000);
    events[0].TenantId.Should().Be("acme");
}

Passed!  - Failed: 0, Passed: 1
Thời gian: 43 ms

So sánh với test dùng broker thật:

Fake:                    43 ms
Testcontainers RabbitMQ: 8.400 ms lần đầu (kéo image + khởi động)
                         1.200 ms các lần sau

Chênh lệch 28 lần, và nó quyết định cách bạn làm việc: với 43 ms, bạn chạy test sau mỗi thay đổi; với 1,2 giây × 50 test, bạn chỉ chạy khi CI báo đỏ.

Ba khẳng định mà fake làm tốt:

// 1. Đúng event, đúng số lượng
publisher.LayTheoKieu<LeadDaChotV1>().Should().ContainSingle();

// 2. Đúng nội dung
publisher.LayTheoKieu<LeadDaChotV1>().Single().GiaTri.Should().Be(5_000_000);

// 3. KHÔNG publish khi thao tác thất bại — khẳng định hay bị quên
var kq = await handler.Handle(new ChotLeadCommand(leadIdKhongTonTai), ct);
kq.ThanhCong.Should().BeFalse();
publisher.DaPublish.Should().BeEmpty("không được phát event khi thao tác thất bại");

Khẳng định thứ ba bắt được một lỗi thật: publish trước khi kiểm tra kết quả, dẫn tới việc phát event cho một thao tác đã rollback.

Hai loại test mà fake KHÔNG thay thế được:

Loại 1 — test hợp đồng serialize.

Fake lưu OBJECT trong bộ nhớ.
Broker thật nhận BYTE đã serialize.

Fake KHÔNG phát hiện:
    - Kiểu không serialize được (vòng tham chiếu, interface, kiểu không có constructor rỗng)
    - Trường bị mất khi serialize (private setter không có [JsonInclude])
    - DateTime mất Kind
    - decimal mất độ chính xác
    - Enum serialize thành số trong khi consumer mong đợi chuỗi

[Fact]
public void Integration_event_phai_serialize_va_deserialize_khong_mat_du_lieu()
{
    var goc = new LeadDaChotV1(
        EventId: Guid.CreateVersion7(),
        LeadId: Guid.CreateVersion7(),
        GiaTri: 5_000_000.50m,
        TienTe: "VND",
        ThoiDiemUtc: new DateTime(2026, 9, 25, 8, 14, 22, DateTimeKind.Utc),
        TenantId: "acme");

    var json = JsonSerializer.Serialize(goc);
    var docLai = JsonSerializer.Deserialize<LeadDaChotV1>(json);

    docLai.Should().BeEquivalentTo(goc);
    docLai!.ThoiDiemUtc.Kind.Should().Be(DateTimeKind.Utc);   // hay mất
    docLai.GiaTri.Should().Be(5_000_000.50m);                  // kiểm tra độ chính xác
}

Test này không cần broker nhưng cũng không dùng fake — nó kiểm tra đúng tầng serialize.

Loại 2 — test hành vi của broker và consumer thật.

Fake KHÔNG phát hiện:
    - Binding sai: message publish nhưng không tới queue nào
    - Consumer không được đăng ký
    - Cấu hình retry, DLQ, prefetch sai
    - Hành vi khi message trùng lặp thật sự được giao hai lần
    - Thứ tự khi có nhiều consumer

Dùng MassTransit.TestFramework cho tầng này — nó nằm giữa fake và broker thật:

[Fact]
public async Task Message_publish_phai_toi_dung_consumer()
{
    await using var provider = new ServiceCollection()
        .AddMassTransitTestHarness(x => x.AddConsumer<TaoSubscriptionConsumer>())
        .AddScoped<CrmDbContext>(_ => TaoSqliteInMemory())
        .BuildServiceProvider(true);

    var harness = provider.GetRequiredService<ITestHarness>();
    await harness.Start();

    await harness.Bus.Publish(new LeadConvertedV1(Guid.CreateVersion7(), leadId, customerId, 1000));

    (await harness.Published.Any<LeadConvertedV1>()).Should().BeTrue();
    (await harness.Consumed.Any<LeadConvertedV1>()).Should().BeTrue();

    var consumerHarness = harness.GetConsumerHarness<TaoSubscriptionConsumer>();
    (await consumerHarness.Consumed.Any<LeadConvertedV1>()).Should().BeTrue();
}

Thời gian: ~180 ms — dùng transport in-memory, không cần RabbitMQ

Bốn tầng test, mỗi tầng bắt một loại lỗi:

Tầng	Công cụ	Bắt được	Thời gian
1. Use case	Fake publisher	Logic nghiệp vụ, đúng event	~40 ms
2. Hợp đồng	Serialize trực tiếp	Mất dữ liệu khi serialize	~5 ms
3. Luồng message	TestHarness	Đăng ký, định tuyến, consumer	~180 ms
4. Đầu-cuối	Testcontainers	Cấu hình broker thật, khoá, DLQ	~2.000 ms

Tỷ lệ nên có:

Tầng 1:  ~80% số test    — chạy trong mọi lần build
Tầng 2:  ~10%            — một test cho mỗi integration event
Tầng 3:  ~8%             — một test cho mỗi luồng message
Tầng 4:  ~2%             — chỉ cho những thứ CHỈ database/broker thật kiểm được
                            (READPAST, unique constraint, DLQ)

Đây là hình kim tự tháp quen thuộc, áp cho messaging: nhiều test rẻ ở đáy, ít test đắt ở đỉnh.

Và một cải tiến cho fake — ghi lại thứ tự và thời điểm:

public sealed record MessageDaGhi(object Message, DateTime ThoiDiem, Type Kieu);

public sealed class FakeEventPublisher(TimeProvider clock) : IEventPublisher
{
    private readonly List<MessageDaGhi> _daPublish = [];

    public IReadOnlyList<MessageDaGhi> DaPublish => _daPublish.AsReadOnly();

    public Task PublishAsync<T>(T @event, CancellationToken ct = default) where T : class
    {
        _daPublish.Add(new MessageDaGhi(@event, clock.GetUtcNow().UtcDateTime, typeof(T)));
        return Task.CompletedTask;
    }
}

[Fact]
public async Task Chot_lead_phat_event_theo_dung_thu_tu()
{
    await _handler.Handle(new ChotLeadCommand(leadId), ct);

    publisher.DaPublish.Select(m => m.Kieu).Should().Equal(
        typeof(LeadDaChotV1),
        typeof(DonHangDaTaoV1));
}

Khẳng định về thứ tự hữu ích khi một use case phát nhiều event và thứ tự có ý nghĩa nghiệp vụ — nhưng dùng tiết kiệm, vì nó dễ trở thành test cài đặt (bài 16.9).

Tự kiểm tra

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

Ba thư viện này khác nhau ở điểm nào rõ nhất?

Ở mức trừu tượng và phạm vi tính năng. MassTransit và NServiceBus có abstraction dày với outbox, saga, retry hai tầng và hỗ trợ nhiều broker. EasyNetQ mỏng, chỉ RabbitMQ, không có saga hay outbox, nhưng dùng được ngay trong vài dòng.

Vì sao phải kiểm tra giấy phép trước khi chọn thư viện nền tảng?

Vì mô hình cấp phép của dự án mã nguồn mở có thể thay đổi giữa các phiên bản. Khi đó bạn phải chọn giữa trả tiền, ở lại phiên bản cũ không còn cập nhật, hoặc viết lại lớp hạ tầng. Nên đọc giấy phép của đúng phiên bản định dùng và ước lượng trước chi phí thoát.

Vì sao Application layer không nên phụ thuộc trực tiếp IPublishEndpoint?

Vì khi đó mọi handler đều là một chỗ phải sửa nếu đổi thư viện. Định nghĩa interface IEventPublisher của riêng mình và cài đặt nó ở tầng hạ tầng thì đổi thư viện chỉ phải viết lại một file.

Ngoài khả năng đổi thư viện, tách abstraction còn lợi gì?

Test không cần dựng broker vì có thể dùng fake publisher thu message vào danh sách, và Application không bị ràng buộc vào vòng đời DI của thư viện, thứ hay gây lỗi khó chẩn đoán về scoped service.

Giới hạn của việc bọc thư viện sau interface là gì?

Không nên bọc tính năng nâng cao. Saga state machine không thể đưa sau một interface chung mà vẫn giữ giá trị, nên nếu dùng saga thì chấp nhận phụ thuộc và cô lập nó trong một project riêng.

Khi nào dùng client gốc là đủ, không cần thư viện nào?

Khi chỉ có một broker, chỉ publish vài loại event, không cần saga hay scheduling, và bạn đã tự viết outbox. Chọn thư viện khi cần saga, cần retry và DLQ có sẵn, hoặc muốn đổi broker mà không sửa code nghiệp vụ.

Kết luận

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

1. Kiểm tra giấy phép của đúng phiên bản trước khi cam kết cho dự án dài hạn.
2. Saga là lý do chính đáng nhất để dùng thư viện — phần đó tự viết đúng rất tốn công.
3. Giữ Application layer không biết gì về thư viện. Đó là thứ biến "đổi thư viện" từ dự án lớn thành một file.

Tham khảo

• MassTransit documentation
• NServiceBus Outbox
• EasyNetQ trên GitHub
• Saga pattern

Điều hướng

• Bài trước: 17.7 — 7. Demo tinh thần: Outbox + idempotency
• Bài tiếp theo: 17.9 — Bổ sung (đối chiếu roadmap.sh)
• Về module: Trang mục lục