(SpringBoot) (Kotlin) (Kafka) 기본 Kafka 소비자

안녕하세요. 저는 홈디벨로퍼 금입니다.

최근에는 Kafka가 발행한 메시지를 외부 시스템에서 수신하는 기능을 개발했습니다.
(지금까지는 SQS SQS-SNS를 주로 사용했는데 회사 정책 변경으로 인해 kafka로 마이그레이션 중입니다.)

Spring Boot를 사용하지 않는 상황에서는 Bean을 등록하여 사용해야 합니다. 하지만 요즘은 대부분의 시스템이 Spring Boot로 구축되기 때문에,

설정이 쉽고 옵션을 변경하기 쉽습니다. 나는 주로 @KafkaListener를 사용합니다. 이 기사에서는 @KafkaListener를 사용하여 구현을 요약합니다.

Kotlin + SpringBoot + kafka 소비자의 기본 사용법

가장 먼저 @KafkaListener를 사용하여 기본 소비자를 구현해 봅시다.

– 먼저 spring-kafka 라이브러리 추가
* Spring Boot를 사용할 때(프로젝트를 생성하기 위해 start.spring.io를 사용하지 않음) 버전을 생략하면 Boot는 자동으로 Boot 버전과 호환되는 올바른 버전을 가져옵니다.

compile 'org.springframework.kafka:spring-kafka'

– application.yml 설정

기본 리스너 컨테이너를 사용하는 경우 yaml 구성 파일에서 소비자 및 리스너 옵션을 설정할 수 있으며 옵션 값은 애플리케이션 실행 시 자동으로 재정의됩니다. Spring Kafka에서 사용하는 옵션에 대한 설명은 다음과 같습니다. 스프링 공식 홈페이지에서 찾을 수 있습니다

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092 //kafka 메시지를 송신하는 서버 URL
    listener:
      type: BATCH
      ack-mode: MANUAL_IMMEDIATE

* 소비자 옵션

* 리스너 옵션

– @KafkaListener 사용

@Component
class ConsumerTest {
    @KafkaListener(topics = "topic-name", groupId = "group-01")
    public void onMessage(kafkaSchema: kafkaSchema, Acknowledgment ack) {
        //kafkaSchema 사용 처리
        ack.acknowledge(); // 커밋 수행
    }
}

data class KafkaSchema(
    val name: String,
    val columnNames: List<String> = listOf() 
)

리스너 유형 및 커밋 모드(AcksMode)

이전에 기본 소비자를 구현했습니다. @kafkaListener 생성 방식의 파라미터는 리스너 유형과 커밋 방식에 따라 변경할 수 있습니다. 그리고 커밋 모드(AcksMode)에 따라 커밋 방식을 선택할 수 있습니다.

– 리스너 유형 및 리스너 유형

• 레코드 리스너(MessageListener): 1개의 레코드만 처리합니다. (Spring Kafka 소비자의 기본 리스너 유형)
• Batch Listener(BatchMessageListener): 한 번에 여러 레코드를 처리합니다.

– AcksMode 유형

* spring-kafka Consumer의 기본 AckMode는 BATCH이고 enable.auto.commit 옵션은 false로 설정되어 있습니다.

참조
– https://docs.spring.io/spring-kafka/