[java] 카프카와 자바를 이용한 실시간 로그 처리

16 Nov 2023

java

소개

실시간 로그 처리는 현대 소프트웨어 시스템에서 매우 중요한 요구 사항입니다. 로그는 시스템의 상태와 동작을 기록하고, 문제 진단 및 성능 분석에 필요한 정보를 제공합니다. 이러한 로그를 실시간으로 처리하면, 시스템의 현재 상태를 실시간으로 파악하고 문제를 조기에 탐지할 수 있습니다.

카프카는 아파치 소프트웨어 재단에서 개발된 분산 메시지 큐 시스템으로, 대량의 데이터를 처리하고 여러 컴포넌트 간에 안정적으로 데이터를 전달할 수 있습니다. 자바는 기업용 애플리케이션 개발에서 많이 사용되는 프로그래밍 언어로, 카프카와 함께 사용하면 실시간 로그 처리를 쉽게 구현할 수 있습니다.

이 글에서는 카프카와 자바를 이용하여 실시간 로그를 처리하는 방법에 대해 알아보겠습니다.

카프카와 자바 연동하기

카프카와 자바를 연동하기 위해서는 카프카의 자바 클라이언트를 사용해야 합니다. 자바 클라이언트는 메시지를 카프카에 전송하고, 수신된 메시지를 소비하는 데 사용됩니다. 다음은 자바에서 카프카 메시지를 전송하는 예제 코드입니다.

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

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

        // 카프카 프로듀서 생성
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 카프카 토픽에 메시지 전송
        producer.send(new ProducerRecord<>("my-topic", "Hello, Kafka!"));

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

위 예제에서는 KafkaProducer 클래스를 사용하여 my-topic 토픽에 “Hello, Kafka!” 메시지를 전송하고 있습니다.

카프카로부터 메시지 수신하기

카프카로부터 메시지를 수신하기 위해서는 카프카의 컨슈머를 사용해야 합니다. 컨슈머는 지정된 토픽에서 메시지를 소비하고 처리하는 역할을 합니다. 다음은 자바에서 카프카 메시지를 수신하는 예제 코드입니다.

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 카프카 컨슈머 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 카프카 컨슈머 생성
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 메시지를 가져올 토픽 설정
        consumer.subscribe(Collections.singletonList("my-topic"));

        // 메시지 수신 및 처리
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Received message: " + record.value());
        }

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

위 예제에서는 KafkaConsumer 클래스를 사용하여 “my-topic” 토픽에서 메시지를 수신하고, 받은 메시지를 출력하고 있습니다.

참고 자료

이 글에서는 카프카와 자바를 이용하여 실시간 로그를 처리하는 방법에 대해서 알아보았습니다. 카프카와 자바를 연동하여 실시간으로 로그를 처리하면, 시스템의 상태를 실시간으로 파악하고 문제를 조기에 탐지할 수 있습니다. 카프카와 자바를 사용하여 실시간 로그 처리를 시작해보세요!