CQRS (Command Query Responsibility Segregation) 패턴을 Kafka와 함께 Java (Spring Boot 기반)로 구현하는 샘플

# application.yml (Command Service 또는 Query Service)

spring:
  kafka:
    # Kafka 서버 주소
    bootstrap-servers: localhost:9092
    
    # Producer 설정 (Command Service)
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # JSON 직렬화
      
    # Consumer 설정 (Query Service)
    consumer:
      group-id: product-group # 소비자 그룹 ID
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer # JSON 역직렬화
      auto-offset-reset: earliest

// ProductCreatedEvent.java
public class ProductCreatedEvent {
    private String id;
    private String name;
    private double price;

    // Getters, Setters, Constructors...
}

// ProductEventProducer.java

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class ProductEventProducer {

    private final KafkaTemplate<String, ProductCreatedEvent> kafkaTemplate;
    private static final String TOPIC = "product-events";

    // Lombok의 @RequiredArgsConstructor 등으로 주입 가능
    public ProductEventProducer(KafkaTemplate<String, ProductCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendProductCreatedEvent(ProductCreatedEvent event) {
        System.out.println("[COMMAND] 이벤트 발행: " + event.getName());
        // Kafka에 이벤트 발행 (Key: ID, Value: Event 객체)
        kafkaTemplate.send(TOPIC, event.getId(), event);
    }
}

// ProductCommandService.java (Command Service의 핵심 비즈니스 로직)

@Service
public class ProductCommandService {

    private final ProductRepository commandRepository; // 쓰기 전용 DB Repository
    private final ProductEventProducer eventProducer;

    // Lombok의 @RequiredArgsConstructor 등으로 주입 가능
    public ProductCommandService(ProductRepository commandRepository, ProductEventProducer eventProducer) {
        this.commandRepository = commandRepository;
        this.eventProducer = eventProducer;
    }

    // Command: 새로운 상품 생성 요청 처리
    public Product createProduct(Product newProduct) {
        // 1. 쓰기 전용 DB에 저장 (트랜잭션 보장)
        Product savedProduct = commandRepository.save(newProduct);

        // 2. Event 객체 생성
        ProductCreatedEvent event = new ProductCreatedEvent(
            savedProduct.getId(), 
            savedProduct.getName(), 
            savedProduct.getPrice()
        );

        // 3. Kafka에 이벤트 발행
        eventProducer.sendProductCreatedEvent(event);
        
        System.out.println("[COMMAND] DB 저장 및 이벤트 발행 완료.");
        return savedProduct;
    }
}

// ProductEventListener.java (Query Service의 Event Handler)

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class ProductEventListener {

    private final ProductViewRepository queryRepository; // 읽기 전용 DB Repository

    // Lombok의 @RequiredArgsConstructor 등으로 주입 가능
    public ProductEventListener(ProductViewRepository queryRepository) {
        this.queryRepository = queryRepository;
    }

    @KafkaListener(topics = "product-events", groupId = "product-group")
    public void handleProductCreated(ProductCreatedEvent event) {
        System.out.println("[QUERY] Kafka 이벤트 수신: " + event.getName());
        
        // 1. Read Model (조회용 엔티티) 생성 또는 업데이트
        ProductView productView = new ProductView();
        productView.setId(event.getId());
        productView.setName(event.getName());
        productView.setPrice(event.getPrice());
        
        // 2. 읽기 전용 DB에 저장 (데이터 동기화)
        queryRepository.save(productView);

        System.out.println("[QUERY] Read DB 업데이트 완료: " + productView.getName());
    }
}

// ProductQueryService.java

@Service
public class ProductQueryService {

    private final ProductViewRepository queryRepository; // 읽기 전용 DB Repository

    // Lombok의 @RequiredArgsConstructor 등으로 주입 가능
    public ProductQueryService(ProductViewRepository queryRepository) {
        this.queryRepository = queryRepository;
    }

    // Query: 상품 ID로 조회 요청 처리
    public ProductView getProductById(String id) {
        System.out.println("[QUERY] Read DB에서 상품 조회: " + id);
        // 읽기 전용 DB에서 조회
        return queryRepository.findById(id).orElse(null);
    }
}