はじめに
「注文をDBに保存したら、在庫サービスや通知サービスにイベントを送る」という処理は、イベント駆動の構成ではごく普通に出てきます。
ところが、この「DBに書く」と「メッセージを送る」の2つを確実にそろえるのは、思っているよりずっと難しい問題です。
DBへの保存は成功したのにメッセージ送信だけ失敗した、あるいはその逆、ということが現実に起こります。
この問題に対する定番の解決策が、Transactional Outbox(トランザクショナル・アウトボックス)パターンです。
名前は聞いたことがあっても、なぜそれで解決するのか、何を代償にしているのかまで説明できる自信がなかったので、調べた内容を整理しました。
本記事では、問題の正体である「デュアルライト」から始めて、Outboxパターンの仕組み、PostgreSQLとJavaでの実装例、運用で気をつける点までを順に見ていきます。
サンプルコードはJava 17以降とPostgreSQL(JDBCドライバーorg.postgresql:postgresql)を前提としています。
解決したい課題：デュアルライト問題
1つの処理の中で、DBとメッセージブローカー(KafkaやAmazon SQSなど、サービス間でメッセージを中継する仕組み)という2つの異なるシステムに書き込むことを、デュアルライト(二重書き込み)と呼びます。
素直に書くと、次のようなコードになります。
// よくない例：DBのコミットとメッセージ送信が別々に行われる
public void placeOrder(Order order) throws SQLException {
try (Connection conn = dataSource.getConnection()) {
conn.setAutoCommit(false);
insertOrder(conn, order);
conn.commit();
}
// ここでプロセスが落ちたら、注文は保存されたのにイベントは送られない
messageBroker.send("order-events", toJson(order));
}
このコードには、少なくとも2つの失敗パターンがあります。
| タイミング | 起こること | 結果 |
|---|---|---|
| コミット後、送信前にプロセスが停止 | 注文はDBにあるが、イベントは送られない | 在庫サービスは注文を知らないまま |
| 送信を先に行い、その後のコミットが失敗 | イベントは送られたが、注文はDBにない | 存在しない注文に対して出荷が始まる |
順番を入れ替えても、どちらかの失敗パターンが残ります。
DBのトランザクションはDBの中の操作しか束ねられず、メッセージブローカーへの送信はその外側にあるためです。
DBとブローカーをまたいでトランザクションを組む2相コミット(2PC: Two-Phase Commit、複数のシステムに「コミットしてよいか」を確認してから一斉にコミットする仕組み)という方法もあります。
しかし、Richardsonのパターン解説では、DBやブローカーが2PCに対応していないことが多く、サービスを両方に強く結びつけることも望ましくないとして、2PCは選択肢から外されています。
Transactional Outboxパターンの仕組み
Outboxパターンの考え方は単純です。
メッセージを直接ブローカーに送るのをやめ、いったん同じDBの中の「送信待ちテーブル(outbox、送信箱)」に書き込みます。
- 業務データの更新と、outboxテーブルへのメッセージの書き込みを、1つのDBトランザクションで行う
- 別のプロセス(メッセージリレー)がoutboxテーブルを読み、ブローカーへ送信する
- 送信できたメッセージには「送信済み」の印を付ける(または削除する)
ステップ1はDBの中だけで完結するので、注文の保存とイベントの記録は「両方成功する」か「両方なかったことになる」かのどちらかになります。
Richardsonの解説の言葉を借りれば、メッセージはDBトランザクションがコミットされた場合に限って送られることが保証されます。
ブローカーへの送信(ステップ2)は後から行われるため、送信は「すぐ」ではなく「いずれ必ず」になります。
この、少し遅れて全体の整合性が取れる状態を結果整合性(Eventual Consistency)と呼びます。
メッセージリレーの2つの実装方式
outboxテーブルからブローカーへメッセージを運ぶリレーには、大きく分けて2つの方式があります。
| 方式 | 仕組み | 長所 | 短所 |
|---|---|---|---|
| ポーリング(Polling Publisher) | outboxテーブルを定期的にSELECTし、未送信のものを送る |
どのSQLデータベースでも使える。仕組みが単純 | ポーリング間隔の分だけ遅延する。DBに定期的な負荷がかかる。順序を保つのに工夫が要る |
| トランザクションログの追跡(Transaction Log Tailing) | DBの変更ログ(PostgreSQLのWAL、MySQLのbinlogなど)を読み、outboxへのINSERTを検知して送る |
遅延が小さい。アプリ側のポーリング処理が不要 | DB固有の仕組みに依存する。CDCツールの運用が必要 |
トランザクションログを読む方式は、CDC(Change Data Capture、DBの変更をイベントとして取り出す技術)を使って実現するのが一般的です。代表的なツールのDebeziumには、outboxテーブルの変更をKafkaのトピックへ振り分ける「Outbox Event Router」という機能が用意されています。
まずはポーリング方式で仕組みを理解し、遅延やDB負荷が問題になってきたらCDCへの移行を検討する、という順番が取り組みやすいと思います。
本記事の実装例もポーリング方式です。
PostgreSQLとJavaでの実装例
注文サービスが注文を受け付けたら、OrderPlaced(注文確定)イベントを発行する例で実装します。
テーブル定義
CREATE TABLE orders (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
customer_id TEXT NOT NULL,
amount INTEGER NOT NULL,
status TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE outbox (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
event_id UUID NOT NULL UNIQUE, -- 受信側の重複排除に使う
aggregate_type TEXT NOT NULL, -- 例: Order
aggregate_id TEXT NOT NULL, -- 例: 注文ID
event_type TEXT NOT NULL, -- 例: OrderPlaced
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ -- NULLなら未送信
);
-- 未送信の行だけを対象にした部分インデックス
CREATE INDEX outbox_unpublished_idx ON outbox (id) WHERE published_at IS NULL;
aggregate_typeとaggregate_idは、どの業務データ(集約)についてのイベントかを表す列です。
Debeziumのoutboxテーブルの標準的な列構成(aggregatetype、aggregateid、type、payloadなど)も同じ考え方になっています。
送信側：業務データとoutboxを同じトランザクションで書く
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.UUID;
public class OrderService {
private final DataSource dataSource;
public OrderService(DataSource dataSource) {
this.dataSource = dataSource;
}
public long placeOrder(String customerId, int amount) throws SQLException {
try (Connection conn = dataSource.getConnection()) {
conn.setAutoCommit(false);
try {
long orderId = insertOrder(conn, customerId, amount);
// 実務ではJacksonなどのライブラリでJSONを組み立てる
String payload = String.format(
"{\"orderId\": %d, \"customerId\": \"%s\", \"amount\": %d}",
orderId, customerId, amount);
insertOutbox(conn, "Order", String.valueOf(orderId), "OrderPlaced", payload);
conn.commit(); // 注文とイベントが同時に確定する
return orderId;
} catch (SQLException | RuntimeException e) {
conn.rollback(); // どちらか一方だけが残ることはない
throw e;
}
}
}
private long insertOrder(Connection conn, String customerId, int amount) throws SQLException {
String sql = "INSERT INTO orders (customer_id, amount, status) VALUES (?, ?, 'PLACED') RETURNING id";
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setString(1, customerId);
ps.setInt(2, amount);
try (ResultSet rs = ps.executeQuery()) {
rs.next();
return rs.getLong(1);
}
}
}
private void insertOutbox(Connection conn, String aggregateType, String aggregateId,
String eventType, String payload) throws SQLException {
String sql = "INSERT INTO outbox (event_id, aggregate_type, aggregate_id, event_type, payload) "
+ "VALUES (?, ?, ?, ?, CAST(? AS jsonb))";
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setObject(1, UUID.randomUUID());
ps.setString(2, aggregateType);
ps.setString(3, aggregateId);
ps.setString(4, eventType);
ps.setString(5, payload);
ps.executeUpdate();
}
}
}
注目してほしいのは、OrderServiceがメッセージブローカーを一切知らない点です。
やっているのは2つのテーブルへのINSERTだけなので、DBのトランザクションだけで原子性(全部成功するか全部失敗するか)を保てます。
リレー：outboxを読んでブローカーへ送る
import java.util.UUID;
public record OutboxMessage(long id, UUID eventId, String aggregateType,
String aggregateId, String eventType, String payload) {
}
// ブローカーへの送信を抽象化したインターフェース。
// 本番ではKafkaProducerやSQSクライアントを使った実装に差し替える
public interface MessagePublisher {
void publish(OutboxMessage message);
}
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
public class OutboxRelay {
private static final int BATCH_SIZE = 100;
private final DataSource dataSource;
private final MessagePublisher publisher;
public OutboxRelay(DataSource dataSource, MessagePublisher publisher) {
this.dataSource = dataSource;
this.publisher = publisher;
}
// 未送信のメッセージを1バッチ分送信し、送信した件数を返す
public int relayOnce() throws SQLException {
try (Connection conn = dataSource.getConnection()) {
conn.setAutoCommit(false);
try {
List<OutboxMessage> messages = fetchUnpublished(conn);
for (OutboxMessage message : messages) {
publisher.publish(message);
markPublished(conn, message.id());
}
conn.commit();
return messages.size();
} catch (SQLException | RuntimeException e) {
conn.rollback();
throw e;
}
}
}
private List<OutboxMessage> fetchUnpublished(Connection conn) throws SQLException {
// FOR UPDATE SKIP LOCKED：他のリレーが処理中の行は飛ばして取得する
String sql = "SELECT id, event_id, aggregate_type, aggregate_id, event_type, payload::text AS payload "
+ "FROM outbox WHERE published_at IS NULL "
+ "ORDER BY id LIMIT ? FOR UPDATE SKIP LOCKED";
List<OutboxMessage> messages = new ArrayList<>();
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setInt(1, BATCH_SIZE);
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
messages.add(new OutboxMessage(
rs.getLong("id"),
rs.getObject("event_id", UUID.class),
rs.getString("aggregate_type"),
rs.getString("aggregate_id"),
rs.getString("event_type"),
rs.getString("payload")));
}
}
}
return messages;
}
private void markPublished(Connection conn, long id) throws SQLException {
try (PreparedStatement ps = conn.prepareStatement(
"UPDATE outbox SET published_at = now() WHERE id = ?")) {
ps.setLong(1, id);
ps.executeUpdate();
}
}
}
FOR UPDATE SKIP LOCKEDは、他のトランザクションがロックしている行を待たずに飛ばして取得するPostgreSQLの機能です。
公式ドキュメントでも、一般的な用途には向かないが、複数の利用者がキューのようなテーブルを読む場面でロックの競合を避けるのに使える、と説明されています。
リレーを複数台で動かしても、同じ行を二重に処理しにくくなります。
動かしてみる
ブローカーの代わりに、標準出力へ書き出すだけの実装を使って動作を確認します。
import org.postgresql.ds.PGSimpleDataSource;
public class Main {
public static void main(String[] args) throws Exception {
PGSimpleDataSource dataSource = new PGSimpleDataSource();
dataSource.setUrl("jdbc:postgresql://localhost:5432/shop");
dataSource.setUser("app");
dataSource.setPassword("app");
OrderService orderService = new OrderService(dataSource);
orderService.placeOrder("C001", 3200);
orderService.placeOrder("C002", 1500);
MessagePublisher consolePublisher = message ->
System.out.println("[publish] " + message.eventType() + " " + message.payload());
OutboxRelay relay = new OutboxRelay(dataSource, consolePublisher);
int count = relay.relayOnce();
System.out.println(count + "件送信しました");
System.out.println(relay.relayOnce() + "件送信しました");
}
}
テーブルを作成した直後に実行すると、次のように出力されます(payloadはjsonb型に保存したものを文字列で取り出しているため、PostgreSQLが整形した形になります)。
[publish] OrderPlaced {"amount": 3200, "orderId": 1, "customerId": "C001"}
[publish] OrderPlaced {"amount": 1500, "orderId": 2, "customerId": "C002"}
2件送信しました
0件送信しました
2回目のrelayOnce()では、1回目でpublished_atが埋まっているため、送信対象がなくなっています。
実際のリレーは、このrelayOnce()を数百ミリ秒〜数秒おきに繰り返し呼ぶ常駐プロセスとして動かします。
受信側は冪等にする
Outboxパターンを使っても、メッセージが「ちょうど1回」届く保証にはなりません。
上のOutboxRelayで、publisher.publish()が成功した直後、conn.commit()の前にリレーが落ちたとします。
published_atの更新はロールバックされるので、次回のrelayOnce()で同じメッセージがもう一度送られます。
つまりOutboxパターンが保証するのは「少なくとも1回(at-least-once)」の配送です。
Richardsonの解説でも、リレーが同じメッセージを複数回送ることがあるため、受信側は冪等(べきとう:同じ処理を何回実行しても結果が変わらない性質)でなければならない、とされています。
冪等にする典型的な方法は、処理済みのイベントIDを記録しておき、同じIDが届いたら無視することです。
CREATE TABLE processed_messages (
event_id UUID PRIMARY KEY,
processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE shipments (
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
order_payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.UUID;
public class OrderPlacedHandler {
private final DataSource dataSource;
public OrderPlacedHandler(DataSource dataSource) {
this.dataSource = dataSource;
}
public void handle(UUID eventId, String payload) throws SQLException {
try (Connection conn = dataSource.getConnection()) {
conn.setAutoCommit(false);
try {
if (!markAsProcessed(conn, eventId)) {
conn.rollback();
System.out.println("処理済みのため無視: " + eventId);
return;
}
// 処理済みの記録と業務処理を、同じトランザクションで行う
createShipment(conn, payload);
conn.commit();
} catch (SQLException | RuntimeException e) {
conn.rollback();
throw e;
}
}
}
// 初めて届いたイベントならtrue、処理済みならfalse
private boolean markAsProcessed(Connection conn, UUID eventId) throws SQLException {
String sql = "INSERT INTO processed_messages (event_id) VALUES (?) ON CONFLICT (event_id) DO NOTHING";
try (PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setObject(1, eventId);
return ps.executeUpdate() == 1;
}
}
private void createShipment(Connection conn, String payload) throws SQLException {
try (PreparedStatement ps = conn.prepareStatement(
"INSERT INTO shipments (order_payload) VALUES (CAST(? AS jsonb))")) {
ps.setString(1, payload);
ps.executeUpdate();
}
}
}
processed_messagesへの記録と出荷データの登録を同じトランザクションに入れているのがポイントです。
別々にすると、「処理済みと記録したのに出荷データは保存されていない」という、送信側と同じ種類の問題が受信側で起こります。
運用で気をつけること
メッセージの順序
ポーリング方式でORDER BY idを使っても、厳密な順序は保証しにくいのが実情です。
リレーを複数台で動かすと、別々のバッチが並行して送られます。
また、IDの採番順とコミット順は一致しないため、先に採番された行が後からコミットされ、後のポーリングで拾われることもあります。
同じ注文についてのイベント(作成 → 支払い → キャンセルなど)の順序が重要な場合は、リレーを1台にするか、aggregate_idごとに担当を分けるのが現実的です。
Kafkaを使うなら、aggregate_idをメッセージのキーにすると、同じキーのメッセージは同じパーティションに入り、その中での順序が保たれます。
DebeziumのOutbox Event Routerも、標準でaggregateid列をメッセージキーに使います。
outboxテーブルの肥大化
送信済みの行を残し続けると、outboxテーブルはどんどん大きくなります。
調査用に一定期間残し、定期的に削除するのが一般的です。
DELETE FROM outbox
WHERE published_at < now() - interval '7 days';
送信したらすぐ削除する設計もあります。
AWSのサンプル実装は、SQSへの送信に成功した行をその場で削除しています。
送信漏れの監視
リレーが止まっても、送信側のアプリケーションはエラーにならず注文を受け付け続けます。
published_at IS NULLの行の件数や、最も古い未送信行の経過時間を監視しておくと、リレーの停止に早く気づけます。
SELECT count(*) AS unpublished,
now() - min(created_at) AS oldest_age
FROM outbox
WHERE published_at IS NULL;
書き忘れのリスク
Richardsonの解説では、このパターンの欠点として、DBを更新した後にイベントを発行し忘れる、という開発者のミスが起こりやすい点も挙げられています。
業務データの更新とoutboxへの書き込みを1つのメソッドにまとめる、ドメインイベントの仕組みを用意するなど、書き忘れにくい形にしておくと安心です。
関連するパターン
| パターン | Outboxとの関係 |
|---|---|
| Saga | 複数サービスにまたがる処理を、ローカルトランザクションとメッセージの連鎖で実現するパターン。各ステップでイベントを確実に送る必要があるため、Outboxが土台になる |
| Polling Publisher / Transaction Log Tailing | Outboxのメッセージリレーの実装方式 |
| 冪等な受信者(Idempotent Consumer) | at-least-once配送で起こる重複を、受信側で吸収する |
| イベントソーシング | 状態ではなくイベントそのものを保存する設計。イベントの保存が送信の起点になるため、Outboxの代わりになる |
まとめ
| 項目 | 内容 |
|---|---|
| 解決する課題 | DB更新とメッセージ送信を別々に行うと、片方だけが成功する(デュアルライト問題) |
| 仕組み | 業務データとメッセージを同じDBトランザクションでoutboxテーブルに書き、別プロセスのリレーがブローカーへ送る |
| リレーの方式 | ポーリング(単純でどのDBでも使える)と、トランザクションログの追跡(CDC。低遅延だがDB固有) |
| 保証されること | コミットされた場合に限り、メッセージがいずれ送られる |
| 保証されないこと | ちょうど1回の配送。重複がありうるので受信側を冪等にする |
| 運用上の注意 | 順序の保証、outboxテーブルの掃除、未送信メッセージの監視 |
Outboxパターンは、分散した2つのシステムへの書き込みを、1つのDBへの書き込みに置き換える工夫です。
その代わりに、送信の遅延と重複を受け入れることになります。
導入するときは、送信側だけでなく、受信側の冪等性と運用監視までをセットで設計しておく必要があります。
おすすめ書籍
Outboxパターンとその周辺の考え方を深めたい方に、以下の書籍をおすすめします。
- マイクロサービスパターン[実践的システムデザインのためのコード解説] [ Chris Richardson ](Outboxパターンを紹介しているRichardson本人による本。サービス間の通信やSagaの中で、Outboxがどこに位置づくかがわかる)
- データ指向アプリケーションデザイン 信頼性、拡張性、保守性の高い分散システム設計の原理 [ Martin Kleppmann ](トランザクション、レプリケーション、ストリーム処理など、デュアルライト問題の背景にある分散データの基礎を体系的に学べる)
- ドメイン駆動設計をはじめよう ソフトウェアの実装と事業戦略を結びつける実践技法 [ Vlad Khononov ](ドメインイベントや集約の考え方から、境界をまたぐ連携の設計までを学べる)
※上記リンクはアフィリエイトリンクを含みます。
参考
- Pattern: Transactional outbox - microservices.io
- Pattern: Polling publisher - microservices.io
- Pattern: Transaction log tailing - microservices.io
- Transactional outbox pattern - AWS Prescriptive Guidance
- Outbox Event Router - Debezium Documentation
- SELECT (The Locking Clause) - PostgreSQL Documentation