メッセージキューイング活用：非同期処理によるシステム応答性と信頼性向上

// APIエンドポイント（Express.js）
app.post('/api/reports', async (req, res) => {
  try {
    // リクエストを検証
    const reportRequest = validateReportRequest(req.body);
    
    // 一意なジョブIDを生成
    const jobId = uuidv4();
    
    // キューにメッセージを送信
    await channel.sendToQueue(
      'report_generation_queue',
      Buffer.from(JSON.stringify({
        jobId,
        reportType: reportRequest.type,
        parameters: reportRequest.parameters,
        userId: req.user.id
      })),
      { persistent: true }
    );
    
    // レポート生成状態をデータベースに記録
    await db.reportJobs.create({
      id: jobId,
      status: 'PENDING',
      userId: req.user.id,
      requestedAt: new Date()
    });
    
    // 即座に応答を返す
    res.status(202).json({
      jobId,
      status: 'accepted',
      statusUrl: `/api/reports/status/${jobId}`
    });
  } catch (error) {
    console.error('Failed to queue report generation:', error);
    res.status(500).json({ error: 'Failed to process report request' });
  }
});

// ステータス確認エンドポイント
app.get('/api/reports/status/:jobId', async (req, res) => {
  const { jobId } = req.params;
  
  const job = await db.reportJobs.findOne({ where: { id: jobId } });
  if (!job) {
    return res.status(404).json({ error: 'Job not found' });
  }
  
  return res.json({
    jobId,
    status: job.status,
    resultUrl: job.status === 'COMPLETED' ? `/api/reports/download/${jobId}` : null,
    createdAt: job.completedAt || null,
    error: job.errorMessage || null
  });
});

// ワーカープロセス
channel.consume('report_generation_queue', async (msg) => {
  if (!msg) return;
  
  const content = JSON.parse(msg.content.toString());
  const { jobId, reportType, parameters, userId } = content;
  
  try {
    // ジョブステータスを更新
    await db.reportJobs.update(
      { status: 'PROCESSING' },
      { where: { id: jobId } }
    );
    
    // レポート生成（時間のかかる処理）
    const reportResult = await generateReport(reportType, parameters);
    
    // 結果をストレージに保存
    const reportUrl = await storeReportFile(jobId, reportResult);
    
    // ジョブ完了をマーク
    await db.reportJobs.update(
      {
        status: 'COMPLETED',
        resultUrl: reportUrl,
        completedAt: new Date()
      },
      { where: { id: jobId } }
    );
    
    // オプション：完了通知をユーザーに送信
    await notificationService.notify(userId, {
      type: 'REPORT_COMPLETED',
      message: `Your ${reportType} report is ready.`,
      actionUrl: `/reports/${jobId}`
    });
  } catch (error) {
    console.error(`Error processing job ${jobId}:`, error);
    
    // エラー状態をマーク
    await db.reportJobs.update(
      {
        status: 'FAILED',
        errorMessage: error.message
      },
      { where: { id: jobId } }
    );
  } finally {
    // メッセージの処理完了を確認
    channel.ack(msg);
  }
}, { noAck: false });

┌────────────────┐      ┌───────────────────┐
│                │      │                   │
│  User Service  │─────▶│  user_events      │
│                │      │  Kafka Topic      │
└────────────────┘      └─────────┬─────────┘
                                   │
                   ┌───────────────┼───────────────┐
                   │               │               │
                   ▼               ▼               ▼
         ┌─────────────────┐ ┌────────────────┐ ┌─────────────────┐
         │ Email Service   │ │ Analytics      │ │ Customer        │
         │                 │ │ Service        │ │ Service         │
         └─────────────────┘ └────────────────┘ └─────────────────┘

// イベントプロデューサー（User Service）
@Service
public class UserEventProducer {
    private final KafkaTemplate<String, UserEvent> kafkaTemplate;
    private static final String TOPIC = "user_events";
    
    @Autowired
    public UserEventProducer(KafkaTemplate<String, UserEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void publishUserCreated(User user) {
        UserEvent event = UserEvent.builder()
            .type("USER_CREATED")
            .userId(user.getId())
            .email(user.getEmail())
            .timestamp(LocalDateTime.now())
            .data(Map.of(
                "firstName", user.getFirstName(),
                "lastName", user.getLastName(),
                "registrationType", user.getRegistrationType()
            ))
            .build();
            
        kafkaTemplate.send(TOPIC, user.getId(), event);
    }
    
    public void publishUserUpdated(User user, Map<String, Object> changedFields) {
        UserEvent event = UserEvent.builder()
            .type("USER_UPDATED")
            .userId(user.getId())
            .timestamp(LocalDateTime.now())
            .data(changedFields)
            .build();
            
        kafkaTemplate.send(TOPIC, user.getId(), event);
    }
}

// イベントコンシューマー（Email Service）
@Service
public class UserEventConsumer {
    private final EmailService emailService;
    private final Logger logger = LoggerFactory.getLogger(UserEventConsumer.class);
    
    @Autowired
    public UserEventConsumer(EmailService emailService) {
        this.emailService = emailService;
    }
    
    @KafkaListener(topics = "user_events", groupId = "email-service")
    public void consume(UserEvent event) {
        logger.info("Received user event: {}", event.getType());
        
        switch (event.getType()) {
            case "USER_CREATED":
                handleNewUserRegistration(event);
                break;
            case "USER_UPDATED":
                if (event.getData().containsKey("email")) {
                    handleEmailUpdate(event);
                }
                break;
            default:
                logger.debug("No handler for event type: {}", event.getType());
        }
    }
    
    private void handleNewUserRegistration(UserEvent event) {
        try {
            emailService.sendWelcomeEmail(
                event.getEmail(),
                (String) event.getData().get("firstName")
            );
            logger.info("Welcome email sent to user {}", event.getUserId());
        } catch (Exception e) {
            logger.error("Failed to send welcome email to {}", event.getEmail(), e);
            // 失敗したメールを再試行キューに入れる実装も可能
        }
    }
    
    private void handleEmailUpdate(UserEvent event) {
        // メールアドレス変更確認の処理
    }
}

┌───────────────┐       ┌───────────────┐
│ Command API   │◄──────┤ Client        │
└───────┬───────┘       └───────▲───────┘
        │                       │
        ▼                       │
┌───────────────┐       ┌───────────────┐
│ Command       │       │ Query         │
│ Handlers      │       │ Handlers      │
└───────┬───────┘       └───────▲───────┘
        │                       │
        ▼                       │
┌───────────────┐       ┌───────────────┐
│ Event Store   │──────▶│ Read Models   │
└───────────────┘       └───────────────┘

// コマンドハンドラー（注文作成）
class CreateOrderCommandHandler {
  constructor(
    private eventStore: EventStore,
    private eventBus: EventBus
  ) {}

  async handle(command: CreateOrderCommand): Promise<string> {
    // 注文IDの生成
    const orderId = uuidv4();
    
    // 注文作成イベントの生成
    const orderCreatedEvent = new OrderCreatedEvent({
      orderId,
      customerId: command.customerId,
      items: command.items,
      totalAmount: calculateTotal(command.items),
      createdAt: new Date()
    });
    
    // イベントの永続化
    await this.eventStore.saveEvents(orderId, [orderCreatedEvent]);
    
    // イベントバスを通じてイベントを発行（非同期処理）
    this.eventBus.publish(orderCreatedEvent);
    
    return orderId;
  }
}

// イベントハンドラー（読み取りモデル更新）
class OrderCreatedEventHandler {
  constructor(private readModelRepository: OrderReadModelRepository) {}
  
  async handle(event: OrderCreatedEvent): Promise<void> {
    // 読み取りモデルの作成
    const orderReadModel = {
      id: event.orderId,
      customerId: event.customerId,
      items: event.items,
      totalAmount: event.totalAmount,
      status: 'CREATED',
      createdAt: event.createdAt,
      updatedAt: event.createdAt
    };
    
    // 読み取り用データストアに保存
    await this.readModelRepository.save(orderReadModel);
  }
}

// クエリハンドラー（注文一覧取得）
class GetCustomerOrdersQueryHandler {
  constructor(private readModelRepository: OrderReadModelRepository) {}
  
  async handle(query: GetCustomerOrdersQuery): Promise<OrderReadModel[]> {
    // 読み取り専用モデルからクエリを実行
    return this.readModelRepository.findByCustomerId(
      query.customerId,
      query.page,
      query.limit
    );
  }
}

// At least onceの実装例（受信確認後の明示的な確認応答）
channel.consume('important_task_queue', async (msg) => {
  if (!msg) return;
  
  try {
    // メッセージ処理
    await processMessage(msg.content);
    
    // 処理成功後に確認応答
    channel.ack(msg);
  } catch (error) {
    console.error('Processing failed:', error);
    
    // 再試行可能なエラーの場合は再キューイング
    // requeue: trueで同じキューに戻す
    channel.nack(msg, false, true);
    
    // または完全に失敗した場合はデッドレターキューに移動
    // channel.nack(msg, false, false);
  }
}, { noAck: false }); // 自動確認応答を無効化

@Service
public class IdempotentMessageProcessor {
    private final ProcessedMessageRepository repository;
    
    public boolean processIdempotently(String messageId, Runnable processor) {
        // すでに処理済みかチェック
        if (repository.existsById(messageId)) {
            log.info("Message {} already processed, skipping", messageId);
            return false;
        }
        
        try {
            // メッセージを処理
            processor.run();
            
            // 処理完了を記録
            repository.save(new ProcessedMessage(messageId, LocalDateTime.now()));
            return true;
        } catch (Exception e) {
            log.error("Error processing message {}", messageId, e);
            throw e;
        }
    }
}

-- 注文状態の更新（SQL）
UPDATE orders
SET status = 'SHIPPED', shipped_at = NOW(), version = version + 1
WHERE id = ? AND version = ?

// デッドレターキューの設定
await channel.assertExchange('dlx', 'direct', { durable: true });
await channel.assertQueue('failed_messages', { durable: true });
await channel.bindQueue('failed_messages', 'dlx', 'routing_key');

// メインキューの設定（デッドレター設定付き）
await channel.assertQueue('main_queue', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'dlx',
    'x-dead-letter-routing-key': 'routing_key',
    'x-message-ttl': 60000 // 1分後に自動的にDLQに移動
  }
});