mybook

CQRSパターン — 読み書きを分離する

「画面が遅い」という相談

月曜日の朝、ユウキはアヤカに声をかけた。

「注文一覧ページが重くて、ユーザーから苦情が来てます」

アヤカがクエリを見た。

# 今の実装
def index
  @orders = Order.includes(:user, :order_items, :products, :coupon, :payment)
                 .where(user: current_user)
                 .order(created_at: :desc)
                 .page(params[:page])
  # 画面には「商品名、合計金額、注文日、ステータス」だけ表示
end

「注文一覧に必要な情報って?」

「商品名と合計金額と注文日とステータスです」

「じゃあ User や Coupon や Payment の全カラムを取ってくる必要ある?」

「……ないですね」

「SQLのEXPLAINを見てみよう」

EXPLAIN ANALYZE
SELECT orders.*, users.*, order_items.*, products.*, coupons.*, payments.*
FROM orders
LEFT JOIN users ON users.id = orders.user_id
LEFT JOIN order_items ON order_items.order_id = orders.id
LEFT JOIN products ON products.id = order_items.product_id
LEFT JOIN coupons ON coupons.id = orders.coupon_id
LEFT JOIN payments ON payments.order_id = orders.id
WHERE orders.user_id = 123
ORDER BY orders.created_at DESC
LIMIT 20;
 
-- 実行計画
Limit (cost=245.32..245.37 rows=20 width=1847) (actual time=52.312..52.341 rows=20 loops=1)
  -> Sort (rows=847)
      -> Hash Left Join (rows=847)
          -> ... （多数のJOIN処理）
Planning time: 8.723 ms
Execution time: 52.891 ms

「52ミリ秒。これが20件の一覧表示に……」

「そう。それがCQRSの出発点。読み取り（Query）と書き込み（Command）を分けて最適化する」

CQRSとは

CQRS（Command Query Responsibility Segregation）は、データの読み取りと書き込みを別のモデルで扱うパターン。

Loading diagram...

INFO

CQRSは「書き込みと読み取りは要件が違う」という観察から生まれた。書き込みは整合性が最重要、読み取りはパフォーマンスが最重要。同じモデルで両方を最適化するのは難しい。

CQRSの名前の由来は、Bertrand Meyerが提唱した「コマンドクエリ分離（CQS）」原則。「副作用のある関数（コマンド）と副作用のない関数（クエリ）を分ける」という原則をアーキテクチャレベルに適用したもの。

Command Side：書き込みに特化

# app/commands/create_order_command.rb
# コマンド：意図を表現する値オブジェクト
class CreateOrderCommand
  attr_reader :user_id, :items, :coupon_code, :shipping_address
 
  def initialize(user_id:, items:, coupon_code: nil, shipping_address: nil)
    @user_id = user_id
    @items = items
    @coupon_code = coupon_code
    @shipping_address = shipping_address
  end
 
  def valid?
    user_id.present? && items.present? && items.all? { |i| i[:product_id].present? && i[:quantity].to_i > 0 }
  end
end
 
# app/commands/update_order_status_command.rb
class UpdateOrderStatusCommand
  attr_reader :order_id, :new_status, :reason, :updated_by
 
  def initialize(order_id:, new_status:, reason: nil, updated_by: nil)
    @order_id = order_id
    @new_status = new_status
    @reason = reason
    @updated_by = updated_by
  end
end
# app/command_handlers/create_order_handler.rb
class CreateOrderHandler
  Result = Data.define(:success?, :order_id, :errors)
 
  def handle(command)
    raise ArgumentError, "無効なコマンドです" unless command.valid?
 
    ActiveRecord::Base.transaction do
      user = User.find(command.user_id)
      order = build_order(user, command)
      order.save!
 
      EventBus.publish(OrderCreatedEvent.new(order))
 
      Result.new(success?: true, order_id: order.id, errors: [])
    end
  rescue => e
    Result.new(success?: false, order_id: nil, errors: [e.message])
  end
 
  private
 
  def build_order(user, command)
    coupon = resolve_coupon(command.coupon_code)
    total = calculate_total(command.items, coupon)
 
    order = Order.new(
      user: user,
      total_amount: total,
      status: :pending,
      shipping_address: command.shipping_address,
      coupon: coupon
    )
 
    command.items.each do |item|
      product = Product.lock.find(item[:product_id])
      raise "#{product.name}の在庫が不足しています" unless product.in_stock?(item[:quantity].to_i)
 
      order.order_items.build(
        product: product,
        quantity: item[:quantity].to_i,
        price: product.price
      )
    end
 
    order
  end
 
  def resolve_coupon(code)
    return nil if code.blank?
    coupon = Coupon.find_by(code: code)
    raise "クーポンが無効です" unless coupon&.active?
    coupon
  end
 
  def calculate_total(items, coupon)
    subtotal = items.sum { |i| Product.find(i[:product_id]).price * i[:quantity].to_i }
    coupon ? coupon.apply_to(subtotal) : subtotal
  end
end
# app/command_handlers/update_order_status_handler.rb
class UpdateOrderStatusHandler
  VALID_TRANSITIONS = {
    "pending" => %w[confirmed cancelled],
    "confirmed" => %w[shipped cancelled],
    "shipped" => %w[delivered],
    "delivered" => [],
    "cancelled" => []
  }.freeze
 
  def handle(command)
    order = Order.lock.find(command.order_id)
 
    unless VALID_TRANSITIONS[order.status].include?(command.new_status)
      raise "#{order.status} から #{command.new_status} への遷移は無効です"
    end
 
    order.update!(
      status: command.new_status,
      "#{command.new_status}_at" => Time.current
    )
 
    EventBus.publish(OrderStatusChangedEvent.new(
      order: order,
      from: order.status_previously_was,
      to: command.new_status,
      reason: command.reason
    ))
 
    order
  end
end

Query Side：読み取りに特化

# app/queries/order_list_query.rb
# 読み取りモデル：画面に必要なデータだけを最速で返す
class OrderListQuery
  OrderSummary = Data.define(
    :id, :status, :status_label, :total_amount,
    :created_at, :item_count, :thumbnail_url
  )
 
  CACHE_TTL = 3.minutes
 
  def call(user_id:, page: 1, per: 20, status: nil)
    cache_key = "order_list:#{user_id}:p#{page}:s#{status}"
 
    Rails.cache.fetch(cache_key, expires_in: CACHE_TTL) do
      fetch_from_db(user_id: user_id, page: page, per: per, status: status)
    end
  end
 
  private
 
  def fetch_from_db(user_id:, page:, per:, status:)
    Order.connected_to(role: :reading) do
      scope = Order.where(user_id: user_id)
      scope = scope.where(status: status) if status.present?
 
      # 必要なカラムだけを SELECT（N+1 なし）
      rows = scope
        .select("orders.id, orders.status, orders.total_amount, orders.created_at,
                 COUNT(DISTINCT order_items.id) as item_count,
                 MIN(products.image_url) as thumbnail_url")
        .joins("LEFT JOIN order_items ON order_items.order_id = orders.id")
        .joins("LEFT JOIN products ON products.id = order_items.product_id")
        .group("orders.id")
        .order(created_at: :desc)
        .offset((page - 1) * per)
        .limit(per)
 
      rows.map { |row|
        OrderSummary.new(
          id: row.id,
          status: row.status,
          status_label: I18n.t("orders.status.#{row.status}"),
          total_amount: row.total_amount,
          created_at: row.created_at,
          item_count: row.item_count.to_i,
          thumbnail_url: row.thumbnail_url || "/images/placeholder.png"
        )
      }
    end
  end
end
# app/queries/order_detail_query.rb
class OrderDetailQuery
  OrderDetail = Data.define(
    :id, :status, :status_label, :total_amount,
    :created_at, :confirmed_at, :shipped_at, :delivered_at,
    :user_name, :user_email, :shipping_address,
    :items, :payment_info, :coupon_info
  )
 
  def call(order_id:, user_id:)
    order = Order.includes(
      :user,
      :coupon,
      :payment,
      order_items: :product
    ).find_by!(id: order_id, user_id: user_id)
 
    OrderDetail.new(
      id: order.id,
      status: order.status,
      status_label: I18n.t("orders.status.#{order.status}"),
      total_amount: order.total_amount,
      created_at: order.created_at,
      confirmed_at: order.confirmed_at,
      shipped_at: order.shipped_at,
      delivered_at: order.delivered_at,
      user_name: order.user.name,
      user_email: order.user.email,
      shipping_address: order.shipping_address,
      items: format_items(order.order_items),
      payment_info: format_payment(order.payment),
      coupon_info: format_coupon(order.coupon)
    )
  rescue ActiveRecord::RecordNotFound
    raise OrderNotFoundError, "Order##{order_id} が見つかりません"
  end
 
  private
 
  def format_items(order_items)
    order_items.map { |item|
      {
        product_id: item.product_id,
        product_name: item.product.name,
        product_sku: item.product.sku,
        quantity: item.quantity,
        unit_price: item.price,
        subtotal: item.price * item.quantity,
        thumbnail_url: item.product.image_url
      }
    }
  end
 
  def format_payment(payment)
    return nil unless payment
    {
      method: payment.payment_method,
      status: payment.status,
      amount: payment.amount,
      charged_at: payment.charged_at
    }
  end
 
  def format_coupon(coupon)
    return nil unless coupon
    {
      code: coupon.code,
      description: coupon.discount_description
    }
  end
end
# app/queries/order_analytics_query.rb
# 管理画面用：集計クエリ
class OrderAnalyticsQuery
  def daily_revenue(from:, to:)
    Order.connected_to(role: :reading) do
      Order.delivered
           .where(created_at: from..to)
           .group("DATE(created_at)")
           .select("DATE(created_at) as date, SUM(total_amount) as revenue, COUNT(*) as order_count")
           .order("date")
           .map { |row|
             {
               date: row.date,
               revenue: row.revenue.to_i,
               order_count: row.order_count.to_i
             }
           }
    end
  end
 
  def top_products(limit: 10, period: 30.days)
    Order.connected_to(role: :reading) do
      OrderItem.joins(:order, :product)
               .where("orders.created_at > ?", period.ago)
               .where(orders: { status: :delivered })
               .group("products.id, products.name")
               .select("products.id, products.name, SUM(order_items.quantity) as total_quantity, SUM(order_items.price * order_items.quantity) as total_revenue")
               .order("total_quantity DESC")
               .limit(limit)
    end
  end
end

コントローラーでのCQRS

# app/controllers/orders_controller.rb
class OrdersController < ApplicationController
  before_action :authenticate_user!
 
  # Query: 読み取り
  def index
    @order_summaries = OrderListQuery.new.call(
      user_id: current_user.id,
      page: params[:page]&.to_i || 1,
      status: params[:status]
    )
    @total_count = current_user.orders.count  # キャッシュ可能
  end
 
  # Query: 読み取り
  def show
    @order = OrderDetailQuery.new.call(
      order_id: params[:id].to_i,
      user_id: current_user.id
    )
  rescue OrderNotFoundError
    redirect_to orders_path, alert: "注文が見つかりません"
  end
 
  # Command: 書き込み
  def create
    command = CreateOrderCommand.new(
      user_id: current_user.id,
      items: order_params[:items],
      coupon_code: order_params[:coupon_code],
      shipping_address: order_params[:shipping_address]
    )
 
    result = CreateOrderHandler.new.handle(command)
 
    if result.success?
      redirect_to order_path(result.order_id), notice: "注文が完了しました"
    else
      @errors = result.errors
      render :new, status: :unprocessable_entity
    end
  end
 
  # Command: 書き込み
  def update_status
    command = UpdateOrderStatusCommand.new(
      order_id: params[:id].to_i,
      new_status: params[:status],
      updated_by: current_user.id
    )
 
    UpdateOrderStatusHandler.new.handle(command)
    redirect_to order_path(params[:id]), notice: "ステータスを更新しました"
  rescue => e
    redirect_to order_path(params[:id]), alert: e.message
  end
 
  private
 
  def order_params
    params.require(:order).permit(
      :coupon_code,
      :shipping_address,
      items: [:product_id, :quantity]
    )
  end
end

AWS での読み書き分離

Loading diagram...
# config/database.yml
production:
  primary:
    adapter: postgresql
    host: <%= ENV["DB_PRIMARY_HOST"] %>
    database: myapp_production
    username: <%= ENV["DB_USER"] %>
    password: <%= ENV["DB_PASSWORD"] %>
    pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
    prepared_statements: false
 
  replica:
    adapter: postgresql
    host: <%= ENV["DB_REPLICA_HOST"] %>
    database: myapp_production
    username: <%= ENV["DB_READONLY_USER"] %>
    password: <%= ENV["DB_READONLY_PASSWORD"] %>
    replica: true
    pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } * 2 %>  # レプリカは接続数を増やせる
# app/queries/order_list_query.rb（読み取りレプリカ使用）
class OrderListQuery
  def call(user_id:, **options)
    Order.connected_to(role: :reading) do
      fetch_from_db(user_id: user_id, **options)
    end
  end
end
 
# app/command_handlers/create_order_handler.rb（プライマリ使用）
class CreateOrderHandler
  def handle(command)
    ActiveRecord::Base.connected_to(role: :writing) do
      # 書き込み処理
    end
  end
end

レプリカラグの考慮：

# レプリカラグが許容できない場合（注文直後の詳細表示など）
class OrderDetailQuery
  def call(order_id:, user_id:, force_primary: false)
    if force_primary
      Order.connected_to(role: :writing) do
        fetch_detail(order_id, user_id)
      end
    else
      Order.connected_to(role: :reading) do
        fetch_detail(order_id, user_id)
      end
    end
  end
end
 
# コントローラーで: 注文作成直後はプライマリから読む
def create
  result = CreateOrderHandler.new.handle(command)
  if result.success?
    @order = OrderDetailQuery.new.call(
      order_id: result.order_id,
      user_id: current_user.id,
      force_primary: true  # 直後はレプリカラグを避ける
    )
    render :show
  end
end

読み取りモデルのキャッシュ戦略

# app/queries/order_list_query.rb
class OrderListQuery
  CACHE_TTL = 3.minutes
 
  def call(user_id:, page: 1, per: 20, status: nil)
    cache_key = cache_key_for(user_id, page, per, status)
 
    Rails.cache.fetch(cache_key, expires_in: CACHE_TTL) do
      fetch_from_db(user_id: user_id, page: page, per: per, status: status)
    end
  end
 
  private
 
  def cache_key_for(user_id, page, per, status)
    "order_list:v3:#{user_id}:p#{page}:pp#{per}:s#{status}"
  end
end
 
# キャッシュの無効化（書き込み側で実施）
class CreateOrderHandler
  def handle(command)
    order = create_order(command)
    invalidate_caches(command.user_id)
    order
  end
 
  private
 
  def invalidate_caches(user_id)
    Rails.cache.delete_matched("order_list:v3:#{user_id}:*")
    Rails.cache.delete("user_order_count:#{user_id}")
  end
end

WARNING

キャッシュの無効化タイミングを慎重に設計する。書き込み後にキャッシュを削除しないと、古いデータが表示され続ける。また、ページキャッシュの delete_matched は一部のキャッシュバックエンドでパフォーマンスが落ちる。Redisの場合はキーのパターンマッチングが可能だが、Memcachedでは使えない。

どこまで分けるか？

レベル実装向き合う問題工数
ライト（小規模）クエリオブジェクト + スコープN+1、複雑なSELECT小
ミドル（中規模）読み取りレプリカ + キャッシュ高負荷な読み取り中
ヘビー（大規模）別DBスキーマ（読み取り専用マテリアライズドビュー）超高負荷大

「最初から全部やる必要は?」

「ない」アヤカが即答した。「プロファイリングで問題が確認されてから。まずクエリオブジェクトで整理、遅ければレプリカ、それでも足りなければ別スキーマ。いきなりCQRSを導入すると、複雑さだけが増えてメリットが出ない」

Golang でのCQRS

// query/order_list.go
package query
 
import (
    "context"
    "time"
)
 
type OrderSummary struct {
    ID          int64
    Status      string
    TotalAmount int64
    CreatedAt   time.Time
    ItemCount   int
}
 
type OrderListQuery struct {
    db ReadDatabase  // 読み取り専用DBのインターフェース
}
 
func (q *OrderListQuery) Execute(ctx context.Context, userID int64, page int) ([]OrderSummary, error) {
    rows, err := q.db.QueryContext(ctx, `
        SELECT o.id, o.status, o.total_amount, o.created_at,
               COUNT(oi.id) as item_count
        FROM orders o
        LEFT JOIN order_items oi ON oi.order_id = o.id
        WHERE o.user_id = $1
        GROUP BY o.id
        ORDER BY o.created_at DESC
        LIMIT 20 OFFSET $2
    `, userID, (page-1)*20)
 
    if err != nil {
        return nil, err
    }
    defer rows.Close()
 
    var summaries []OrderSummary
    for rows.Next() {
        var s OrderSummary
        if err := rows.Scan(&s.ID, &s.Status, &s.TotalAmount, &s.CreatedAt, &s.ItemCount); err != nil {
            return nil, err
        }
        summaries = append(summaries, s)
    }
 
    return summaries, nil
}

まとめ

CQRSのメリット：

  • 読み取り専用クエリを極限まで最適化できる
  • 書き込みモデルは整合性に集中できる
  • スケールアウトの方向が選べる（読み取りだけ増やすなど）
  • 読み取り側のキャッシュが安全に設計できる

CQRSのコスト：

  • コードの量が増える
  • 読み書きの結果整合性が生じる（レプリカラグ最大数秒）
  • キャッシュ無効化の複雑さ
  • 初期設計・実装コスト

注文一覧ページの例では、適切なクエリオブジェクトと必要なカラムだけのSELECTで、52msから3msに改善した。それがCQRSの実利的な価値だ。

次章では、マイクロサービス間の複雑なトランザクション管理を扱うサーガパターンを学びます。

前の章パイプ＆フィルタパターン — データの流れを設計する
次の章サーガパターン — 分散トランザクションを管理する