[java] 자바로 카프카에서 메시지 읽기

16 Nov 2023

java

카프카(Kafka)는 대용량의 실시간 데이터를 처리하기 위한 분산형 메시지 큐 시스템입니다. 이번에는 자바 언어를 사용하여 카프카에서 메시지를 읽는 방법에 대해 알아보겠습니다.

카프카 클라이언트 라이브러리 추가

먼저 카프카를 사용하기 위해 카프카 클라이언트 라이브러리를 프로젝트에 추가해야 합니다. Maven을 사용한다면 아래와 같이 pom.xml 파일에 의존성을 추가합니다.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

Gradle을 사용한다면 아래와 같이 build.gradle 파일에 의존성을 추가합니다.

implementation 'org.apache.kafka:kafka-clients:2.8.0'

카프카에서 메시지 읽기

카프카에서 메시지를 읽기 위해서는 KafkaConsumer를 사용해야 합니다. 아래는 카프카 클러스터의 bootstrap.servers 설정과 토픽 이름을 지정하여 KafkaConsumer 객체를 생성하는 예제입니다.

import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaMessageReader {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received message: key = %s, value = %s\n", record.key(), record.value());
            }
        }
    }
}

위 예제에서는 bootstrap.servers, group.id 등의 카프카 설정을 지정하고, subscribe 메소드를 통해 읽을 토픽을 지정합니다. 그리고 poll 메소드를 통해 주기적으로 메시지를 받아옵니다. 받아온 메시지는 ConsumerRecord 객체로 처리할 수 있습니다.

실행 및 메시지 확인

위의 코드를 실행하면 지정한 토픽에 있는 메시지를 읽어오게 됩니다. poll 메소드의 매개변수로 지정한 시간마다 새로운 메시지를 확인하며 처리합니다. ConsumerRecord 객체를 통해 읽은 메시지의 키(key)와 값(value)을 확인할 수 있습니다.

자바로 카프카에서 메시지를 읽는 방법에 대해 알아보았습니다. 카프카를 사용하여 대용량의 데이터를 실시간으로 처리하기 위해서는 카프카의 기본 개념에 대해 이해하는 것이 중요합니다. 자세한 내용은 카프카 문서를 참고하시기 바랍니다.