Posted inDevelopment

Apache KafkaとNestJSによるイベント駆動型マイクロサービス構築：Producer、ConsumerからDead Letter Queue（DLQ）まで

Development tutorial - IT technology blog
Development tutorial - IT technology blog

現場で起きる課題：REST APIが引き起こすドミノ倒し

大規模セールの現場でよくあるシナリオを思い浮かべてみてください。アクセス数が毎秒5,000リクエストに急増し、ユーザーが決済ボタンを押すと、Order ServiceはPayment ServiceへHTTP POSTリクエストを送り、応答を待機します。しかし、銀行決済ゲートウェイの応答が遅延。わずか15秒でOrder Serviceのスレッドプールが枯渇し、最終的にシステム全体が共倒れしてしまいます。

これが同期通信（Synchronous）の代償です。チェーン上の1つのリンクに障害が発生すると、手前のサービスすべてに波及して停止してしまいます。この問題を根本から解消するアプローチが、イベント駆動型アーキテクチャ（Event-Driven Architecture：EDA）への移行です。

ユーザーを3秒待たせて決済処理を完了させる代わりに、Order Serviceはorder.createdイベントをメッセージブローカーに書き込み、即座にHTTP 202 Acceptedを返します。これによりユーザー側のレイテンシーは3,200msから80ms未満へと激減します。あとの重い処理は、Apache Kafkaと背後のワーカーが非同期で自律的にさばいてくれます。

しかし、現実はそれほど単純ではありません。Kafkaにメッセージを投入するのは簡単ですが、本当に厄介なのは不正なペイロード（壊れたデータ）が混入した時です。Consumerがクラッシュを繰り返し、オフセットがコミットされず、対象のパーティションが完全に詰まってしまいます。本番環境（Production）でシステムを安定稼働させるには、Dead Letter Queue（DLQ）の導入が不可欠です。

押さえておくべき3つの技術要素

プロジェクトの構築に入る前に、中核となる3つのコンポーネントの本質を整理しておきましょう。

  • Apache Kafka: 分散型のAppend-Onlyログとしてデータを永続化するメッセージブローカーです。トピックは複数の独立したパーティションに分割されます。ディスクへのシーケンシャルI/O機構により、毎秒数十万件のイベントを軽々と処理でき、イベントの追跡や履歴データの再再生（Replay）においてRabbitMQを大きく上回る強みを持ちます。
  • NestJS Microservices: kafkajsドライバをラップする@nestjs/microservicesモジュールです。一方向のPub/Subフローには@EventPattern()デコレータ、Request-Reply形式には@MessagePattern()を提供し、TypeScriptによる簡潔で直感的なコード記述を可能にします。
  • Dead Letter Queue (DLQ): フォーマット不正やビジネスロジック違反を含むメッセージ（いわゆるPoison Pill）を受信した場合、単純なリトライ機構ではワーカーが同じオフセットで立ち往生してしまいます。解決策として、異常なメッセージを待避用トピック（DLQ）に転送します。これによりメインパイプラインの処理を止めずに継続でき、エラーメッセージのデバッグや再処理（Reprocess）は後から安全に行えます。

実践ハンズオン：Docker環境構築からNestJSの実装まで

ステップ1：KRaftモードでローカルKafkaクラスタを起動

環境を軽量に保つためZooKeeperは使わず、Kafka 3.x以降に組み込まれているKRaftモードを採用します。docker-compose.ymlを作成します：

version: '3.8'
services:
  kafka:
    image: bitnami/kafka:3.7
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_KRAFT_CLUSTER_ID=L7HkIldmQ0aM153nBwA56w

コンテナを起動します：

docker compose up -d

ステップ2：NestJSでイベントを発行するProducerを設定

NestJSプロジェクトに必要な2つのコアライブラリをインストールします：

npm install @nestjs/microservices kafkajs

order.module.tsでKafkaクライアントを宣言します。このサービスがメッセージ送信のみを行う場合は、リソースを最適化するためにproducerOnlyModeフラグを有効にします：

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { OrderController } from './order.controller';
import { OrderService } from './order.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'KAFKA_PRODUCER',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'order-service',
            brokers: ['localhost:9092'],
          },
          producerOnlyMode: true,
        },
      },
    ]),
  ],
  controllers: [OrderController],
  providers: [OrderService],
})
export class OrderModule {}

order.service.tsにイベント発行ロジックを実装します。同一ユーザーの一連の操作が必ず同じパーティションに振り分けられるよう、常にkeyを指定するのが重要です：

import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';

@Injectable()
export class OrderService implements OnModuleInit {
  constructor(
    @Inject('KAFKA_PRODUCER') private readonly kafkaClient: ClientKafka,
  ) {}

  async onModuleInit() {
    await this.kafkaClient.connect();
  }

  async createOrder(orderData: { orderId: string; amount: number; userId: string }) {
    this.kafkaClient.emit('order.created', {
      key: orderData.userId,
      value: orderData,
    });
    return { status: 'queued', orderId: orderData.orderId };
  }
}

ステップ3：ブローカーからイベントを受信するConsumerを構築

次にbilling-serviceの実装に移ります。main.tsでNestJSをマイクロサービスモードに切り替え、Kafkaブローカーをリッスンするように設定します：

import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.KAFKA,
    options: {
      client: {
        brokers: ['localhost:9092'],
      },
      consumer: {
        groupId: 'billing-consumer-group',
      },
    },
  });
  await app.listen();
}
bootstrap();

ステップ4：Dead Letter Queue（DLQ）によるエラーハンドリング戦略

ローカル環境では問題なく動作しても、本番環境では必須フィールドの欠落したJSON、マイナスの金額、DBタイムアウトなど予期せぬ事象に遭遇します。適切にエラーを捕捉しないと、Consumerはクラッシュを繰り返し、オフセットのコミットが停止してしまいます。

標準的なベストプラクティスは、リトライ回数を制限（例：3回）することです。それでも処理できない場合は、エラー原因をヘッダーに付与した上で、ペイロードをorder.created.dlqトピックへ退避させます：

import { Controller, Logger, OnModuleInit } from '@nestjs/common';
import { EventPattern, Payload, Ctx, KafkaContext } from '@nestjs/microservices';
import { Kafka, Producer } from 'kafkajs';

@Controller()
export class BillingConsumer implements OnModuleInit {
  private readonly logger = new Logger(BillingConsumer.name);
  private dlqProducer: Producer;

  constructor() {
    const kafka = new Kafka({ brokers: ['localhost:9092'] });
    this.dlqProducer = kafka.producer();
  }

  async onModuleInit() {
    await this.dlqProducer.connect();
  }

  @EventPattern('order.created')
  async handleOrderCreated(@Payload() message: any, @Ctx() context: KafkaContext) {
    const rawMessage = context.getMessage();
    const maxRetries = 3;
    let attempts = 0;

    while (attempts < maxRetries) {
      try {
        attempts++;
        this.processPayment(message);
        return;
      } catch (err) {
        this.logger.warn(`試行 ${attempts}/${maxRetries} 回目が失敗: ${err.message}`);
        if (attempts >= maxRetries) {
          await this.forwardToDLQ(rawMessage, err.message);
        }
      }
    }
  }

  private processPayment(data: any) {
    if (!data.amount || data.amount <= 0) {
      throw new Error(`不正な金額です: ${data.amount}`);
    }
    this.logger.log(`注文 ${data.orderId} の決済が完了しました: $${data.amount}`);
  }

  private async forwardToDLQ(rawMessage: any, reason: string) {
    this.logger.error(`異常メッセージをDLQへ転送: ${reason}`);
    await this.dlqProducer.send({
      topic: 'order.created.dlq',
      messages: [
        {
          key: rawMessage.key,
          value: rawMessage.value,
          headers: {
            'x-dead-letter-reason': reason,
            'x-failed-at': new Date().toISOString(),
            'x-original-topic': 'order.created',
          },
        },
      ],
    });
  }
}

Kafkaを本番運用するための3つの鉄則

  • Message Keyを決して忘れないこと： Keyを指定しない場合、Kafkaはラウンドロビン方式でメッセージを各パーティションに分散します。その結果、注文作成イベントよりも前に決済イベントが処理されてしまうような順序逆転が起き得ます。処理順序を厳格に保証するには、必ずuserIdやorderIdをパーティションキーとして指定してください。
  • Consumerは常に冪等性（Idempotency）を担保して設計すること： Kafkaのデフォルトの配信保証はat-least-once（最低1回）です。ネットワークの瞬断やConsumerのリバランス時には、同一メッセージが重複して届くことが日常茶飯事です。Redisに24時間のTTL付きでidempotency_keyを保持するか、PostgreSQLの一意制約（Unique Constraint）を活用して、二重課金などの不整合を確実に防ぎましょう。
  • Consumer Lagを厳密にモニタリングすること： Consumer Lagは、トピックに書き込まれた最新メッセージとConsumerが読み取りを完了したメッセージとの差分（遅延量）を表します。この値が数千件を超えて増加し続ける場合、システムでボトルネックが発生しています。即座にConsumerインスタンスのスケールアウトやI/O処理の最適化を行ってください。

まとめ

KafkaとNestJSを組み合わせることで、従来のRESTアーキテクチャが抱えていた同期通信のボトルネックを完全に解消できます。ProducerとConsumerが疎結合になり、DLQという防波堤によってデータパイプラインの安定性が担保されます。

これから導入を検討している場合は、まず認証メールの送信、検索インデックスの同期、プッシュ通知といった非クリティカルな非同期タスクから始めることをおすすめします。パーティションやLagの制御に慣れてからコアなビジネスロジックをイベント駆動型へと移行すれば、より安全かつ確実なシステム刷新が実現できます。

Share:
Tags: