CQRSとは｜イベントソーシングとの違いをSQLとコードで解説

CQRS（Command Query Responsibility Segregation）とは、データを更新するためのモデルと、読み取るためのモデルを分ける設計のことです。 更新は業務ルールを守りやすい形で、読み取りは画面に出しやすい形で、それぞれ別に作ります。

よく一緒に語られるイベントソーシングは別の考え方で、片方だけでも採用できます。この記事では、注文データを題材に分け方を4段階に並べ、各段階のCREATE TABLEとPythonコードをPostgreSQLで実行した結果を載せます。読み取り側への反映が遅れて更新前の値が見える現象も再現し、対処をコードで示します。

確認環境：Ubuntu 24.04 / PostgreSQL 16.15 / Python 3.11.15（psycopg2 2.9）。確認日 2026-10-06。出力例のデータベース名は appdb・appdb_read に置き換えています。

この記事の結論

• CQRSは「読み書きのモデルを分ける」こと、イベントソーシングは「状態ではなく出来事を保存する」ことで、別の話です。 片方だけ採用できます。
• CQRSには段階があります。同じDB・同じテーブルのまま、コードの中で読み取り用と書き込み用を分けるだけでもCQRSです。DBを分けることは必須ではありません。
• 読み取り用テーブルを別のタイミングで更新する段階から、反映の遅れが生まれます。対処は「書き込み直後だけ書き込み側を読む」「版番号で追いつくまで待つ」の2つが基本です。
• イベントソーシングは、履歴・過去時点の状態・読み取り用テーブルの作り直しが得られる代わりに、仕組みが増えます。必要になった集約（まとまり）にだけ使うのが現実的な入り方です。

CQRSの4段階の早見表

段階4だけがイベントソーシングで、段階1〜3は現在の状態をそのまま保存する形です。逆に、イベントソーシングを読み取り用テーブルなしで使うこともできます（状態を毎回イベントから組み立てる）。

CQRSとイベントソーシングの違い

Martin Fowler 氏の解説は、CQRSを「情報の更新に使うモデルと、読み取りに使うモデルを別にできるという考え方」と説明し、イベントソーシングを「アプリケーションの状態へのすべての変更を、イベントの列として記録する」と定義しています（記事末尾の参考・出典）。

準備：接続の共通コード

# conn.py … 接続先。自分の環境に合わせて書き換える
import psycopg2

WRITE_DSN = "host=127.0.0.1 dbname=appdb user=app password=secret"
READ_DSN = "host=127.0.0.1 dbname=appdb_read user=app password=secret"

def connect(dsn=WRITE_DSN):
    return psycopg2.connect(dsn)

段階1：同一DBでモデルだけ分ける

テーブルは正規化した形のままです。

-- 段階1：テーブルは今までどおり。コードの中で「書き込み用」と「読み取り用」を分ける
DROP TABLE IF EXISTS order_items, orders, order_summaries, outbox CASCADE;
CREATE TABLE orders (
  id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_name TEXT NOT NULL,
  status        TEXT NOT NULL DEFAULT 'draft' CHECK (status IN ('draft', 'confirmed', 'cancelled')),
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE order_items (
  order_id   BIGINT NOT NULL REFERENCES orders (id),
  product    TEXT NOT NULL,
  unit_price INT NOT NULL CHECK (unit_price >= 0),
  quantity   INT NOT NULL CHECK (quantity > 0),
  PRIMARY KEY (order_id, product)
);

コードを「更新する関数」と「読み取る関数」に分けます。更新側は業務ルールを確かめ、読み取り側は画面の形に合わせたSQLを直接書きます。

# ---------- 書き込みモデル（3段階で共通）：業務ルールを守って正規化されたテーブルを更新する ----------
def place_order(cur, customer_name, items):
    if not items:
        raise ValueError("明細が1件以上必要です")
    cur.execute("INSERT INTO orders (customer_name) VALUES (%s) RETURNING id", (customer_name,))
    order_id = cur.fetchone()[0]
    cur.executemany("INSERT INTO order_items VALUES (%s, %s, %s, %s)",
                    [(order_id, p, price, qty) for p, price, qty in items])
    return order_id

def confirm_order(cur, order_id):
    cur.execute("UPDATE orders SET status = 'confirmed' WHERE id = %s AND status = 'draft'", (order_id,))
    if cur.rowcount == 0:
        raise ValueError("確定できるのは下書きの注文だけです")

# ---------- 読み取りモデル ----------
SUMMARY_SQL = """
SELECT o.id, o.customer_name, o.status, COUNT(i.product) AS item_count,
       COALESCE(SUM(i.unit_price * i.quantity), 0) AS total_amount
FROM orders o LEFT JOIN order_items i ON i.order_id = o.id
{where}
GROUP BY o.id ORDER BY o.id"""

def list_summaries_by_join(cur):                 # 段階1：毎回 JOIN と集計で作る
    cur.execute(SUMMARY_SQL.format(where=""))
    return cur.fetchall()

def list_summaries_from_table(cur):              # 段階2・3：出来上がった行を読むだけ
    cur.execute("SELECT order_id, customer_name, status, item_count, total_amount FROM order_summaries ORDER BY order_id")
    return cur.fetchall()

読み取り側は、更新用のクラスやルールを経由しません。 一覧画面が必要とする列だけを、JOINと集計で直接取り出します。これだけで「更新のためのモデルを変えずに、読み取りのSQLを自由に調整できる」状態になります。

段階2：読み取り用テーブルを別に持つ

一覧のたびにJOINと集計をするのが重くなってきたら、出来上がった形の行を持つテーブルを足します。

-- 段階2：同じ DB の中に、一覧画面の形に合わせた読み取り用テーブルを足す
CREATE TABLE order_summaries (
  order_id      BIGINT PRIMARY KEY,
  customer_name TEXT NOT NULL,
  status        TEXT NOT NULL,
  item_count    INT NOT NULL,
  total_amount  INT NOT NULL,
  version       INT NOT NULL
);

更新のたびに、同じトランザクションの中で読み取り用の行を作り直します。

# ---------- 段階2：同じトランザクションの中で読み取り用テーブルを更新する ----------
def refresh_summary(cur, order_id):
    cur.execute("""
INSERT INTO order_summaries (order_id, customer_name, status, item_count, total_amount, version)
SELECT o.id, o.customer_name, o.status, COUNT(i.product),
       COALESCE(SUM(i.unit_price * i.quantity), 0), 1
FROM orders o LEFT JOIN order_items i ON i.order_id = o.id
WHERE o.id = %s GROUP BY o.id
ON CONFLICT (order_id) DO UPDATE
SET customer_name = EXCLUDED.customer_name, status = EXCLUDED.status,
    item_count = EXCLUDED.item_count, total_amount = EXCLUDED.total_amount,
    version = order_summaries.version + 1""", (order_id,))

同じトランザクションなので、反映の遅れはありません。その代わり、更新のたびに集計が走るぶん書き込みが少し重くなります。

段階3：DBも分ける

読み取りの量が書き込みよりずっと多いときや、読み取り側を検索向けの別製品に置きたいときは、DBを分けます。2つのDBを1つのトランザクションでは更新できないので、書き込み側のDBに「運ぶべき変更」を残すテーブル（outbox）を置きます。

-- 段階3（書き込み側の DB）：変更内容を、業務データと同じトランザクションで outbox に残す
CREATE TABLE outbox (
  id         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  order_id   BIGINT NOT NULL,
  payload    JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  sent_at    TIMESTAMPTZ
);

-- 段階3（読み取り側の DB）：読み取り用テーブルだけを置く
DROP TABLE IF EXISTS order_summaries;
CREATE TABLE order_summaries (
  order_id      BIGINT PRIMARY KEY,
  customer_name TEXT NOT NULL,
  status        TEXT NOT NULL,
  item_count    INT NOT NULL,
  total_amount  INT NOT NULL,
  version       BIGINT NOT NULL
);

# ---------- 段階3：outbox に書き、別プロセス（ここでは関数）が読み取り側の DB へ運ぶ ----------
def write_outbox(cur, order_id):
    cur.execute(SUMMARY_SQL.format(where="WHERE o.id = %s"), (order_id,))
    oid, name, status, count, total = cur.fetchone()
    cur.execute("INSERT INTO outbox (order_id, payload) VALUES (%s, %s)",
                (oid, json.dumps({"customer_name": name, "status": status,
                                  "item_count": count, "total_amount": int(total)}, ensure_ascii=False)))

def relay(wcon, rcon):
    """まだ送っていない outbox の行を、id の順に読み取り側へ反映する。"""
    with wcon.cursor() as w, rcon.cursor() as r:
        w.execute("SELECT id, order_id, payload FROM outbox WHERE sent_at IS NULL ORDER BY id FOR UPDATE SKIP LOCKED")
        rows = w.fetchall()
        for outbox_id, order_id, p in rows:
            r.execute("""
INSERT INTO order_summaries (order_id, customer_name, status, item_count, total_amount, version)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (order_id) DO UPDATE
SET customer_name = EXCLUDED.customer_name, status = EXCLUDED.status,
    item_count = EXCLUDED.item_count, total_amount = EXCLUDED.total_amount, version = EXCLUDED.version
WHERE order_summaries.version < EXCLUDED.version""",
                      (order_id, p["customer_name"], p["status"], p["item_count"], p["total_amount"], outbox_id))
            w.execute("UPDATE outbox SET sent_at = now() WHERE id = %s", (outbox_id,))
        rcon.commit(); wcon.commit()                 # 読み取り側を先に確定する
        return len(rows)

relay は、同じ行が2回届いても結果が変わらないよう、版番号（ここでは outbox の id）が大きいときだけ上書きしています。

段階1〜3の実行結果

if __name__ == "__main__":
    wcon = connect(WRITE_DSN); rcon = connect(READ_DSN)
    ITEMS = [("ノート", 300, 2), ("ペン", 120, 5)]

    print("== 段階1：同じテーブルを、書き込み用の関数と読み取り用の関数で使い分ける ==")
    run_file(wcon, "stage1.sql")
    with wcon.cursor() as cur:
        oid = place_order(cur, "青木", ITEMS); confirm_order(cur, oid)
        place_order(cur, "井上", [("付箋", 200, 1)])
        wcon.commit()
        print(list_summaries_by_join(cur))

    print("== 段階2：同じ DB の読み取り用テーブルを、同じトランザクションで更新する ==")
    run_file(wcon, "stage2.sql")
    with wcon.cursor() as cur:
        oid = place_order(cur, "上田", ITEMS); refresh_summary(cur, oid)
        wcon.commit()
        confirm_order(cur, oid); refresh_summary(cur, oid)
        wcon.commit()
        print(list_summaries_from_table(cur))
        cur.execute("SELECT version FROM order_summaries WHERE order_id = %s", (oid,)); print("version:", cur.fetchone()[0])

    print("== 段階3：読み取り用テーブルを別の DB に置き、outbox 経由で運ぶ ==")
    run_file(wcon, "stage3_write.sql"); run_file(rcon, "stage3_read.sql")
    with wcon.cursor() as cur, rcon.cursor() as rcur:
        oid = place_order(cur, "江口", ITEMS); write_outbox(cur, oid)
        wcon.commit()
        print("書き込み直後の読み取り側:", list_summaries_from_table(rcur)); rcon.commit()
        print("運んだ件数:", relay(wcon, rcon))
        print("中継後の読み取り側:", list_summaries_from_table(rcur)); rcon.commit()
        confirm_order(cur, oid); write_outbox(cur, oid); wcon.commit()
        print("運んだ件数:", relay(wcon, rcon))
        print("確定後の読み取り側:", list_summaries_from_table(rcur)); rcon.commit()
        print("もう一度中継:", relay(wcon, rcon), "件")

== 段階1：同じテーブルを、書き込み用の関数と読み取り用の関数で使い分ける ==
[(1, '青木', 'confirmed', 2, 1200), (2, '井上', 'draft', 1, 200)]
== 段階2：同じ DB の読み取り用テーブルを、同じトランザクションで更新する ==
[(3, '上田', 'confirmed', 2, 1200)]
version: 2
== 段階3：読み取り用テーブルを別の DB に置き、outbox 経由で運ぶ ==
書き込み直後の読み取り側: []
運んだ件数: 1
中継後の読み取り側: [(4, '江口', 'draft', 2, 1200)]
運んだ件数: 1
確定後の読み取り側: [(4, '江口', 'confirmed', 2, 1200)]
もう一度中継: 0 件

段階3の「書き込み直後の読み取り側」が空になっています。書き込みは終わっているのに、読み取り側にはまだ届いていない。これが反映の遅れです。

段階4：イベント保存まで行う（イベントソーシング）

ここからは保存の考え方が変わります。orders のような状態のテーブルを持たず、「注文された」「明細が追加された」といった出来事を追記するだけにします。

-- 段階4：状態のテーブルを持たず、出来事（イベント）だけを追記する
DROP TABLE IF EXISTS events, order_summaries, projector_offsets CASCADE;
CREATE TABLE events (
  seq         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,  -- 全体の通し番号（読み取り側が追いかける位置）
  stream_id   TEXT NOT NULL,                                    -- 例: order-1
  version     INT NOT NULL,                                     -- その注文の中での版番号 1,2,3…
  type        TEXT NOT NULL,
  payload     JSONB NOT NULL,
  occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  CONSTRAINT uq_events_stream_version UNIQUE (stream_id, version)
);
-- 読み取り用テーブル（イベントから組み立てる。消しても作り直せる）
CREATE TABLE order_summaries (
  order_id      TEXT PRIMARY KEY,
  customer_name TEXT NOT NULL,
  status        TEXT NOT NULL,
  item_count    INT NOT NULL,
  total_amount  INT NOT NULL,
  version       INT NOT NULL
);
-- 読み取り側がどこまで反映したかの記録
CREATE TABLE projector_offsets (
  name     TEXT PRIMARY KEY,
  last_seq BIGINT NOT NULL
);
INSERT INTO projector_offsets VALUES ('order_summaries', 0);

UNIQUE (stream_id, version) が要です。同じ注文の同じ版に2件は入らないので、2人が同時に更新しようとしたとき、後から来た側がエラーで止まります。

イベントから現在の状態と読み取り用テーブルを組み立てるコード

# es.py … イベントを保存し、現在の状態と読み取り用テーブルを組み立てる
import json, threading, time
from dataclasses import dataclass, field
import psycopg2
from conn import connect

# ---------- 書き込みモデル：イベントを順に当てはめて現在の状態を作る ----------
@dataclass
class Order:
    customer_name: str = ""
    status: str = "none"
    items: dict = field(default_factory=dict)      # product -> (unit_price, quantity)
    version: int = 0

    @property
    def total_amount(self):
        return sum(price * qty for price, qty in self.items.values())

def apply(order: Order, type_: str, p: dict) -> Order:
    if type_ == "OrderPlaced":
        order.customer_name, order.status = p["customer_name"], "draft"
    elif type_ == "ItemAdded":
        order.items[p["product"]] = (p["unit_price"], p["quantity"])
    elif type_ == "ItemRemoved":
        del order.items[p["product"]]
    elif type_ == "OrderConfirmed":
        order.status = "confirmed"
    order.version += 1
    return order

def load(cur, stream_id, up_to_version=None) -> Order:
    cur.execute("SELECT type, payload FROM events WHERE stream_id = %s AND version <= %s ORDER BY version",
                (stream_id, up_to_version or 2**31 - 1))
    order = Order()
    for type_, payload in cur.fetchall():
        apply(order, type_, payload)
    return order

def append(cur, stream_id, expected_version, type_, payload):
    """expected_version の次の版としてイベントを1件追記する。"""
    cur.execute("INSERT INTO events (stream_id, version, type, payload) VALUES (%s, %s, %s, %s)",
                (stream_id, expected_version + 1, type_, json.dumps(payload, ensure_ascii=False)))
    return expected_version + 1

# ---------- コマンド：現在の状態で業務ルールを確かめてから、イベントを追記する ----------
def place_order(con, stream_id, customer_name):
    with con.cursor() as cur:
        v = append(cur, stream_id, 0, "OrderPlaced", {"customer_name": customer_name})
    con.commit(); return v

def add_item(con, stream_id, product, unit_price, quantity):
    with con.cursor() as cur:
        order = load(cur, stream_id)
        if order.status != "draft":
            raise ValueError("明細を変えられるのは下書きの注文だけです")
        v = append(cur, stream_id, order.version, "ItemAdded",
                   {"product": product, "unit_price": unit_price, "quantity": quantity})
    con.commit(); return v

def remove_item(con, stream_id, product):
    with con.cursor() as cur:
        order = load(cur, stream_id)
        if product not in order.items:
            raise ValueError("その明細はありません")
        v = append(cur, stream_id, order.version, "ItemRemoved", {"product": product})
    con.commit(); return v

def confirm_order(con, stream_id):
    with con.cursor() as cur:
        order = load(cur, stream_id)
        if order.status != "draft" or not order.items:
            raise ValueError("確定できるのは明細のある下書きの注文だけです")
        v = append(cur, stream_id, order.version, "OrderConfirmed", {})
    con.commit(); return v

# ---------- 読み取りモデル：イベントを通し番号の順に読み、読み取り用テーブルを更新する ----------
def project_once(con, batch=100) -> int:
    with con.cursor() as cur:
        cur.execute("SELECT last_seq FROM projector_offsets WHERE name = 'order_summaries' FOR UPDATE")
        last_seq = cur.fetchone()[0]
        cur.execute("SELECT seq, stream_id, version, type, payload FROM events WHERE seq > %s ORDER BY seq LIMIT %s",
                    (last_seq, batch))
        rows = cur.fetchall()
        for seq, stream_id, version, type_, p in rows:
            if type_ == "OrderPlaced":
                cur.execute("INSERT INTO order_summaries VALUES (%s, %s, 'draft', 0, 0, %s)",
                            (stream_id, p["customer_name"], version))
            elif type_ == "ItemAdded":
                cur.execute("UPDATE order_summaries SET item_count = item_count + 1,"
                            " total_amount = total_amount + %s, version = %s WHERE order_id = %s",
                            (p["unit_price"] * p["quantity"], version, stream_id))
            elif type_ == "ItemRemoved":
                removed = load(cur, stream_id, version - 1).items[p["product"]]
                cur.execute("UPDATE order_summaries SET item_count = item_count - 1,"
                            " total_amount = total_amount - %s, version = %s WHERE order_id = %s",
                            (removed[0] * removed[1], version, stream_id))
            elif type_ == "OrderConfirmed":
                cur.execute("UPDATE order_summaries SET status = 'confirmed', version = %s WHERE order_id = %s",
                            (version, stream_id))
            last_seq = seq
        cur.execute("UPDATE projector_offsets SET last_seq = %s WHERE name = 'order_summaries'", (last_seq,))
    con.commit()                                    # テーブルの更新と位置の記録を同時に確定する
    return len(rows)

def get_summary(con, stream_id):
    with con.cursor() as cur:
        cur.execute("SELECT status, item_count, total_amount, version FROM order_summaries WHERE order_id = %s", (stream_id,))
        row = cur.fetchone()
    con.commit(); return row

# ---------- 反映遅れへの対処 ----------
def get_summary_from_write_side(con, stream_id):
    """対処1：書き込み直後だけ、イベントから組み立てた状態を返す。"""
    with con.cursor() as cur:
        o = load(cur, stream_id)
    con.commit(); return (o.status, len(o.items), o.total_amount, o.version)

def get_summary_at_least(con, stream_id, min_version, timeout=2.0, interval=0.05):
    """対処2：読み取り側が min_version 以上になるまで待つ。間に合わなければ None。"""
    deadline = time.monotonic() + timeout
    while True:
        row = get_summary(con, stream_id)
        if row and row[3] >= min_version:
            return row
        if time.monotonic() >= deadline:
            return None
        time.sleep(interval)

実行する

if __name__ == "__main__":
    con = connect()
    with con.cursor() as cur:
        cur.execute(open("stage4.sql", encoding="utf-8").read())
    con.commit()

    print("== 1. イベントを追記する ==")
    place_order(con, "order-1", "青木")
    add_item(con, "order-1", "ノート", 300, 2)
    add_item(con, "order-1", "ペン", 120, 5)
    remove_item(con, "order-1", "ペン")
    add_item(con, "order-1", "付箋", 200, 1)
    confirm_order(con, "order-1")
    with con.cursor() as cur:
        cur.execute("SELECT seq, stream_id, version, type, payload FROM events ORDER BY seq")
        for row in cur.fetchall():
            print(" ", row[0], row[1], f"v{row[2]}", row[3], json.dumps(row[4], ensure_ascii=False))

        print("== 2. イベントから現在の状態を組み立てる ==")
        print(" 現在:", load(cur, "order-1"))
        print(" 版3の時点:", load(cur, "order-1", 3))
    con.commit()

    print("== 3. 確定済みの注文に明細を足そうとする ==")
    try:
        add_item(con, "order-1", "消しゴム", 80, 1)
    except ValueError as e:
        con.rollback(); print(" 受け付けない:", e)

    print("== 4. 同じ版に2人が追記しようとする ==")
    place_order(con, "order-2", "井上")
    con2 = connect()
    with con.cursor() as c1, con2.cursor() as c2:
        v1 = load(c1, "order-2").version; v2 = load(c2, "order-2").version      # 2人とも版1を読む
        append(c1, "order-2", v1, "ItemAdded", {"product": "ノート", "unit_price": 300, "quantity": 1}); con.commit()
        try:
            append(c2, "order-2", v2, "ItemAdded", {"product": "ペン", "unit_price": 120, "quantity": 1}); con2.commit()
        except psycopg2.errors.UniqueViolation as e:
            con2.rollback(); print(" 後から来た側:", str(e).strip().splitlines()[0])

    print("== 5. 読み取り用テーブルを組み立てる ==")
    print(" 反映前:", get_summary(con, "order-1"))
    print(" 反映したイベント数:", project_once(con))
    print(" 反映後:", get_summary(con, "order-1"))
    with con.cursor() as cur:
        cur.execute("TRUNCATE order_summaries; UPDATE projector_offsets SET last_seq = 0")
    con.commit()
    print(" 消して作り直し:", project_once(con), "件 →", get_summary(con, "order-1"))

    print("== 6. 反映遅れ：0.5 秒おきに反映する別スレッドを動かす ==")
    stop = threading.Event()
    def projector_loop():
        pcon = connect()
        while not stop.is_set():
            time.sleep(0.5)
            project_once(pcon)
        pcon.close()
    t = threading.Thread(target=projector_loop); t.start()
    time.sleep(0.6)
    rcon = connect()
    v = add_item(con, "order-2", "付箋", 200, 3)
    print(f" 追記した版: {v}")
    print(" 直後に読み取り側を読む      :", get_summary(rcon, "order-2"))
    print(" 対処1 書き込み側を読む      :", get_summary_from_write_side(con, "order-2"))
    t0 = time.monotonic()
    row = get_summary_at_least(rcon, "order-2", v)
    print(f" 対処2 版{v}以上になるまで待つ:", row, f"(待ち {time.monotonic() - t0:.2f} 秒)")
    stop.set(); t.join()
    v = add_item(con, "order-2", "消しゴム", 80, 1)
    t0 = time.monotonic()
    print(f" 反映が止まっているとき（上限 1 秒）:", get_summary_at_least(rcon, "order-2", v, timeout=1.0), f"(待ち {time.monotonic() - t0:.2f} 秒)")

== 1. イベントを追記する ==
  1 order-1 v1 OrderPlaced {"customer_name": "青木"}
  2 order-1 v2 ItemAdded {"product": "ノート", "quantity": 2, "unit_price": 300}
  3 order-1 v3 ItemAdded {"product": "ペン", "quantity": 5, "unit_price": 120}
  4 order-1 v4 ItemRemoved {"product": "ペン"}
  5 order-1 v5 ItemAdded {"product": "付箋", "quantity": 1, "unit_price": 200}
  6 order-1 v6 OrderConfirmed {}
== 2. イベントから現在の状態を組み立てる ==
 現在: Order(customer_name='青木', status='confirmed', items={'ノート': (300, 2), '付箋': (200, 1)}, version=6)
 版3の時点: Order(customer_name='青木', status='draft', items={'ノート': (300, 2), 'ペン': (120, 5)}, version=3)
== 3. 確定済みの注文に明細を足そうとする ==
 受け付けない: 明細を変えられるのは下書きの注文だけです
== 4. 同じ版に2人が追記しようとする ==
 後から来た側: duplicate key value violates unique constraint "uq_events_stream_version"
== 5. 読み取り用テーブルを組み立てる ==
 反映前: None
 反映したイベント数: 8
 反映後: ('confirmed', 2, 800, 6)
 消して作り直し: 8 件 → ('confirmed', 2, 800, 6)
== 6. 反映遅れ：0.5 秒おきに反映する別スレッドを動かす ==
 追記した版: 3
 直後に読み取り側を読む      : ('draft', 1, 300, 2)
 対処1 書き込み側を読む      : ('draft', 2, 900, 3)
 対処2 版3以上になるまで待つ: ('draft', 2, 900, 3) (待ち 0.41 秒)
 反映が止まっているとき（上限 1 秒）: None (待ち 1.02 秒)

出力から読み取れることを整理します。

反映遅れで更新前の値が見える現象と対処

段階3と段階4では、書き込みが終わってから読み取り側に届くまでに時間差があります。上の「6」では、明細を足して版3になった直後に読み取り側を読むと、版2の内容（1件・300）が返りました。

「版番号で待つ」は、上のコードの get_summary_at_least です。反映の処理が止まっているときは上限の1秒で None を返しました。上限なしで待つ作りにしないことが大切です。

読み飛ばしに気をつける：通し番号の順とコミットの順

読み取り側は「前回の通し番号より大きいイベント」を読みます。ここに見落としやすい点があります。番号を先に取ったトランザクションが、後からコミットする場合です。

# gap.py … 通し番号の順とコミットの順が入れ替わると、読み取り側が1件読み飛ばす
import threading, time
from conn import connect
from es import append, project_once, get_summary

def setup(con):
    with con.cursor() as cur:
        cur.execute(open("stage4.sql", encoding="utf-8").read())
    con.commit()

def scenario(use_lock):
    con = connect(); setup(con)
    a, b, p = connect(), connect(), connect()
    def slow_writer():                                   # A：先に番号を取り、コミットが遅い
        with a.cursor() as cur:
            if use_lock: cur.execute("SELECT pg_advisory_xact_lock(42)")
            append(cur, "order-A", 0, "OrderPlaced", {"customer_name": "青木"})
            time.sleep(0.6)
        a.commit()
    t = threading.Thread(target=slow_writer); t.start()
    time.sleep(0.2)
    with b.cursor() as cur:                              # B：後から番号を取り、先にコミットする
        if use_lock: cur.execute("SELECT pg_advisory_xact_lock(42)")
        append(cur, "order-B", 0, "OrderPlaced", {"customer_name": "井上"})
    b.commit()
    n1 = project_once(p)                                 # A のコミット前に反映が走る
    t.join()
    n2 = project_once(p)
    with con.cursor() as cur:
        cur.execute("SELECT seq, stream_id FROM events ORDER BY seq"); ev = cur.fetchall()
        cur.execute("SELECT last_seq FROM projector_offsets"); last = cur.fetchone()[0]
        cur.execute("SELECT order_id FROM order_summaries ORDER BY 1"); done = [r[0] for r in cur.fetchall()]
    print(f"[{'追記の前に pg_advisory_xact_lock' if use_lock else 'ロックなし'}]")
    print("  events:", ev)
    print(f"  1回目の反映 {n1} 件、2回目 {n2} 件、last_seq={last}")
    print("  読み取り用テーブルにある注文:", done)

scenario(False)
scenario(True)

[ロックなし]
  events: [(1, 'order-A'), (2, 'order-B')]
  1回目の反映 1 件、2回目 0 件、last_seq=2
  読み取り用テーブルにある注文: ['order-B']
[追記の前に pg_advisory_xact_lock]
  events: [(1, 'order-A'), (2, 'order-B')]
  1回目の反映 2 件、2回目 0 件、last_seq=2
  読み取り用テーブルにある注文: ['order-A', 'order-B']

ロックなしでは、番号1（order-A）がコミットされる前に番号2まで反映が進み、order-A が読み取り用テーブルに入らないままになりました。追記の前にロックを取って番号の順とコミットの順をそろえると、2件とも反映されました。追記の量が多い場合は、ロックの代わりに「少し前の位置から読み直し、同じイベントが2回来ても結果が変わらない更新にする」方法も選べます。

リードレプリカやビューとの違い

「読み取りを分ける」方法はほかにもあります。分けているものが違います。

CRUDのままビューを足した場合の実行結果です（PostgreSQL 16）。段階1のテーブルに、注文1件（青木）と明細1件（ノート 300×2）を入れた状態から始めています。

CREATE VIEW order_summaries_v AS
SELECT o.id AS order_id, o.customer_name, o.status, COUNT(i.product) AS item_count,
       COALESCE(SUM(i.unit_price * i.quantity), 0) AS total_amount
FROM orders o LEFT JOIN order_items i ON i.order_id = o.id GROUP BY o.id;
CREATE MATERIALIZED VIEW order_summaries_mv AS SELECT * FROM order_summaries_v;

INSERT INTO order_items VALUES (1, 'ペン', 120, 5);   -- ビューを作った後に明細を足す

SELECT 'view' AS src, item_count, total_amount FROM order_summaries_v
UNION ALL SELECT 'materialized view', item_count, total_amount FROM order_summaries_mv;
REFRESH MATERIALIZED VIEW order_summaries_mv;
SELECT 'materialized view (REFRESH 後)' AS src, item_count, total_amount FROM order_summaries_mv;

        src        | item_count | total_amount 
-------------------+------------+--------------
 view              |          2 |         1200
 materialized view |          1 |          600
(2 rows)

              src               | item_count | total_amount 
--------------------------------+------------+--------------
 materialized view (REFRESH 後) |          2 |         1200
(1 row)

ビューは常に最新で、マテリアライズドビューは REFRESH まで前の内容のままでした。「一覧のSQLが複雑」というだけなら、まずビューで足ります。 読み取りの形を1行ずつ細かく更新したい、別のDBや検索製品に置きたい、となったときが読み取り用テーブルの出番です。

採用するか見送るかの判断表

Fowler 氏の解説も、CQRSは複雑さを大きく増やすことがあるので慎重に使うこと、読み取りの要求だけが重い場合は報告用のデータベースを別に持つ方法もあることを述べています。段階は上から順に試し、必要が出た所だけ次へ進むのが無理のない進め方です。

確かめて分かったこと

• 段階1・2では反映の遅れは起きませんでした。 遅れが現れたのは、読み取り用テーブルを別のタイミングで更新する段階3・4からです。CQRSにすれば必ず遅れる、というわけではありません。
• 読み取り用テーブルは、TRUNCATE してもイベントから同じ内容に戻りました（8件を読み直して同じ行）。読み取りの形を後から変えたくなったときも、同じ手順で作り直せます。
• 通し番号の順とコミットの順が入れ替わると、イベントを1件読み飛ばしました。 単純な「前回より大きい番号を読む」だけでは足りず、順番をそろえる工夫が要ります。
• 同じ版への同時追記は、一意制約で後から来た側が止まりました。 楽観ロックのための列を別に用意しなくても、(stream_id, version) の制約がその役目を果たします。
• 「版番号で待つ」は0.41秒で返り、反映が止まっているときは上限の1秒で打ち切れました。 上限と、超えたときの表示を決めておく必要があります。
• マテリアライズドビューは REFRESH まで前の内容のままでした。 ビューは常に最新でした。

よくある質問

Q. CQRSとは何ですか？ A. 更新のためのモデルと読み取りのためのモデルを分ける設計です。テーブルやDBを分けるかどうかは段階の違いで、コードの中で分けるだけでもCQRSに当たります。

Q. CQRSとイベントソーシングの違いは何ですか？ A. CQRSは読み書きの「形」を分ける話、イベントソーシングは「何を保存するか」の話です。CQRSだけ、イベントソーシングだけ、両方、のどれも選べます。

Q. CQRSにするとDBを2つに分けないといけませんか？ A. 分けなくても構いません。同じDBの中に読み取り用テーブルを持つ形（段階2）なら、同じトランザクションで更新でき、反映の遅れもありません。

Q. CQRSで更新直後に更新前の値が見えるときはどうしますか？ A. 更新した本人の次の読み取りだけ書き込み側から返すか、更新の戻り値の版番号まで読み取り側が追いつくのを上限つきで待ちます。どちらも合わない画面は、同じトランザクションで更新する形に戻します。

Q. CQRSとリードレプリカの違いは何ですか？ A. リードレプリカは同じ形のテーブルを別のサーバーに複製するもので、CQRSは読み取りの形そのものを変えるものです。両方を組み合わせることもできます。

まとめ

• CQRSは読み書きのモデルを分けること、イベントソーシングは出来事を保存すること。 別々に採用できる。
• CQRSは4段階で考える。 モデルだけ分ける → 読み取り用テーブル → DBも分ける → イベント保存。
• 反映の遅れが出るのは、読み取り側を別のタイミングで更新する段階から。 同じトランザクションで更新する段階2までは遅れない。
• 対処は「書き込み直後だけ書き込み側を読む」と「版番号で上限つきで待つ」。
• イベントを通し番号で追うときは、番号の順とコミットの順のずれに備える。
• 一覧のSQLが複雑なだけならビューで足りる。 段階を進めるのは必要が出た所だけ。

参考・出典

• Martin Fowler「CQRS」 https://martinfowler.com/bliki/CQRS.html
• Martin Fowler「Event Sourcing」 https://martinfowler.com/eaaDev/EventSourcing.html

（いずれも 2026-10-06 に内容を確認）

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。