[java] 카프카와 자바를 사용한 멀티스레드 처리

16 Nov 2023

java

카프카(Kafka)는 대용량의 실시간 데이터 스트리밍을 처리하기 위한 분산 메시징 시스템입니다. 자바는 대표적인 프로그래밍 언어 중 하나로, 카프카와 결합하여 멀티스레드 처리를 수행할 수 있습니다. 이번 글에서는 카프카와 자바를 함께 사용하여 멀티스레드 처리를 구현하는 방법에 대해 알아보겠습니다.

1. 카프카의 스레드 모델

카프카는 멀티스레드 환경에서 안전하게 사용할 수 있는 특정한 스레드 모델을 제공합니다. 카프카는 여러 개의 스레드를 사용하여 메시지를 소비하고 처리하는 과정을 동시에 진행할 수 있습니다. 이러한 스레드 모델은 높은 성능과 확장성을 제공하며, 실시간 데이터 처리에 적합합니다.

2. 자바를 사용한 카프카 멀티스레드 처리

자바에서 카프카를 사용하기 위해서는 먼저 카프카 클라이언트를 설치하고 해당 라이브러리를 프로젝트에 추가해야 합니다. 그리고 아래와 같이 자바 코드를 작성하여 멀티스레드 처리를 구현할 수 있습니다.

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaMultiThreadProcessing {

    private static final String TOPIC_NAME = "my_topic";
    private static final String KAFKA_SERVER = "localhost:9092";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        ExecutorService executor = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 3; i++) {
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
            consumer.subscribe(Collections.singletonList(TOPIC_NAME));
            executor.execute(new KafkaMessageProcessor(consumer));
        }
    }

    private static class KafkaMessageProcessor implements Runnable {
        private final KafkaConsumer<String, String> consumer;

        public KafkaMessageProcessor(KafkaConsumer<String, String> consumer) {
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        // 메시지 처리 로직 작성
                    }
                    consumer.commitAsync();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                consumer.close();
            }
        }
    }
}

위의 예제 코드는 카프카로부터 메시지를 소비하고 처리하는 컨슈머를 멀티스레드로 실행하는 예시입니다. KafkaMessageProcessor는 실제로 메시지를 처리하는 로직을 구현하는 클래스로, 원하는 작업을 수행할 수 있습니다.

3. 참고 자료

카프카 공식 문서: https://kafka.apache.org/documentation/
카프카 자바 클라이언트 문서: https://kafka.apache.org/documentation/#javaapi

위의 예시 코드와 참고 자료를 통해 카프카와 자바를 함께 사용하여 멀티스레드 처리를 구현하는 방법을 알아보았습니다. 카프카를 사용하여 대용량의 데이터를 실시간으로 처리해야 하는 경우, 이러한 멀티스레드 처리 방법은 매우 유용할 수 있습니다.