1️⃣ 의존성 추가 (Maven)

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2️⃣ RabbitMQ 설정 클래스

@Configuration
public class RabbitMQConfig {
    
    // 큐 선언
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true); // durable = true
    }
    
    @Bean
    public Queue paymentQueue() {
        return new Queue("payment.queue", true);
    }
    
    // Topic Exchange 선언
    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange("order.exchange");
    }
    
    // 바인딩 설정
    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange orderExchange) {
        return BindingBuilder
            .bind(orderQueue)
            .to(orderExchange)
            .with("order.created");
    }
    
    @Bean
    public Binding paymentBinding(Queue paymentQueue, TopicExchange orderExchange) {
        return BindingBuilder
            .bind(paymentQueue)
            .to(orderExchange)
            .with("order.payment.*");
    }
    
    // JSON 메시지 컨버터
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

3️⃣ 메시지 발행 (Producer)

@Service
public class OrderService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void createOrder(Order order) {
        // 주문 처리 로직...
        
        // 메시지 발행
        rabbitTemplate.convertAndSend(
            "order.exchange",           // Exchange 이름
            "order.created",            // Routing Key
            order,                      // 메시지 객체
            message -> {
                // 메시지 속성 설정
                message.getMessageProperties()
                    .setHeader("orderId", order.getId());
                message.getMessageProperties()
                    .setHeader("timestamp", System.currentTimeMillis());
                return message;
            }
        );
        
        System.out.println("주문 생성 이벤트 발행: " + order.getId());
    }
    
    public void processPayment(String orderId, PaymentInfo payment) {
        // 결제 처리...
        
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "order.payment.completed",
            payment
        );
        
        System.out.println("결제 완료 이벤트 발행: " + orderId);
    }
}

4️⃣ 메시지 소비 (Consumer)

@Component
public class OrderEventListener {
    
    // 기본 리스너
    @RabbitListener(queues = "order.queue")
    public void handleOrderCreated(Order order) {
        System.out.println("주문 수신: " + order.getId());
        
        // 재고 확인 및 예약
        inventoryService.reserveStock(order);
        
        // 배송 준비
        shippingService.prepareShipment(order);
    }
    
    // 메시지와 헤더 함께 받기
    @RabbitListener(queues = "payment.queue")
    public void handlePayment(
        PaymentInfo payment,
        @Header("orderId") String orderId,
        @Header("timestamp") Long timestamp
    ) {
        System.out.println("결제 정보 수신: " + orderId);
        System.out.println("타임스탬프: " + new Date(timestamp));
        
        // 결제 검증 및 처리
        paymentService.verifyAndProcess(payment);
    }
    
    // 에러 핸들링
    @RabbitListener(queues = "order.queue")
    public void handleOrderWithErrorHandling(Order order) {
        try {
            processOrder(order);
        } catch (Exception e) {
            // 에러 로깅
            log.error("주문 처리 실패: " + order.getId(), e);
            
            // DLQ(Dead Letter Queue)로 전송
            rabbitTemplate.convertAndSend(
                "order.dlx",
                "order.failed",
                order
            );
        }
    }
}

🔄 메시지 재시도 (Retry) 설정

@Configuration
public class RetryConfig {
    
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        ConnectionFactory connectionFactory
    ) {
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // 재시도 정책 설정
        factory.setAdviceChain(
            RetryInterceptorBuilder.stateless()
                .maxAttempts(3)                    // 최대 3번 시도
                .backOffOptions(1000, 2.0, 10000)  // 지수 백오프
                .recoverer(new RejectAndDontRequeueRecoverer())
                .build()
        );
        
        return factory;
    }
}

💀 Dead Letter Queue (DLQ) 설정

@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order.queue")
        .withArgument("x-dead-letter-exchange", "order.dlx")
        .withArgument("x-dead-letter-routing-key", "order.failed")
        .withArgument("x-message-ttl", 60000)  // 60초 TTL
        .build();
}

@Bean
public Queue deadLetterQueue() {
    return new Queue("order.dlq", true);
}

@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange("order.dlx");
}

@Bean
public Binding deadLetterBinding() {
    return BindingBuilder
        .bind(deadLetterQueue())
        .to(deadLetterExchange())
        .with("order.failed");
}

⏱️ 지연 메시지 (Delayed Message)

public void sendDelayedNotification(String userId, String message, int delayMinutes) {
    rabbitTemplate.convertAndSend(
        "notification.exchange",
        "notification.delayed",
        message,
        msg -> {
            msg.getMessageProperties()
                .setDelay(delayMinutes * 60 * 1000);  // 밀리초 단위
            return msg;
        }
    );
    
    System.out.println(delayMinutes + "분 후 알림 발송 예약");
}

1️⃣ 의존성 추가 (Maven)

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

2️⃣ Kafka 설정 (application.yml)

spring:
  kafka:
    bootstrap-servers: localhost:9092
    
    # Producer 설정
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all  # 모든 복제본이 확인할 때까지 대기
      retries: 3
      properties:
        linger.ms: 10  # 배치 처리를 위한 대기 시간
        batch.size: 16384
    
    # Consumer 설정
    consumer:
      group-id: order-service-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest  # 처음부터 읽기
      enable-auto-commit: false  # 수동 커밋
      properties:
        spring.json.trusted.packages: "*"
        max.poll.records: 100

3️⃣ Topic 설정 클래스

@Configuration
public class KafkaTopicConfig {
    
    @Bean
    public NewTopic orderTopic() {
        return TopicBuilder.name("orders")
            .partitions(3)           // 3개 파티션
            .replicas(2)             // 2개 복제본
            .config(TopicConfig.RETENTION_MS_CONFIG, "604800000")  // 7일 보관
            .build();
    }
    
    @Bean
    public NewTopic paymentTopic() {
        return TopicBuilder.name("payments")
            .partitions(3)
            .replicas(2)
            .build();
    }
    
    @Bean
    public NewTopic notificationTopic() {
        return TopicBuilder.name("notifications")
            .partitions(5)  // 알림은 더 많은 파티션
            .replicas(2)
            .build();
    }
}

4️⃣ 메시지 발행 (Producer)

@Service
public class OrderProducerService {
    
    @Autowired
    private KafkaTemplate<String, Order> kafkaTemplate;
    
    // 기본 발행
    public void publishOrder(Order order) {
        kafkaTemplate.send("orders", order.getId(), order);
        System.out.println("주문 발행: " + order.getId());
    }
    
    // 콜백과 함께 발행
    public void publishOrderWithCallback(Order order) {
        ListenableFuture<SendResult<String, Order>> future = 
            kafkaTemplate.send("orders", order.getId(), order);
        
        future.addCallback(
            result -> {
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("발행 성공! Topic: " + metadata.topic() 
                    + ", Partition: " + metadata.partition()
                    + ", Offset: " + metadata.offset());
            },
            ex -> {
                System.err.println("발행 실패: " + ex.getMessage());
                // 재시도 로직 또는 DLQ 전송
            }
        );
    }
    
    // 헤더와 함께 발행
    public void publishOrderWithHeaders(Order order) {
        ProducerRecord<String, Order> record = 
            new ProducerRecord<>("orders", order.getId(), order);
        
        // 헤더 추가
        record.headers().add("source", "order-service".getBytes());
        record.headers().add("timestamp", 
            String.valueOf(System.currentTimeMillis()).getBytes());
        record.headers().add("version", "1.0".getBytes());
        
        kafkaTemplate.send(record);
    }
    
    // 파티션 지정 발행
    public void publishToSpecificPartition(Order order, int partition) {
        kafkaTemplate.send("orders", partition, order.getId(), order);
        System.out.println("파티션 " + partition + "에 발행");
    }
    
    // 트랜잭션 발행
    @Transactional
    public void publishOrdersInTransaction(List<Order> orders) {
        kafkaTemplate.executeInTransaction(operations -> {
            for (Order order : orders) {
                operations.send("orders", order.getId(), order);
            }
            return true;
        });
        System.out.println(orders.size() + "개 주문 트랜잭션 발행 완료");
    }
}

5️⃣ 메시지 소비 (Consumer)

@Component
public class OrderConsumerService {
    
    // 기본 컨슈머
    @KafkaListener(topics = "orders", groupId = "order-service-group")
    public void consumeOrder(Order order) {
        System.out.println("주문 수신: " + order.getId());
        processOrder(order);
    }
    
    // 메타데이터와 함께 수신
    @KafkaListener(topics = "orders", groupId = "order-service-group")
    public void consumeOrderWithMetadata(
        @Payload Order order,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
        @Header(KafkaHeaders.OFFSET) long offset,
        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp
    ) {
        System.out.println("주문 수신 - Topic: " + topic 
            + ", Partition: " + partition 
            + ", Offset: " + offset
            + ", Timestamp: " + new Date(timestamp));
        
        processOrder(order);
    }
    
    // 배치 처리
    @KafkaListener(topics = "orders", groupId = "batch-processor-group")
    public void consumeOrdersBatch(List<Order> orders) {
        System.out.println("배치 수신: " + orders.size() + "개");
        
        // 배치로 한번에 처리
        processBatch(orders);
    }
    
    // 수동 커밋
    @KafkaListener(
        topics = "orders", 
        groupId = "manual-commit-group",
        containerFactory = "manualCommitKafkaListenerContainerFactory"
    )
    public void consumeOrderWithManualCommit(
        Order order,
        Acknowledgment acknowledgment
    ) {
        try {
            processOrder(order);
            
            // 처리 성공 시 커밋
            acknowledgment.acknowledge();
            System.out.println("오프셋 커밋 완료");
            
        } catch (Exception e) {
            System.err.println("처리 실패, 커밋하지 않음: " + e.getMessage());
            // 커밋하지 않으면 다음에 다시 읽음
        }
    }
    
    // 여러 토픽 동시 구독
    @KafkaListener(
        topics = {"orders", "payments", "shipments"},
        groupId = "multi-topic-group"
    )
    public void consumeMultipleTopics(
        String message,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
    ) {
        System.out.println("토픽 " + topic + "에서 메시지 수신");
        
        switch (topic) {
            case "orders":
                handleOrder(message);
                break;
            case "payments":
                handlePayment(message);
                break;
            case "shipments":
                handleShipment(message);
                break;
        }
    }
    
    // 에러 핸들링
    @KafkaListener(topics = "orders", groupId = "error-handling-group")
    public void consumeOrderWithErrorHandling(Order order) {
        try {
            processOrder(order);
        } catch (Exception e) {
            System.err.println("처리 중 에러 발생: " + e.getMessage());
            
            // 에러 토픽으로 전송
            kafkaTemplate.send("orders-error", order.getId(), order);
        }
    }
}

6️⃣ 고급 설정 - 수동 커밋 컨테이너

@Configuration
public class KafkaConsumerConfig {
    
    @Bean
    public ConsumerFactory<String, Order> manualCommitConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "manual-commit-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            JsonDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        
        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> 
        manualCommitKafkaListenerContainerFactory() {
        
        ConcurrentKafkaListenerContainerFactory<String, Order> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(manualCommitConsumerFactory());
        factory.getContainerProperties()
            .setAckMode(ContainerProperties.AckMode.MANUAL);
        
        return factory;
    }
}

Kafka Streams 예제 - 주문 통계

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {
    
    @Bean
    public KStream<String, Order> orderStream(StreamsBuilder builder) {
        // 주문 스트림 읽기
        KStream<String, Order> orderStream = builder.stream("orders");
        
        // 1. 필터링 - 금액이 10만원 이상인 주문만
        KStream<String, Order> highValueOrders = orderStream
            .filter((key, order) -> order.getAmount() >= 100000);
        
        highValueOrders.to("high-value-orders");
        
        // 2. 변환 - 주문을 알림 메시지로 변환
        KStream<String, Notification> notifications = orderStream
            .mapValues(order -> new Notification(
                order.getUserId(),
                "주문이 접수되었습니다: " + order.getId(),
                NotificationType.ORDER_CREATED
            ));
        
        notifications.to("notifications");
        
        // 3. 그룹화 및 집계 - 사용자별 주문 수
        KTable<String, Long> orderCountByUser = orderStream
            .groupBy((key, order) -> order.getUserId())
            .count();
        
        orderCountByUser.toStream().to("order-count-by-user");
        
        // 4. 윈도우 집계 - 최근 1시간 동안의 주문 금액 합계
        KTable<Windowed<String>, Double> hourlyRevenue = orderStream
            .groupByKey()
            .windowedBy(TimeWindows.of(Duration.ofHours(1)))
            .aggregate(
                () -> 0.0,  // 초기값
                (key, order, total) -> total + order.getAmount(),
                Materialized.with(Serdes.String(), Serdes.Double())
            );
        
        // 5. 조인 - 주문과 결제 정보 조인
        KStream<String, Payment> paymentStream = builder.stream("payments");
        
        KStream<String, OrderWithPayment> enrichedOrders = orderStream
            .join(
                paymentStream,
                (order, payment) -> new OrderWithPayment(order, payment),
                JoinWindows.of(Duration.ofMinutes(5))
            );
        
        enrichedOrders.to("enriched-orders");
        
        return orderStream;
    }
}

// Kafka에서 이벤트를 받아서 RabbitMQ로 작업 분배
@Service
public class EventDistributionService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    // Kafka에서 주문 이벤트 수신
    @KafkaListener(topics = "orders", groupId = "distributor-group")
    public void handleOrderEvent(Order order) {
        System.out.println("Kafka에서 주문 이벤트 수신: " + order.getId());
        
        // RabbitMQ로 개별 작업 분배
        
        // 1. 재고 확인 작업
        rabbitTemplate.convertAndSend(
            "tasks.exchange",
            "task.inventory.check",
            new InventoryCheckTask(order.getId(), order.getItems())
        );
        
        // 2. 결제 처리 작업 (높은 우선순위)
        rabbitTemplate.convertAndSend(
            "tasks.exchange",
            "task.payment.process",
            new PaymentTask(order.getId(), order.getAmount()),
            message -> {
                message.getMessageProperties().setPriority(10);
                return message;
            }
        );
        
        // 3. 알림 발송 작업
        rabbitTemplate.convertAndSend(
            "notifications.exchange",
            "notification.order.created",
            new NotificationTask(order.getUserId(), "주문이 접수되었습니다")
        );
        
        System.out.println("작업들을 RabbitMQ로 분배 완료");
    }
}

멱등성 구현 예제

@Service
public class IdempotentOrderService {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @KafkaListener(topics = "orders", groupId = "idempotent-group")
    public void processOrder(Order order) {
        String idempotencyKey = "order:processed:" + order.getId();
        
        // Redis를 사용한 중복 체크
        Boolean isNew = redisTemplate.opsForValue()
            .setIfAbsent(idempotencyKey, "processing", 1, TimeUnit.HOURS);
        
        if (Boolean.FALSE.equals(isNew)) {
            System.out.println("이미 처리된 주문: " + order.getId());
            return;  // 중복 처리 방지
        }
        
        try {
            // 실제 주문 처리
            processOrderLogic(order);
            
            // 처리 완료 표시
            redisTemplate.opsForValue().set(
                idempotencyKey, 
                "completed", 
                24, 
                TimeUnit.HOURS
            );
            
            System.out.println("주문 처리 완료: " + order.getId());
            
        } catch (Exception e) {
            // 실패 시 키 삭제 (재시도 가능하도록)
            redisTemplate.delete(idempotencyKey);
            throw e;
        }
    }
}

Choreography 기반 Saga (이벤트 기반)

// 1. 주문 서비스 - Saga 시작
@Service
public class OrderSagaService {
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    public void createOrder(Order order) {
        // 주문 생성 (Pending 상태)
        order.setStatus(OrderStatus.PENDING);
        orderRepository.save(order);
        
        // 재고 예약 이벤트 발행
        kafkaTemplate.send("inventory-events", 
            new ReserveInventoryEvent(order.getId(), order.getItems()));
        
        System.out.println("Saga 시작: 재고 예약 요청");
    }
    
    // 재고 예약 성공 이벤트 수신
    @KafkaListener(topics = "inventory-events", groupId = "order-saga-group")
    public void handleInventoryReserved(InventoryReservedEvent event) {
        System.out.println("재고 예약 성공: " + event.getOrderId());
        
        // 결제 처리 이벤트 발행
        Order order = orderRepository.findById(event.getOrderId()).get();
        kafkaTemplate.send("payment-events",
            new ProcessPaymentEvent(order.getId(), order.getAmount()));
    }
    
    // 결제 성공 이벤트 수신
    @KafkaListener(topics = "payment-events", groupId = "order-saga-group")
    public void handlePaymentProcessed(PaymentProcessedEvent event) {
        System.out.println("결제 성공: " + event.getOrderId());
        
        // 주문 확정
        Order order = orderRepository.findById(event.getOrderId()).get();
        order.setStatus(OrderStatus.CONFIRMED);
        orderRepository.save(order);
        
        // 배송 시작 이벤트 발행
        kafkaTemplate.send("shipping-events",
            new StartShippingEvent(order.getId(), order.getShippingAddress()));
        
        System.out.println("Saga 완료: 주문 확정");
    }
    
    // 보상 트랜잭션 - 재고 예약 실패
    @KafkaListener(topics = "inventory-events", groupId = "order-saga-group")
    public void handleInventoryReservationFailed(InventoryReservationFailedEvent event) {
        System.out.println("재고 예약 실패: " + event.getOrderId());
        
        // 주문 취소
        Order order = orderRepository.findById(event.getOrderId()).get();
        order.setStatus(OrderStatus.CANCELLED);
        order.setCancelReason("재고 부족");
        orderRepository.save(order);
        
        System.out.println("Saga 롤백: 주문 취소");
    }
    
    // 보상 트랜잭션 - 결제 실패
    @KafkaListener(topics = "payment-events", groupId = "order-saga-group")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        System.out.println("결제 실패: " + event.getOrderId());
        
        // 재고 예약 취소 이벤트 발행
        kafkaTemplate.send("inventory-events",
            new CancelInventoryReservationEvent(event.getOrderId()));
        
        // 주문 취소
        Order order = orderRepository.findById(event.getOrderId()).get();
        order.setStatus(OrderStatus.CANCELLED);
        order.setCancelReason("결제 실패");
        orderRepository.save(order);
        
        System.out.println("Saga 롤백: 재고 예약 취소 및 주문 취소");
    }
}

CQRS 구현 예제

// Command Side - 쓰기 모델
@Service
public class OrderCommandService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    @Transactional
    public void createOrder(CreateOrderCommand command) {
        // 주문 생성
        Order order = new Order();
        order.setUserId(command.getUserId());
        order.setItems(command.getItems());
        order.setAmount(calculateAmount(command.getItems()));
        order.setCreatedAt(LocalDateTime.now());
        
        orderRepository.save(order);
        
        // 이벤트 발행 (읽기 모델 업데이트용)
        kafkaTemplate.send("order-events", 
            new OrderCreatedEvent(
                order.getId(),
                order.getUserId(),
                order.getAmount(),
                order.getCreatedAt()
            ));
        
        System.out.println("Command 처리: 주문 생성 완료");
    }
}

// Query Side - 읽기 모델
@Service
public class OrderQueryService {
    
    @Autowired
    private OrderReadModelRepository readModelRepository;
    
    // 이벤트를 받아서 읽기 모델 업데이트
    @KafkaListener(topics = "order-events", groupId = "query-side-group")
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 읽기에 최적화된 모델로 변환
        OrderReadModel readModel = new OrderReadModel();
        readModel.setOrderId(event.getOrderId());
        readModel.setUserId(event.getUserId());
        readModel.setAmount(event.getAmount());
        readModel.setCreatedAt(event.getCreatedAt());
        readModel.setStatus("CREATED");
        
        // 추가 정보 조인 (비정규화)
        User user = userService.getUser(event.getUserId());
        readModel.setUserName(user.getName());
        readModel.setUserEmail(user.getEmail());
        
        readModelRepository.save(readModel);
        
        System.out.println("Query 모델 업데이트: " + event.getOrderId());
    }
    
    // 빠른 조회 (읽기 전용 모델 사용)
    public List<OrderReadModel> getUserOrders(String userId) {
        return readModelRepository.findByUserId(userId);
    }
    
    public OrderStatistics getOrderStatistics(String userId) {
        // 집계된 통계 정보 조회 (미리 계산되어 있음)
        return readModelRepository.getStatistics(userId);
    }
}

조건부 메시지 처리

@Component
public class SmartOrderConsumer {
    
    // 특정 조건의 메시지만 처리
    @KafkaListener(topics = "orders", groupId = "vip-processor-group")
    public void processVipOrders(
        Order order,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
    ) {
        // VIP 고객의 주문만 처리
        if (order.getCustomerTier().equals("VIP")) {
            System.out.println("VIP 주문 우선 처리: " + order.getId());
            processWithHighPriority(order);
        }
    }
    
    // 금액별 다른 처리
    @KafkaListener(topics = "orders", groupId = "amount-based-group")
    public void processOrdersByAmount(Order order) {
        if (order.getAmount() >= 1000000) {
            // 고액 주문 - 추가 검증
            System.out.println("고액 주문 검증: " + order.getId());
            performAdditionalVerification(order);
            notifyManager(order);
        } else if (order.getAmount() >= 100000) {
            // 중간 금액 - 일반 처리
            processNormalOrder(order);
        } else {
            // 소액 주문 - 빠른 처리
            processFastTrack(order);
        }
    }
    
    // 지역별 라우팅
    @RabbitListener(queues = "orders.seoul")
    public void processSeoulOrders(Order order) {
        System.out.println("서울 지역 주문 처리: " + order.getId());
        assignToSeoulWarehouse(order);
    }
    
    @RabbitListener(queues = "orders.busan")
    public void processBusanOrders(Order order) {
        System.out.println("부산 지역 주문 처리: " + order.getId());
        assignToBusanWarehouse(order);
    }
}

메시지 처리 모니터링

@Component
@Slf4j
public class MessageMonitoringAspect {
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    @Around("@annotation(org.springframework.kafka.annotation.KafkaListener)")
    public Object monitorKafkaListener(ProceedingJoinPoint joinPoint) throws Throwable {
        String methodName = joinPoint.getSignature().getName();
        Timer.Sample sample = Timer.start(meterRegistry);
        
        try {
            Object result = joinPoint.proceed();
            
            // 성공 메트릭
            meterRegistry.counter("kafka.message.processed", 
                "method", methodName,
                "status", "success"
            ).increment();
            
            sample.stop(Timer.builder("kafka.message.processing.time")
                .tag("method", methodName)
                .register(meterRegistry));
            
            return result;
            
        } catch (Exception e) {
            // 실패 메트릭
            meterRegistry.counter("kafka.message.processed",
                "method", methodName,
                "status", "failure",
                "error", e.getClass().getSimpleName()
            ).increment();
            
            // 알림 발송
            if (isHighPriorityError(e)) {
                sendAlert("Kafka 메시지 처리 실패: " + methodName, e);
            }
            
            throw e;
        }
    }
    
    // 지연 감지
    @Scheduled(fixedRate = 60000)  // 1분마다
    public void checkConsumerLag() {
        // Kafka Consumer Lag 체크
        Map<TopicPartition, Long> lag = getConsumerLag();
        
        for (Map.Entry<TopicPartition, Long> entry : lag.entrySet()) {
            if (entry.getValue() > 10000) {  // 10,000개 이상 밀렸을 때
                log.warn("높은 Consumer Lag 감지: {} - {}", 
                    entry.getKey(), entry.getValue());
                
                sendAlert("Consumer Lag 경고", 
                    "Topic: " + entry.getKey().topic() + 
                    ", Partition: " + entry.getKey().partition() +
                    ", Lag: " + entry.getValue());
            }
        }
    }
}