ストーリー

同期 vs 非同期

同期処理（Before）

// 全ての処理を順番に実行 → 合計2000ms以上
async function placeOrder(orderData: OrderData): Promise<Order> {
  const order = await db.createOrder(orderData);        // 200ms
  await paymentService.charge(order.totalAmount);        // 500ms
  await inventoryService.reserve(order.items);           // 300ms
  await emailService.sendConfirmation(order);            // 400ms
  await receiptService.generate(order);                  // 300ms
  await analyticsService.trackPurchase(order);           // 200ms
  return order; // ユーザーは2000ms以上待つ
}

非同期処理（After）

// 必須処理だけ同期、残りはキューに投入 → 700ms
async function placeOrder(orderData: OrderData): Promise<Order> {
  // 同期的に必要な処理のみ
  const order = await db.createOrder(orderData);         // 200ms
  await paymentService.charge(order.totalAmount);        // 500ms

  // 残りはメッセージキューに投入（非同期）
  await messageQueue.publish('order.created', {
    orderId: order.id,
    items: order.items,
    userEmail: orderData.email,
  });

  return order; // ユーザーは700msで応答を受け取る
}

// バックグラウンドワーカーで処理
class OrderWorker {
  async handleOrderCreated(event: OrderCreatedEvent): Promise<void> {
    await inventoryService.reserve(event.items);
    await emailService.sendConfirmation(event.orderId);
    await receiptService.generate(event.orderId);
    await analyticsService.trackPurchase(event.orderId);
  }
}

メッセージキューの基本

// メッセージキューの抽象化
interface MessageQueue {
  // メッセージを送信
  publish(topic: string, message: any): Promise<void>;

  // メッセージを受信（ワーカー側）
  subscribe(topic: string, handler: (message: any) => Promise<void>): void;
}

// 代表的なメッセージキュー
const messageQueues = {
  redis: {
    name: 'Redis (Bull/BullMQ)',
    useCase: 'シンプルなジョブキュー、小〜中規模',
    features: ['優先度付きキュー', '遅延実行', 'リトライ'],
  },
  rabbitmq: {
    name: 'RabbitMQ',
    useCase: '複雑なルーティング、メッセージパターン',
    features: ['柔軟なルーティング', 'メッセージの永続化', 'クラスタリング'],
  },
  sqs: {
    name: 'Amazon SQS',
    useCase: 'AWSでの大規模処理',
    features: ['フルマネージド', '無限スケーリング', 'DLQ'],
  },
  kafka: {
    name: 'Apache Kafka',
    useCase: '大量のイベントストリーミング',
    features: ['超高スループット', 'イベントログ保持', 'コンシューマーグループ'],
  },
};

リトライとデッドレターキュー

// リトライ戦略の実装
class RetryableWorker {
  async processWithRetry(
    message: QueueMessage,
    handler: (msg: any) => Promise<void>,
    maxRetries: number = 3
  ): Promise<void> {
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        await handler(message.body);
        await message.ack(); // 処理成功
        return;
      } catch (error) {
        console.error(`Attempt ${attempt}/${maxRetries} failed:`, error);

        if (attempt < maxRetries) {
          // 指数バックオフで待機
          const delay = Math.pow(2, attempt) * 1000; // 2s, 4s, 8s
          await sleep(delay);
        } else {
          // 最大リトライ回数超過 → デッドレターキューへ
          await this.moveToDeadLetterQueue(message, error);
        }
      }
    }
  }

  private async moveToDeadLetterQueue(
    message: QueueMessage,
    error: Error
  ): Promise<void> {
    await this.dlq.publish({
      originalMessage: message.body,
      error: error.message,
      failedAt: new Date().toISOString(),
      attempts: message.retryCount,
    });
    await message.ack(); // 元のキューからは削除
  }
}

指数バックオフとジッター

べき等性（Idempotency）

非同期処理では同じメッセージが複数回処理される可能性があるため、べき等性が重要です。

// べき等性の確保
class IdempotentOrderProcessor {
  async processOrder(event: OrderCreatedEvent): Promise<void> {
    const idempotencyKey = `order-process:${event.orderId}`;

    // 既に処理済みかチェック
    const alreadyProcessed = await this.redis.get(idempotencyKey);
    if (alreadyProcessed) {
      console.log(`Order ${event.orderId} already processed, skipping`);
      return;
    }

    // 処理を実行
    await this.inventoryService.reserve(event.items);
    await this.emailService.sendConfirmation(event.orderId);

    // 処理完了を記録（24時間保持）
    await this.redis.setex(idempotencyKey, 86400, 'processed');
  }
}

非同期化の判断基準

まとめ

チェックリスト

• 同期処理と非同期処理の使い分けを判断できる
• メッセージキューの役割と代表的な製品を理解した
• リトライとデッドレターキューの仕組みを把握した
• べき等性の重要性と実装方法を理解した

次のステップへ

次は演習です。スケーラブルなアーキテクチャを設計し、ここまで学んだスケーリング技法を統合しましょう。

推定読了時間: 40分