[java] 자바 배치 프로세싱에서의 실시간 데이터 유입 방법은 어떻게 되는가?

12 Dec 2023

java

데이터 수신을 위해 메시지 큐에 메시지를 전송하고, 배치 프로세싱 애플리케이션에서는 이러한 메시지를 소비하여 처리하는 방식으로 실시간 데이터 유입이 가능합니다. 메시지 큐를 통해 데이터를 비동기식으로 송수신하고 처리함으로써, 실시간 대규모 데이터 유입에 적합한 확장 가능한 시스템을 구축할 수 있습니다.

아래는 Apache Kafka를 사용한 자바에서의 간단한 데이터 수신 및 처리 예시 코드입니다.

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

public class KafkaConsumerExample {
   public static void main(String[] args) throws Exception {
      String topic = "my_topic";
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test-consumer-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(topic));
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
      }
   }
}

위 코드는 Apache Kafka를 사용하여 데이터를 수신하는 간단한 자바 예시입니다. 메시지 큐를 활용하여 데이터를 비동기적으로 소비하고 처리할 수 있게 됩니다.