[java] 자바로 카프카와 함께하는 스트리밍 애플리케이션 개발

16 Nov 2023

java

스트리밍 애플리케이션은 실시간으로 데이터를 처리하고 결과를 생성하는데 사용되는 애플리케이션입니다. 이러한 애플리케이션은 대용량의 데이터를 다루는데 효율적이며, IoT, 로그 분석, 실시간 분석 등 다양한 분야에서 사용됩니다.

카프카는 대용량의 실시간 데이터를 처리하기 위한 분산형 스트리밍 플랫폼입니다. 카프카는 메세지 큐와 같은 기능을 제공하여 데이터를 저장하고 전달하는 역할을 합니다. 자바는 카프카와 함께 사용하기 위한 강력한 언어입니다.

이번 글에서는 자바로 카프카와 함께하는 스트리밍 애플리케이션을 개발하는 방법에 대해 알아보겠습니다.

카프카 설치하기

먼저, 카프카를 설치해야 합니다. 카프카는 Apache 프로젝트로 제공되며, 카프카 공식 웹사이트에서 다운로드 받을 수 있습니다. 설치 방법은 공식 문서를 참조하시기 바랍니다.

카프카 프로듀서 구현하기

카프카 프로듀서는 데이터를 생성하고 카프카 클러스터로 전송하는 역할을 합니다. 자바에서는 KafkaProducer 클래스를 사용하여 프로듀서를 구현할 수 있습니다.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class KafkaStreamingProducer {
    public static void main(String[] args) {
        // 카프카 프로듀서 설정
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 메세지 전송
        String topic = "streaming_topic";
        String message = "Hello, Kafka!";
        producer.send(new ProducerRecord<>(topic, message));

        // 프로듀서 종료
        producer.close();
    }
}

위 코드에서는 먼저 KafkaProducer 객체를 생성한 후, 필요한 설정을 추가합니다. bootstrap.servers는 카프카 클러스터의 주소를 설정하는 부분입니다. key.serializer와 value.serializer는 메세지의 키와 값에 대한 직렬화 방식을 설정하는 부분입니다.

메세지를 전송할 토픽과 메세지 내용을 설정한 후, producer.send() 메서드를 호출하여 메세지를 전송합니다. 마지막으로, producer.close()를 호출하여 프로듀서를 종료합니다.

카프카 컨슈머 구현하기

카프카 컨슈머는 카프카 클러스터로부터 데이터를 가져와서 처리하는 역할을 합니다. 자바에서는 KafkaConsumer 클래스를 사용하여 컨슈머를 구현할 수 있습니다.

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;

public class KafkaStreamingConsumer {
    public static void main(String[] args) {
        // 카프카 컨슈머 설정
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer", StringDeserializer.class.getName());
        properties.put("group.id", "my_consumer_group");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        // 컨슈머에 토픽 할당
        String topic = "streaming_topic";
        consumer.subscribe(Collections.singletonList(topic));

        // 메세지 수신
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            records.forEach(record -> System.out.println("Received message: " + record.value()));
        }

        // 컨슈머 종료
        consumer.close();
    }
}

위 코드에서는 KafkaConsumer 객체를 생성한 후, 필요한 설정을 추가합니다. bootstrap.servers는 카프카 클러스터의 주소를 설정하는 부분입니다. key.deserializer와 value.deserializer는 메세지의 키와 값에 대한 역직렬화 방식을 설정하는 부분입니다. group.id는 컨슈머 그룹의 식별자를 설정하는 부분입니다.

consumer.subscribe() 메서드를 사용하여 컨슈머에 토픽을 할당합니다. consumer.poll() 메서드를 호출하여 메세지를 수신하고, 수신된 메세지를 처리하는 로직을 작성합니다.

마지막으로, consumer.close()를 호출하여 컨슈머를 종료합니다.

결론

이번 글에서는 자바로 카프카와 함께하는 스트리밍 애플리케이션을 개발하는 방법에 대해 알아보았습니다. 카프카의 프로듀서와 컨슈머를 자바로 구현하고, 메세지를 전송하고 수신하는 방법을 살펴보았습니다.

카프카를 사용한 스트리밍 애플리케이션은 대용량 데이터를 실시간으로 처리하는 데 매우 효과적입니다. 자바를 통해 카프카와 함께하는 스트리밍 애플리케이션을 개발하여 다양한 분야에서 이점을 경험해보세요.