[java] 카프카와 자바를 이용한 이벤트 소싱 구현하기

16 Nov 2023

java

소개

이벤트 소싱(Event Sourcing)은 시스템의 상태를 이벤트(상태 변경 사항)의 연속으로 기록하고 저장하는 패턴입니다. 이 패턴은 시스템의 변화를 추적하고 상태를 리플레이하여 현재 상태를 재구축하는 데 사용됩니다. 이 패턴을 구현하는 도구들 중 하나로 아파치 카프카(Apache Kafka)가 있습니다.

이 튜토리얼에서는 카프카와 자바를 사용하여 이벤트 소싱을 구현하는 방법을 알아보겠습니다.

카프카 소개

카프카는 대용량 실시간 스트리밍 플랫폼으로, 수많은 프로듀서와 컨슈머가 고성능으로 메세지를 발행하고 구독할 수 있도록 지원합니다. 카프카는 분산 아키텍처로 설계되어 있어 데이터의 안정성과 확장성을 보장합니다. 또한, 메세지를 영구적으로 보관하기 때문에 데이터 손실을 최소화할 수 있습니다.

이벤트 소싱 구현하기

카프카 설정하기

먼저, 카프카를 사용하기 위해 카프카 클러스터를 설치하고 설정해야 합니다. 설치 및 설정 방법에 대한 자세한 내용은 카프카 공식 문서를 참조하십시오.

이벤트 발생시키기

이벤트 소싱을 구현하기 위해 먼저 이벤트를 발생시켜야 합니다. 이벤트는 상태 변경에 대한 정보를 가지고 있으며, 그 정보를 읽고 처리하는 로직을 가지고 있어야 합니다. 자바로 이벤트를 발생시키는 예제 코드를 작성해보겠습니다.

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

public class EventProducer {

    private final static String TOPIC = "events";

    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);

        // 이벤트 발생
        for (int i = 1; i <= 10; i++) {
            String key = "event-" + i;
            String value = "Event " + i;

            // 카프카로 이벤트 전송
            producer.send(new ProducerRecord<>(TOPIC, key, value));
        }

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

위의 코드는 events라는 토픽에 1부터 10까지의 이벤트를 발생시키는 예제입니다.

이벤트 처리하기

이벤트를 발생시켰다면, 해당 이벤트를 읽고 처리해야 합니다. 이벤트를 읽는 컨슈머를 구현하여 처리하는 예제 코드를 작성해보겠습니다.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.*;
import java.util.*;

public class EventConsumer {

    private final static String TOPIC = "events";
    private final static String GROUP_ID = "event-consumer-group";

    public static void main(String[] args) {

        // 카프카 컨슈머 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", GROUP_ID);
        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(TOPIC));

        // 이벤트 처리
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received event: key = %s, value = %s%n", record.key(), record.value());
            }
        }
    }
}

위의 코드는 events 토픽을 구독하고, 수신된 이벤트를 출력하는 예제입니다.

결론

이벤트 소싱은 시스템의 변경 사항을 추적하고 상태를 재구축하는 데 유용한 패턴입니다. 카프카는 이벤트 소싱을 구현하기 위한 강력한 도구로 활용할 수 있습니다. 이 튜토리얼에서는 카프카와 자바를 사용한 이벤트 소싱의 기본 개념과 구현 방법을 살펴보았습니다.

더 자세한 내용은 아파치 카프카 공식 문서와 관련 서적을 참고하시기 바랍니다.