dotnet-messaging-patterns

dotnet-messaging-patterns

.NET のイベント駆動型アーキテクチャ向けの耐久性のあるメッセージングパターンです。Azure Service Bus、RabbitMQ、MassTransit を使用したパブリッシュ／サブスクライブ、競合コンシューマー、デッドレターキュー、サガ／プロセスマネージャーによるオーケストレーション、配信保証の戦略を扱います。

対象外： バックグラウンドサービスのライフサイクルと IHostedService の登録については、[skill:dotnet-background-services] を参照してください。レジリエンスパイプラインと再試行ポリシーについては、[skill:dotnet-resilience] を参照してください。JSON／バイナリのシリアル化設定については、[skill:dotnet-serialization] を参照してください。Channel<T> を使用したプロセス内のプロデューサー／コンシューマーキューについては、[skill:dotnet-channels] を参照してください。

相互参照：メッセージコンシューマーのホスティングについては [skill:dotnet-background-services]、メッセージハンドラーのフォールトトレランスについては [skill:dotnet-resilience]、メッセージエンベロープのシリアル化については [skill:dotnet-serialization]、プロセス内のキューイングパターンについては [skill:dotnet-channels] を参照してください。

---

メッセージングの基本

メッセージの種類

種類	目的	例
コマンド	アクションを要求します（宛先は1つ）	PlaceOrder、ShipPackage
イベント	何かが起きたことを通知します（宛先は複数）	OrderPlaced、PaymentReceived
ドキュメント	システム間でデータを転送します	CustomerProfile、ProductCatalog

コマンドは特定のキューに送信されます。イベントはトピック／エクスチェンジにパブリッシュされ、すべてのサブスクライバーに配信されます。この違いによって、ポイントツーポイントとパブリッシュ／サブスクライブのどちらのトポロジを選ぶかが決まります。

配信保証

保証	動作	実装
最大1回	ファイア・アンド・フォーゲット。メッセージが失われる可能性があります	ACK なし、再試行なし
最低1回	確認されるまでメッセージを再試行します。重複する可能性があります	処理後に ACK を返し、失敗時に再試行
厳密に1回	各メッセージを厳密に1回処理します	最低1回の配信と冪等なコンシューマー

冪等なコンシューマーによる最低1回の配信が、耐久性のあるメッセージングの標準的なアプローチです。真の厳密に1回の配信を実現するには、分散トランザクション（ほとんどのブローカーはサポートしていません）か、コンシューマー側での重複排除が必要です。

---

パブリッシュ／サブスクライブ

Azure Service Bus トピック

// Publisher -- send event to a topic
await using var client = new ServiceBusClient(connectionString);
await using var sender = client.CreateSender("order-events");

var message = new ServiceBusMessage(
    JsonSerializer.SerializeToUtf8Bytes(new OrderPlaced(orderId, total)))
{
    Subject = nameof(OrderPlaced),
    ContentType = "application/json",
    MessageId = Guid.NewGuid().ToString()
};

await sender.SendMessageAsync(message, cancellationToken);

// Subscriber -- process events from a subscription
await using var processor = client.CreateProcessor(
    topicName: "order-events",
    subscriptionName: "billing-service",
    new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = 10,
        AutoCompleteMessages = false
    });

processor.ProcessMessageAsync += async args =>
{
    var body = args.Message.Body.ToObjectFromJson<OrderPlaced>();
    await HandleOrderPlacedAsync(body);
    await args.CompleteMessageAsync(args.Message);
};

processor.ProcessErrorAsync += args =>
{
    logger.LogError(args.Exception, "Error processing message");
    return Task.CompletedTask;
};

await processor.StartProcessingAsync(cancellationToken);

主要なパッケージ：

<PackageReference Include="Azure.Messaging.ServiceBus" Version="7.*" />

RabbitMQ ファンアウトエクスチェンジ

// Publisher -- declare exchange and publish
var factory = new ConnectionFactory { HostName = "localhost" };
await using var connection = await factory.CreateConnectionAsync();
await using var channel = await connection.CreateChannelAsync();

await channel.ExchangeDeclareAsync(
    exchange: "order-events",
    type: ExchangeType.Fanout,
    durable: true);

var body = JsonSerializer.SerializeToUtf8Bytes(
    new OrderPlaced(orderId, total));

await channel.BasicPublishAsync(
    exchange: "order-events",
    routingKey: string.Empty,
    body: body);

主要なパッケージ：

<PackageReference Include="RabbitMQ.Client" Version="7.*" />

MassTransit によるパブリッシュ

MassTransit はブローカーを抽象化し、Azure Service Bus、RabbitMQ、Amazon SQS、インメモリトランスポート向けの統一された API を提供します。

// Registration
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });
        cfg.ConfigureEndpoints(context);
    });
});

// Publisher
public sealed class OrderService(IPublishEndpoint publishEndpoint)
{
    public async Task PlaceOrderAsync(
        Guid orderId, decimal total, CancellationToken ct)
    {
        // Process order...
        await publishEndpoint.Publish(
            new OrderPlaced(orderId, total), ct);
    }
}

// Consumer
public sealed class OrderPlacedConsumer(
    ILogger<OrderPlacedConsumer> logger)
    : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        logger.LogInformation(
            "Processing order {OrderId}", context.Message.OrderId);
        await ProcessAsync(context.Message);
    }
}

// Message contract (use records in a shared contracts assembly)
public record OrderPlaced(Guid OrderId, decimal Total);

主要なパッケージ：

<PackageReference Include="MassTransit" Version="8.*" />

<PackageReference Include="MassTransit.RabbitMQ" Version="8.*" />

<PackageReference Include="MassTransit.Azure.ServiceBus.Core" Version="8.*" />

---

競合コンシューマー

複数のコンシューマーインスタンスが、同じキューのメッセージを並行して処理します。ブローカーは各メッセージを1つのコンシューマーだけに配信し、インスタンス間で負荷を分散します。

パターン

Queue: order-processing
  ├── Consumer Instance A  (picks message 1)
  ├── Consumer Instance B  (picks message 2)
  └── Consumer Instance C  (picks message 3)

Azure Service Bus — コンシューマーのスケーリング

// Multiple instances reading from the same queue automatically compete.
// MaxConcurrentCalls controls per-instance parallelism.
var processor = client.CreateProcessor("order-processing",
    new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls = 20,
        PrefetchCount = 50,
        AutoCompleteMessages = false
    });

MassTransit — 同時実行数の制限

x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseConcurrentMessageLimit(10);
});

順序に関する考慮事項

競合コンシューマーでは、スループットのために厳密な順序が犠牲になります。順序が重要な場合：

• Azure Service Bus：セッション（RequiresSession = true）を使用して、セッション ID 内（顧客ごとなど）の FIFO を保証します
• RabbitMQ：キューごとにコンシューマーを1つだけ使用するか、コンシステントハッシュエクスチェンジを使用してキーごとにパーティション分割します
• MassTransit：キーに基づく順序付けには UseMessagePartitioner を構成します

---

デッドレターキュー

デッドレターキュー（DLQ）は、再試行を使い切っても処理できないメッセージを受け取ります。これにより、処理不能なメッセージがメインキューを塞ぐのを防ぎます。

メッセージがデッドレターになる理由

理由	発生条件
最大配信試行回数を超過	メッセージの処理に N 回失敗した
TTL の期限切れ	メッセージが有効期限を過ぎてもキューに残っていた
コンシューマーによる拒否	コンシューマーが明示的にメッセージをデッドレターにした
キュー長の超過	キューのオーバーフローポリシーによって DLQ にルーティングされた

Azure Service Bus DLQ

// Dead-letter a message with reason
await args.DeadLetterMessageAsync(
    args.Message,
    deadLetterReason: "ValidationFailed",
    deadLetterErrorDescription: "Missing required field: CustomerId");

// Read from the dead-letter sub-queue
await using var dlqReceiver = client.CreateReceiver(
    "order-processing",
    new ServiceBusReceiverOptions
    {
        SubQueue = SubQueue.DeadLetter
    });

while (true)
{
    var message = await dlqReceiver.ReceiveMessageAsync(
        TimeSpan.FromSeconds(5), cancellationToken);
    if (message is null) break;

    logger.LogWarning(
        "DLQ message: {Reason} - {Description}",
        message.DeadLetterReason,
        message.DeadLetterErrorDescription);

    // Inspect, fix, and re-submit or discard
    await dlqReceiver.CompleteMessageAsync(message);
}

MassTransit のエラー／フォールトキュー

MassTransit は _error キューと _skipped キューを自動的に作成します。再試行を使い切った後に失敗したメッセージは、フォールトのメタデータとともにエラーキューに移動されます。

// Configure retry before dead-lettering
x.AddConsumer<OrderProcessor>(cfg =>
{
    cfg.UseMessageRetry(r => r.Intervals(
        TimeSpan.FromSeconds(1),
        TimeSpan.FromSeconds(5),
        TimeSpan.FromSeconds(15)));
});

DLQ の監視

DLQ の深さを常に監視し、アラートを設定してください。監視されていない DLQ は、データが失われたり古くなったりするまで、気付かれないまま増え続けます。

---

Saga／プロセスマネージャー

Saga は、サービス間にまたがる複数ステップのビジネスプロセスを調整します。各ステップでイベントを発行し、次のステップをトリガーします。また、失敗時の補償ロジックも備えています。

コレオグラフィとオーケストレーションの比較

スタイル	仕組み	使用する状況
コレオグラフィ	各サービスが独立してイベントに反応し、中央の調整役がいません	ステップ数が少なく、結合度の低いシンプルなフロー
オーケストレーション	Saga／プロセスマネージャーが各ステップを指揮します	複雑なフロー、補償が必要な場合、可視性が求められる場合

MassTransit のステートマシン Saga

// Saga state
public class OrderState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = default!;
    public Guid OrderId { get; set; }
    public decimal Total { get; set; }
    public DateTime? PaymentReceivedAt { get; set; }
}

// State machine definition
public sealed class OrderStateMachine : MassTransitStateMachine<OrderState>
{
    public State Submitted { get; private set; } = default!;
    public State PaymentPending { get; private set; } = default!;
    public State Completed { get; private set; } = default!;
    public State Faulted { get; private set; } = default!;

    public Event<OrderSubmitted> OrderSubmitted { get; private set; } = default!;
    public Event<PaymentReceived> PaymentReceived { get; private set; } = default!;
    public Event<PaymentFailed> PaymentFailed { get; private set; } = default!;

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

        Event(() => OrderSubmitted,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentReceived,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentFailed,
            x => x.CorrelateById(ctx => ctx.Message.OrderId));

        Initially(
            When(OrderSubmitted)
                .Then(ctx =>
                {
                    ctx.Saga.OrderId = ctx.Message.OrderId;
                    ctx.Saga.Total = ctx.Message.Total;
                })
                .Publish(ctx => new RequestPayment(
                    ctx.Saga.OrderId, ctx.Saga.Total))
                .TransitionTo(PaymentPending));

        During(PaymentPending,
            When(PaymentReceived)
                .Then(ctx =>
                    ctx.Saga.PaymentReceivedAt = DateTime.UtcNow)
                .Publish(ctx => new FulfillOrder(ctx.Saga.OrderId))
                .TransitionTo(Completed),
            When(PaymentFailed)
                .Publish(ctx => new CancelOrder(ctx.Saga.OrderId))
                .TransitionTo(Faulted));
    }
}

// Registration -- requires MassTransit.EntityFrameworkCore package for EF persistence
// NuGet: MassTransit.EntityFrameworkCore Version="8.*"
builder.Services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .EntityFrameworkRepository(r =>
        {
            r.ExistingDbContext<SagaDbContext>();
            r.UsePostgres();
        });

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

Saga の永続化

ストア	パッケージ	使用する状況
Entity Framework Core	MassTransit.EntityFrameworkCore	すでに EF Core を使用しており、トランザクションが必要な場合
MongoDB	MassTransit.MongoDb	ドキュメント指向の状態管理や高スループットが必要な場合
Redis	MassTransit.Redis	一時的な Saga や低レイテンシが必要な場合
インメモリ	組み込み	テスト専用 — 再起動すると状態が失われます

補償パターン

Saga のステップが失敗した場合は、先行するステップを取り消す補償コマンドを発行します。

OrderSubmitted -> RequestPayment -> PaymentReceived -> ReserveInventory
                                                          |
                                                     InventoryFailed
                                                          |
                                                    RefundPayment (compensation)
                                                          |
                                                    CancelOrder (compensation)

---

冪等性のあるコンシューマー

少なくとも 1 回の配信では、コンシューマーが同じメッセージを複数回受信することがあります。冪等性のあるコンシューマーを使うと、処理を繰り返しても同じ結果が得られます。

データベースを利用した重複排除

public sealed class IdempotentOrderConsumer(
    AppDbContext db,
    ILogger<IdempotentOrderConsumer> logger)
    : IConsumer<OrderPlaced>
{
    public async Task Consume(ConsumeContext<OrderPlaced> context)
    {
        var messageId = context.MessageId
            ?? throw new InvalidOperationException("Missing MessageId");

        // Check if already processed
        var exists = await db.ProcessedMessages
            .AnyAsync(m => m.MessageId == messageId);

        if (exists)
        {
            logger.LogInformation(
                "Duplicate message {MessageId}, skipping", messageId);
            return;
        }

        // Process the message
        await ProcessOrderAsync(context.Message);

        // Record as processed
        db.ProcessedMessages.Add(new ProcessedMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow,
            ConsumerType = nameof(IdempotentOrderConsumer)
        });

        await db.SaveChangesAsync();
    }
}

自然な冪等性

自然に冪等性を持つ操作を優先してください。

• アップサート（INSERT ... ON CONFLICT UPDATE）を使い、無条件の挿入を避けます
• 条件付き更新（UPDATE ... WHERE Status = 'Pending'）を使い、無条件の更新を避けます
• 自動生成 ID ではなく、メッセージの内容から導出した決定的な IDを使用します

---

メッセージエンベロープパターン

メッセージのペイロードを標準的なエンベロープで包み、トレース、バージョン管理、ルーティング用のメタデータを付与します。

public sealed record MessageEnvelope<T>(
    string MessageId,
    string MessageType,
    DateTimeOffset Timestamp,
    string CorrelationId,
    string Source,
    int Version, // Schema version for backward-compatible deserialization
    T Payload);

MassTransit では、ConsumeContext（MessageId、CorrelationId、Headers）を介してこれが自動的に提供されます。ブローカーのクライアントを直接使用する場合は、エンベロープを明示的に実装してください。

---

エージェントが注意すべき点

1. Azure Service Bus では自動完了を使用しないでください — AutoCompleteMessages = false に設定し、処理が正常に完了した後で CompleteMessageAsync を呼び出します。自動完了では処理の終了前に確認応答が行われるため、失敗時にデータが失われるおそれがあります。
2. 不正なメッセージへの対処を忘れないでください — 最大配信回数を必ず設定し、DLQ を監視してください。これらを設定しないと、不正なメッセージ 1 件によってキュー全体が無期限にブロックされるおそれがあります。
3. 本番環境ではインメモリの Saga 永続化を使用しないでください — 再起動すると Saga の状態が失われ、ビジネスプロセスの状態が不明になります。Entity Framework、MongoDB、または Redis に永続化してください。
4. パーティション間でメッセージの順序が保たれると想定しないでください — 競合するコンシューマーやトピックサブスクリプションでは、既定で順不同にメッセージが配信されます。順序が重要な場合は、セッションまたはパーティショニングを使用してください。
5. 少なくとも 1 回の配信を行うコンシューマーで、冪等性を省略しないでください — タイムアウト、ネットワークの不具合、コンシューマーの再起動により、ブローカーがメッセージを再配信することがあります。すべてのコンシューマーで重複メッセージを安全に処理できるようにしてください。
6. 接続文字列をハードコードしないでください — 環境変数または Azure Key Vault の参照を使用してください。ローカル開発では、ユーザーシークレットまたはソース管理の対象外にした .env ファイルを使用してください。

---

参考資料

• Azure Service Bus のドキュメント
• .NET 用 Azure Service Bus クライアントライブラリ
• RabbitMQ .NET クライアントのドキュメント
• MassTransit のドキュメント
• MassTransit の Saga
• エンタープライズ統合パターン