(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/