Outboxパターンとトランザクションログで構築する、回復力のあるイベント駆動型マイクロサービス

はじめに

マイクロサービスの分野では、分散システム全体でシームレスな通信とデータの一貫性を achieve することが常に課題となっています。アプリケーションがより小さく独立したサービスに疎結合化されるにつれて、信頼性の高いイベント伝播の必要性が最重要視されます。サービスがローカルの状態変更を行い、その変更を他のサービスに通知するイベントを発行しようとする際に、一般的な落とし穴が生じます。ローカルトランザクションをコミットしてからイベントを正常に発行するまでの間にサービスがクラッシュした場合はどうなるのでしょうか？データの一貫性の欠如やイベントの損失は、システム全体を急速に麻痺させる可能性があります。この記事では、「Outboxパターン」をデータベーストランザクションログまたはポーリングと組み合わせて、この信頼性のギャップをエレガントに解決し、回復力と堅牢性を備えたイベント駆動型マイクロサービスを促進する方法を掘り下げます。基盤となる原則、実際の実装、そしてこれらの技術がいかに高可用性と一貫性のある分散アーキテクチャの構築において貴重な役割を果たすかを考察します。

信頼性の高いイベント発行の基盤

詳細に入る前に、この議論のバックボーンを形成する重要な用語について共通の理解を確立しましょう。

マイクロサービス: アプリケーションを、疎結合で独立してデプロイ可能なサービスの集まりとして構造化するアーキテクチャスタイル。

イベント駆動型アーキテクチャ (EDA): 疎結合されたソフトウェアコンポーネントが、イベントの非同期発行と購読を通じて相互作用するソフトウェア設計パラダイム。

トランザクションOutboxパターン: ローカルトランザクションと対応するイベントの発行をアトミックに実行することを保証する設計パターン。イベントを直接発行する代わりに、イベントはまずローカル状態変更と同じデータベーストランザクション内の専用「Outbox」テーブルに保存されます。

データベーストランザクションログ (変更データキャプチャ - CDC): データベースに加えられたすべての変更のシーケンシャルレコード。多くの最新データベース（PostgreSQL、MySQL、SQL Serverなど）は、リカバリとレプリケーションのためにこれらのログを維持しています。CDCツールはこれらのログを活用して、プライマリアプリケーションに影響を与えることなくリアルタイムのデータ変更をキャプチャします。

ポーリング: Outboxパターンでは、これは、Outboxテーブルを定期的にクエリして新しい未発行イベントを取得し、それらをメッセージブローカーに発行する別のプロセスを指します。

直接イベント発行の問題点

UserCreatedEventを公開して新しいユーザーを作成する必要があるUserServiceを検討してください。

//Simplified example, not production-ready
@Transactional
public User createUser(User user) {
    userRepository.save(user); // Local database transaction
    eventPublisher.publish(new UserCreatedEvent(user.getId())); // Try to publish event
    return user;
}

システムがuserRepository.save(user)のコミット後、しかしeventPublisher.publish()がメッセージを正常に送信する前にクラッシュした場合、ユーザーはローカルで作成されますが、UserCreatedEventは決して発行されません。このイベントに依存するダウンストリームサービス（例：ウェルカムメールを送信するEmailService）は通知を受け取ることがなく、一貫性のない状態につながります。

Outboxパターンが救世主に

Outboxパターンは、アトミック性を保証することで、この問題をエレガントに解決します。イベントを直接発行するのではなく、イベントの詳細はプライマリビジネスロジックと同じデータベーストランザクション内のOutboxテーブルに保存されます。

Outboxテーブルを使用した実装

Outboxパターンを使用してUserServiceの例を改善しましょう。

まず、Outboxエンティティを定義します。

// Outbox.java
@Entity
@Table(name = "outbox")
public class Outbox {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "aggregatetype", nullable = false)
    private String aggregateType;

    @Column(name = "aggregateid", nullable = false)
    private String aggregateId;

    @Column(name = "eventtype", nullable = false)
    private String eventType;

    @Column(columnDefinition = "jsonb", nullable = false) // Or VARBINARY for other DBs
    private String payload; // JSON representation of the event

    @Column(name = "createdat", nullable = false)
    private Instant createdAt;

    @Column(name = "processedat")
    private Instant processedAt; // To mark events as processed

    // Getters and Setters
}

次に、UserServiceがOutboxを統合します。

// UserService.java
@Service
public class UserService {

    private final UserRepository userRepository;
    private final OutboxRepository outboxRepository;
    private final ObjectMapper objectMapper; // For JSON serialization

    public UserService(UserRepository userRepository, OutboxRepository outboxRepository, ObjectMapper objectMapper) {
        this.userRepository = userRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public User createUser(User user) throws JsonProcessingException {
        // 1. Perform local business logic
        userRepository.save(user);

        // 2. Create the event
        UserCreatedEvent userCreatedEvent = new UserCreatedEvent(user.getId(), user.getEmail());

        // 3. Save the event to the outbox table within the same transaction
        Outbox outboxEntry = new Outbox();
        outboxEntry.setAggregateType("User");
        outboxEntry.setAggregateId(user.getId().toString());
        outboxEntry.setEventType("UserCreatedEvent");
        outboxEntry.setPayload(objectMapper.writeValueAsString(userCreatedEvent));
        outboxEntry.setCreatedAt(Instant.now());
        outboxEntry.setProcessedAt(null); // Not yet processed

        outboxRepository.save(outboxEntry);

        return user;
    }
}

このアプローチにより、トランザクションが正常にコミットされれば、ユーザーとOutboxエントリの両方が永続化されます。トランザクションが失敗した場合は、どちらも永続化されません。これによりアトミック性が保証されます。

イベントの発行：ポーリング対トランザクションログ (CDC)

イベントがOutboxテーブルに入ったら、それらをメッセージブローカー（例：Kafka、RabbitMQ）に信頼性高く発行する必要があります。これには主に2つの戦略があります。

ポーリング: 別の独立したプロセス（多くの場合、スケジューリングされたジョブまたは専用のマイクロサービス）が、outboxテーブルを定期的にクエリして新しい未処理イベントを取得します。

// Example Poller Service (pseudo-code)
@Service
public class OutboxPollerService {

    private final OutboxRepository outboxRepository;
    private final MessageBrokerPublisher messageBrokerPublisher;
    private final ObjectMapper objectMapper;

    public OutboxPollerService(OutboxRepository outboxRepository, MessageBrokerPublisher messageBrokerPublisher, ObjectMapper objectMapper) {
        this.outboxRepository = outboxRepository;
        this.messageBrokerPublisher = messageBrokerPublisher;
        this.objectMapper = objectMapper;
    }

    @Scheduled(fixedRate = 5000) // Poll every 5 seconds
    @Transactional
    public void processOutbox() {
        List<Outbox> unprocessedEvents = outboxRepository.findTop100ByProcessedAtIsNullOrderByCreatedAtAsc(); // Fetch a batch

        for (Outbox event : unprocessedEvents) {
            try {
                // Publish to message broker
                Object eventPayload = objectMapper.readValue(event.getPayload(), Class.forName(event.getEventType()));
                messageBrokerPublisher.publish(event.getEventType(), eventPayload);

                // Mark as processed if publication successful
                event.setProcessedAt(Instant.now());
                outboxRepository.save(event); // Update in the same transaction for safety
            } catch (Exception e) {
                // Log error, retry mechanism might be implemented here
                // Important: Do not mark as processed if publishing fails.
                // The poller will pick it up again on the next run.
            }
        }
    }
}

利点: 実装が比較的簡単で、特別なデータベース機能は不要です。 欠点: 遅延が高くなる可能性があります（ポーリング間隔に依存）、ポーリング間隔が短すぎる場合やスループットが非常に高い場合にデータベースに負荷をかける可能性があります。順序保証の処理に注意が必要です。コンシューマー側での重複排除がしばしば必要になります。

データベーストランザクションログ (変更データキャプチャ - CDC): これはしばしば好まれる、より堅牢な方法です。ポーリングの代わりに、CDCツール（例：Kafka用のDebezium、またはデータベース固有のCDCソリューション）がデータベースのトランザクションログを監視して、outboxテーブルへの変更をキャプチャします。エントリがoutboxに挿入されると、CDCツールはその変更を即座にキャプチャし、メッセージブローカーにメッセージとしてストリームします。

仕組み:
- アプリケーションはOutboxエントリを作成して保存します。
- CDCコネクタ（例：Debezium）がデータベースのトランザクションログ（例：PostgreSQLのWAL、MySQLのbinlog）に接続します。
- コネクタはログを継続的に読み取り、outboxテーブルへの挿入を検出します。
- 各新しいoutboxエントリについて、コネクタはメッセージ（多くの場合、完全な行データを含む）を構築し、Kafkaトピック（または他のメッセージブローカー）に発行します。
- 別の「イベントディスパッチャー」サービス（またはコンシューマー自体）がこのKafkaトピックを購読し、outboxエントリを読み取り、それらをビジネスイベントに変換し、関連するアプリケーション固有のトピックに発行します。その後、outboxエントリは通常、テーブルをクリーンに保つために別の軽量プロセスによって削除または発行済みとしてマークされます。
利点: ほぼリアルタイムのイベント配信、アプリケーションデータベースへの影響が最小限（CDCツールはライブテーブルではなくログから読み取ります）、信頼性の高い順序付け（イベントはログにコミットされた順序でストリームされます）、高いスケーラビリティ。 欠点: 外部CDCツールが必要、セットアップとインフラストラクチャ管理がより複雑、CDCをサポートするデータベースが必要です。

アプリケーションシナリオ

OutboxパターンとポーリングまたはCDCのいずれかを組み合わせることは、次のような状況に最適です。

アトミックなイベント発行: ローカルトランザクションと関連するイベント発行が両方とも成功するか、両方とも失敗することを保証する必要があります。
サービス間通信: サービスは、直接的な結合なしで、状態変更を他のサービスに信頼性高く通信する必要があります。
イベントソーシング (部分的): 完全なイベントソーシングではありませんが、すべての状態変更がイベントによっても表されることを保証する、さらに一歩進んだものです。
コマンドクエリ責任分離 (CQRS) の更新: CQRSアーキテクチャでリードモデルを信頼性高く更新するために使用されます。

結論

定期的なポーリングまたはより洗練されたデータベーストランザクションログ (CDC) の活用によって実装される Outboxパターンは、信頼性の高いイベント駆動型マイクロサービスの構築の礎となります。ローカルトランザクションの一貫性と分散型結果整合性の間のギャップを効果的に橋渡しし、重要なビジネスイベントが決して失われることなく、分散システムの一貫性が維持されることを保証します。これらのパターンを採用することにより、開発者は、コミットされたすべての書き込み操作が対応する外部通知を信頼性高くトリガーする、回復力とスケーラビリティの高いアーキテクチャを自信を持って構築できます。このパターンにより、マイクロサービスは効果的に通信できるようになり、分散ランドスケープ全体でデータ整合性を維持できます。