[java] Java에서 Apache Avro를 이용한 메시지 큐 구현

28 Nov 2023

java

본 포스트에서는 Java에서 Apache Avro를 이용하여 메시지 큐를 구현하는 방법에 대해 알아보겠습니다.

Apache Avro란?

Apache Avro는 데이터 직렬화 플랫폼으로, 대규모 데이터 저장 및 처리를 위한 고성능 솔루션입니다. Avro는 사용하기 쉽고 효율적이며 자바, C++, Python 등 다양한 언어를 지원합니다. Avro를 사용하면 데이터 스키마를 정의하고, 이를 기반으로 데이터를 직렬화하고 역직렬화할 수 있습니다.

Java에서 Apache Avro 사용하기

Apache Avro를 Java에서 사용하기 위해서는 먼저 Avro 라이브러리를 프로젝트에 추가해야 합니다. Maven을 사용한다면 pom.xml 파일에 다음 의존성을 추가할 수 있습니다:

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.10.2</version>
</dependency>

의존성을 추가한 후, Avro의 기능을 사용할 수 있습니다.

메시지 큐 구현하기

이제 Apache Avro를 이용하여 메시지 큐를 구현해보겠습니다. Java에서 Apache Kafka를 사용해보기로 하겠습니다.

먼저, Avro 스키마로 메시지 형식을 정의합니다. 예를 들어, Message라는 스키마로 key와 value 필드를 가지는 메시지를 정의할 수 있습니다:

{
    "type": "record",
    "name": "Message",
    "fields": [
        {"name": "key", "type": "string"},
        {"name": "value", "type": "string"}
    ]
}

메시지를 Producer에서 생성하여 Kafka에 전송하고, Consumer에서 메시지를 소비합니다. 이를 위해 Kafka의 Producer와 Consumer 클래스를 사용합니다. 먼저, Producer 클래스에서는 Avro 메시지를 생성한 후 Kafka에 전송하는 코드를 작성합니다:

import org.apache.kafka.clients.producer.*;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // Avro 스키마 정의
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse("{\"type\":\"record\",\"name\":\"Message\",\"fields\":[{\"name\":\"key\",\"type\":\"string\"},{\"name\":\"value\",\"type\":\"string\"}]}");

        // Kafka Producer 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
        props.put("schema.registry.url", "http://localhost:8081");

        // Kafka Producer 생성
        KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);

        // Avro 메시지 생성
        GenericRecord message = new GenericData.Record(schema);
        message.put("key", "키");
        message.put("value", "값");

        // Kafka에 메시지 전송
        ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("mytopic", message);
        producer.send(record);

        // Kafka Producer 종료
        producer.close();
    }
}

다음으로, Consumer 클래스에서 Kafka로부터 메시지를 수신하고 처리하는 코드를 작성합니다:

import org.apache.kafka.clients.consumer.*;
import org.apache.avro.generic.GenericRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // Kafka Consumer 설정
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        props.put("schema.registry.url", "http://localhost:8081");
        props.put("group.id", "mygroup");

        // Kafka Consumer 생성
        KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);

        // 구독 시작
        consumer.subscribe(Collections.singletonList("mytopic"));

        // 메시지를 수신하고 처리
        while (true) {
            ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, GenericRecord> record : records) {
                // 메시지 처리 로직 작성
                GenericRecord message = record.value();
                String key = message.get("key").toString();
                String value = message.get("value").toString();
                System.out.println("Received message: key = " + key + ", value = " + value);
            }
        }

        // Kafka Consumer 종료
        consumer.close();
    }
}

위의 코드에서는 bootstrap.servers를 적절한 Kafka 브로커의 호스트와 포트로 설정해야 합니다. 또한, schema.registry.url을 Avro 스키마 레지스트리의 URL로 설정해야 합니다.

결론

이제 Java에서 Apache Avro를 사용하여 메시지 큐를 구현하는 방법에 대해 알아보았습니다. Avro를 사용하면 간편한 데이터 직렬화 및 역직렬화를 통해 메시지 큐를 구축할 수 있습니다. Avro의 풍부한 기능을 활용하여 다양한 데이터 처리 작업을 수행할 수 있습니다.

더 자세한 내용은 Apache Avro 공식 문서를 참조하십시오.

목차

Apache Avro란?

Java에서 Apache Avro 사용하기

메시지 큐 구현하기

결론