イベント駆動アーキテクチャ — リアクティブなシステム

「メール送信が失敗しても注文を止めたくない」

ある夜、カオリにアラートが届いた。メール送信サービス（SendGrid）が5分間ダウンした。その間に来た注文は全て失敗し、エラーレートが急上昇した。

根本原因を調べると、注文確定のトランザクション内でメール送信をしていたことが分かった。

# 問題のあるコード
ActiveRecord::Base.transaction do
  order.save!
  inventory.reserve!(order.quantity)
  
  # ここが失敗すると、注文全体がロールバックされる
  OrderMailer.confirmation(order).deliver_now
end

「メール送信は注文確定の必須処理じゃない。送信が遅れても、後で送れればいい」

これがイベント駆動アーキテクチャの出番だ。

INFO

イベント駆動アーキテクチャでは、サービスがイベント（「何かが起きた」という事実）を発行し、そのイベントに反応するコンシューマーが独立して動作します。プロデューサーはコンシューマーの存在を知らず、疎結合が実現されます。

イベント駆動の基本

Loading diagram...

注文サービスはイベントを発行するだけ。誰が購読しているかを知らない。新しいコンシューマーを追加しても、注文サービスのコードは変わらない。

Railsでのイベント発行

EventBridgeへの発行

# app/services/event_publisher.rb
class EventPublisher
  def self.publish(event_type, detail, source: 'myapp.orders')
    client = Aws::EventBridge::Client.new(region: 'ap-northeast-1')
    
    client.put_events(
      entries: [{
        source: source,
        detail_type: event_type,
        detail: detail.to_json,
        event_bus_name: ENV.fetch('EVENT_BUS_NAME', 'default'),
        # トレーシング用
        resources: ["arn:aws:...#{detail[:order_id]}"]
      }]
    )
  rescue Aws::EventBridge::Errors::ServiceError => e
    Rails.logger.error "EventBridge publish failed: #{e.message}"
    # イベント発行失敗は致命的ではない（補助的な処理用なら）
    # ただしデータ整合性に関わる場合はSQSなどを使う
  end
end
 
# app/services/order_service.rb
class OrderService
  def confirm_order(order)
    ActiveRecord::Base.transaction do
      order.update!(status: 'confirmed')
      Inventory.reserve!(order.product_id, order.quantity)
      # ここではDBの整合性のみ保証する
    end
    
    # トランザクション外でイベントを発行
    EventPublisher.publish('order.confirmed', {
      order_id: order.id,
      user_id: order.user_id,
      user_email: order.user.email,
      product_id: order.product_id,
      product_name: order.product.name,
      quantity: order.quantity,
      total_price: order.total_price,
      confirmed_at: order.updated_at.iso8601
    })
  end
end

Outboxパターン（確実なイベント発行）

「EventBridgeへの発行自体が失敗したら？」という問題への対策。

# db/migrate/create_outbox_messages.rb
class CreateOutboxMessages < ActiveRecord::Migration[7.1]
  def change
    create_table :outbox_messages do |t|
      t.string :aggregate_type, null: false
      t.string :aggregate_id, null: false
      t.string :event_type, null: false
      t.jsonb :payload, null: false, default: {}
      t.string :status, null: false, default: 'pending'
      t.integer :retry_count, null: false, default: 0
      t.datetime :last_attempted_at
      t.datetime :processed_at
      
      t.timestamps
    end
    
    add_index :outbox_messages, [:status, :created_at]
  end
end
 
# トランザクション内でアウトボックスにも書く
class OrderService
  def confirm_order(order)
    ActiveRecord::Base.transaction do
      order.update!(status: 'confirmed')
      Inventory.reserve!(order.product_id, order.quantity)
      
      # 同じトランザクション内でアウトボックスに書く（確実！）
      OutboxMessage.create!(
        aggregate_type: 'Order',
        aggregate_id: order.id.to_s,
        event_type: 'order.confirmed',
        payload: {
          order_id: order.id,
          user_email: order.user.email,
          # ...
        }
      )
    end
    # トランザクションが成功すれば、アウトボックスにも確実に書かれている
  end
end
 
# app/jobs/outbox_relay_job.rb
class OutboxRelayJob < ApplicationJob
  queue_as :outbox
  
  def perform
    OutboxMessage.where(status: 'pending').find_each do |message|
      begin
        EventPublisher.publish(message.event_type, message.payload)
        message.update!(status: 'processed', processed_at: Time.current)
      rescue => e
        message.increment!(:retry_count)
        message.update!(
          status: message.retry_count >= 3 ? 'failed' : 'pending',
          last_attempted_at: Time.current
        )
      end
    end
  end
end

SQSコンシューマーの実装

Lambda コンシューマー

# handlers/send_order_confirmation.rb
require 'json'
require 'aws-sdk-ses'
 
$ses_client ||= Aws::SES::Client.new(region: 'ap-northeast-1')
 
def handler(event:, context:)
  failed_items = []
  
  event['Records'].each do |record|
    message_id = record['messageId']
    
    begin
      # EventBridge → SNS → SQS の場合、bodyはJSON
      body = JSON.parse(record['body'])
      detail = JSON.parse(body['Message'])
      
      send_confirmation_email(detail)
      
      puts "Sent confirmation for order #{detail['order_id']}"
    rescue => e
      puts "Failed to process message #{message_id}: #{e.class} - #{e.message}"
      failed_items << { itemIdentifier: message_id }
    end
  end
  
  # 部分的な失敗を報告（失敗したメッセージだけ再キューイング）
  { batchItemFailures: failed_items }
end

Railsでの非同期コンシューマー

# app/jobs/process_order_confirmed_job.rb
class ProcessOrderConfirmedJob < ApplicationJob
  queue_as :order_events
  
  # 冪等性: 同じイベントを複数回受け取っても安全
  def perform(event_data)
    order_id = event_data['order_id']
    
    # 処理済みチェック
    return if ProcessedEvent.exists?(event_id: event_data['event_id'])
    
    ActiveRecord::Base.transaction do
      # ドメイン処理
      update_analytics(event_data)
      award_loyalty_points(event_data)
      
      # 処理済みとしてマーク
      ProcessedEvent.create!(
        event_id: event_data['event_id'],
        event_type: event_data['event_type'],
        processed_at: Time.current
      )
    end
  rescue ActiveRecord::RecordNotUnique
    # 二重処理の競合: 無視してOK
    Rails.logger.info "Event #{event_data['event_id']} already processed"
  end
  
  private
  
  def update_analytics(event_data)
    DailySales.find_or_initialize_by(date: Date.today).tap do |record|
      record.increment(:order_count)
      record.increment(:total_revenue, event_data['total_price'])
      record.save!
    end
  end
  
  def award_loyalty_points(event_data)
    User.find(event_data['user_id'])
        .loyalty_points
        .award!(event_data['total_price'] / 100)
  end
end

AWSでのイベント駆動構成

Loading diagram...

# AWS SAMでのイベント駆動構成
Resources:
  # カスタムイベントバス
  OrderEventBus:
    Type: AWS::Events::EventBus
    Properties:
      Name: myapp-orders
 
  # SNSトピック（ファンアウト用）
  OrderConfirmedTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: order-confirmed
      
  # EventBridge → SNS のルール
  OrderConfirmedRule:
    Type: AWS::Events::Rule
    Properties:
      EventBusName: !Ref OrderEventBus
      EventPattern:
        source: ["myapp.orders"]
        detail-type: ["order.confirmed"]
      Targets:
        - Id: OrderConfirmedSNS
          Arn: !Ref OrderConfirmedTopic
          
  # SQS: メール送信用
  EmailNotificationQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: email-notification
      VisibilityTimeout: 60
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt EmailNotificationDLQ.Arn
        maxReceiveCount: 3
        
  # デッドレターキュー
  EmailNotificationDLQ:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: email-notification-dlq
      MessageRetentionPeriod: 1209600  # 14日間保持
      
  # SNS → SQS サブスクリプション
  EmailQueueSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref OrderConfirmedTopic
      Protocol: sqs
      Endpoint: !GetAtt EmailNotificationQueue.Arn

デッドレターキュー（DLQ）による障害対策

失敗したメッセージは自動的にDLQへ。後で調査・再処理できる。

# app/jobs/dlq_reprocessing_job.rb
class DlqReprocessingJob < ApplicationJob
  queue_as :admin
  
  def perform(queue_name:, max_messages: 100)
    sqs = Aws::SQS::Client.new(region: 'ap-northeast-1')
    dlq_url = sqs.get_queue_url(queue_name: "#{queue_name}-dlq").queue_url
    original_queue_url = sqs.get_queue_url(queue_name: queue_name).queue_url
    
    processed_count = 0
    
    loop do
      break if processed_count >= max_messages
      
      response = sqs.receive_message(
        queue_url: dlq_url,
        max_number_of_messages: 10
      )
      
      break if response.messages.empty?
      
      response.messages.each do |message|
        # DLQから元のキューに移動
        sqs.send_message(
          queue_url: original_queue_url,
          message_body: message.body
        )
        
        sqs.delete_message(
          queue_url: dlq_url,
          receipt_handle: message.receipt_handle
        )
        
        processed_count += 1
      end
    end
    
    puts "DLQから#{processed_count}件を再処理キューに移動しました"
  end
end

イベントスキーマの管理

イベントのスキーマが変わると、コンシューマーが壊れる。バージョニングで対処する。

# イベントのバージョニング
class OrderConfirmedEvent
  VERSION = '2.0'
  
  def to_event
    {
      version: VERSION,
      event_type: 'order.confirmed',
      # v2で追加されたフィールド
      total_price: total_price,
      currency: 'JPY',
      # 後方互換性のため古いフィールドも維持
      total: total_price  # v1との互換性
    }
  end
end
 
# コンシューマー側でバージョン対応
def process_event(event)
  version = event['version'] || '1.0'
  
  case version
  when '1.0'
    total = event['total']
  when '2.0'
    total = event['total_price']
  else
    raise "未対応のイベントバージョン: #{version}"
  end
  
  # 処理続行...
end

EventBridge Pipes（新しいパターン）

# EventBridge PipesでSQS → Lambda をシンプルに繋ぐ
OrderProcessingPipe:
  Type: AWS::Pipes::Pipe
  Properties:
    Name: order-processing-pipe
    Source: !GetAtt EmailNotificationQueue.Arn
    Target: !GetAtt SendEmailFunction.Arn
    SourceParameters:
      SqsQueueParameters:
        BatchSize: 10
        MaximumBatchingWindowInSeconds: 5
    # フィルタリング: 特定のイベントのみ処理
    Filter:
      Filters:
        - Pattern: '{"body": {"event_type": ["order.confirmed"]}}'

疎結合の恩恵

カオリは振り返った。「SendGridが落ちたあの夜、もしイベント駆動だったら何が起きていたか」

イベント駆動の場合:
1. 注文確定 → DBに保存 ✅
2. EventBridgeにイベント発行 ✅
3. SendGridダウン → Lambdaが失敗
4. SQSがメッセージを保持（VisibilityTimeout後に再試行）
5. SendGrid復帰後、自動的にメール送信 ✅
6. ユーザーへの注文確定に影響なし ✅

INFO

イベント駆動アーキテクチャの最大の強みは「部分的な障害の隔離」です。一つのサービスが落ちても、イベントキューがバッファとなり、システム全体には影響しません。

次章では、ここまで学んだ全アーキテクチャを俯瞰する。「どの文脈でどれを選ぶか」のトレードオフを実践的に分析する最終比較へ進もう。