Kafka와 Spring Boot 연동

메시지큐 Kafka

 

개요

이번에는 Kafka와 Spring Boot를 연동해보려 합니다.

Spring Boot에 메시지를 전송하는 Producer, 메시지를 받는 Consumer의 코드를 추가하여 Kafka를 통해 메시지를 주고 받게 해볼 겁니다.

이 글은 Kafka Consumer 구현, Kafka Producer 구현, 테스트 순으로 정리했습니다.

 

 

 

시작하기 전

Spring Boot에 Producer, Consumer 코드를 추가하기 전에 Kafka에 Topic을 하나 생성해야 합니다. 

또한 테스트를 하려면 ZooKeeper와 Kafka를 실행시켜야 합니다. 

ZooKeeper와 Kafka를 설치하고 실행하는 글은 이전에 정리한 글을 참조하시면 도움이 될 것 입니다.

https://cladren123.tistory.com/190

 

 

 

Kafka Consumer

Consumer는 Kafka의 토픽을 구독하여 토픽에 메시지를 가져오는 역할을 합니다.

 

 

의존성 추가

implementation 'org.springframework.kafka:spring-kafka'

 

 

Kafka Consumer Config 생성

 

KafkaConsumerConfig 파일을 만들어 KafkaConsumer에 대한 설정을 저장합니다.

이 파일에는 ConsumerFactory(), KafkaListenerContainerFactory() 라는 2가지 Bean을 동륵합니다.

 

ConsumerFactory() 는 Kafka Topic의 구독 위치와 직렬화 등에 대한 설정을 합니다. 

KafkaListenerContainerFactory() 는 ConsumerFactory()의 설정 정보를 받아와 Kafka Listener Container를 만듭니다.

이 Kafka Listener Container는 토픽에 변경 사항을 감지하여 정보를 받아오는 리스너 생성에 쓰입니다.

서비스단에 메소드에 @KafkaListenr를 설정하면 Kafka Listener Container에서 Listener를 생성하여  카프카 토픽에 변경 사항을 감지하여 해당 메소드를 실행시킵니다.

package com.dotd.product.practice.consumer;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    // 접속하고자 하는 곳에 정보
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {

        // Map을 이용해 Key, Value 형태로 Kafka Consumer에 필요한 설정을 저장합니다
        Map<String, Object> properties = new HashMap<>();

        // Kafka Broker 위치 저장합니다.
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        // Kafka Consumer 그룹의 ID 지정, 같은 그룹 ID를 가진 컨슈머들은 메시지를 공유하여 처리합니다.
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroupId");

        // Kafka 메시지의 키, 값을 어떻게 역직렬화 할지 설정합니다.
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 위의 설정한 properties 값을 바탕으로 ConsumerFactory를 생성합니다.
        return new DefaultKafkaConsumerFactory<>(properties);
    }

    // 위에서 정의한 consumerFactory를 사용해 Kafka Listener Containers를 만들어 줍니다.
    // 이 설정을 사용하여 해당 토픽의 메시지를 가져옵니다.
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory
                = new ConcurrentKafkaListenerContainerFactory<>();

        kafkaListenerContainerFactory.setConsumerFactory(consumerFactory());

        return kafkaListenerContainerFactory;
    }

}

 

 

 

Kafka Consumer 생성

@KafkaListener(topics = "") 의 어노테이션을 통해 카프카 브로커에 토픽을 구독합니다.

이 KafkaListener는 위의 KafkaConsumerConfig에서 생성된 것 입니다. 

구독한 토픽에 메시지가 들어오면 해당 메시지를 가지고 와서 메소드의 로직을 수행합니다. 

이 때, 서버끼리 통신할 때는 메시지를 직렬화 합니다.(저는 JSON 형태로 직렬화 했습니다.)

그러므로 메시지를 읽을 때는 역직렬화 과정을 수행합니다. 이 작업을 ObjectMapper가 수행합니다.

package com.dotd.product.practice.consumer;


import com.dotd.product.entity.Product;
import com.dotd.product.repository.ProductRepository;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Service
@Slf4j
@RequiredArgsConstructor // 생성자 자동 생성
public class KafkaConsumer {

    private final ProductRepository productRepository;

    // 등록해놓은 카프카 토픽에 데이터가 들어오면 실행되는 로직
    @KafkaListener(topics = "example-product-topic")
    public void updatePrice(String kafkaMessage) {
        log.info("Kafka Message : -> " + kafkaMessage);

        Map<Object, Object> map = new HashMap<>();

        // ObjectMapper를 통해 카프카로부터 받은 메시지를 Map 구조로 변환합니다.
        ObjectMapper mapper = new ObjectMapper();
        try {
            map = mapper.readValue(kafkaMessage, new TypeReference<Map<Object, Object>>() {});
        } catch (JsonProcessingException ex) {
            ex.printStackTrace();
        }

        log.info("결과 확인" + " " + map.get("name") + " " + map.get("price"));

    }

}

 

 

 

 

 

Kafka Producer

Producer 는 지정된 토픽에 메시지를 보내는 역할을 합니다.

 

 

의존성 추가

implementation 'org.springframework.kafka:spring-kafka'

 

 

Kafka Producer Config 생성

Kafka Producer Config 파일을 만들어 설정 정보를 저장합니다.

이 설정 파일은 ProducerFactory(), KafkaTemplate() 두 개의 빈을 등록합니다.

 

ProducerFactory() 는 Producer를 만들기 위한 설정 정보들을 저장합니다. 

Kafka Broker의 주소, 직렬화 방법 등을 설정합니다.

 

KafkaTemplate() 은 위의 ProducerFactory를 사용하여 Kafka Template를 생성합니다.

Kafka Template은 메시지를 전송하기 위한 주요 메소드들을 제공합니다. 

 

package com.dotd.product.practice.producer;



import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaProducerConfig {

    // Kafka Producer에 필요한 설정들을 저장
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> properties = new HashMap<>();

        // kafka broker 위치 지정
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        // kafka 메시지의 키와 값을 어떻게 직렬화하는지 지정
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // 위에서 설정한 properties를 사용해 ProducerFactory를 생성 
        return new DefaultKafkaProducerFactory<>(properties);
    }

    // KafkaTemplate은 메시지를 전송하기 위한 주요 메소드들을 제공
    // ProducerFactory를 사용하여 Kafka Template을 생성
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

 

 

 

KafkaProducer 생성

KafkaProducer는 메시지를 JSON으로 변환하여 원하는 Topic에 메시지를 보냅니다.

이 때 ObjectMapper를 통해 객체를 JSON 형태로 직렬화 합니다.

그 후 KafkaTemplate를 통해 원하는 토픽과 JSON된 직렬화된 객체를 넣어 카프카에 보냅니다.

package com.dotd.product.practice.producer;


import com.dotd.product.dto.ProductRegistDto;
import com.dotd.product.dto.ProductResponseDto;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class KafkaProducer {

    // Kafka 메시지를 전송하기 위한 템플릿 (문자열 키, 값)
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // kafka에 메시지 전송
    public void send(String topic, ProductRegistDto dto) {

        // ObjectMapper를 통해 ProductRegistDto 객체를 JSON 문자열로 변환
        ObjectMapper mapper = new ObjectMapper();
        String jsonInString = "";
        try {
            jsonInString = mapper.writeValueAsString(dto);
        } catch (JsonProcessingException ex) {
            ex.printStackTrace();
        }

        // JSON 문자열을 해당 topic에 전달 
        kafkaTemplate.send(topic, jsonInString);

        log.info("Kafka Producer sent data from the product micro service " + dto);

    }

}

 

 

 

테스트

저는 한 Spring Boot에 Consumer와 Producer를 만들어 테스트를 진행했습니다.

다른 Spring Boot에 Consumer, Producer 를 각각 만들어도 테스트하셔도 됩니다. 

 

 

KafkaProducerController 생성

위에서 만든 KafkaProducer의 send 메시지를 실행할 Controller 입니다.

필자는 테스트를 위해 카프카에 example-product-topic을 미리 만들었습니다. 

package com.dotd.product.practice.producer;


import com.dotd.product.dto.ProductRegistDto;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/kafka")
@RequiredArgsConstructor
public class KafkaProducerController {
    
    private final KafkaProducer kafkaProducer;
    
    // kafka 메시지를 보내는 함수 
    @PostMapping("/send")
    public ResponseEntity<?> send(@RequestBody ProductRegistDto dto) {
        kafkaProducer.send("example-product-topic", dto);
        return ResponseEntity.ok("ok");
    }
}

 

그리고 테스트용으로 사용한 ProductRegistDto 의 코드는 다음과 같습니다.

package com.dotd.product.dto;


import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ProductRegistDto {

    private String sellerId;

    private Integer categoryId;

    private String code;
    private String name;
    private Integer price;
    private String description;

}

 

위의 Controller를 만들어 PostMan으로 요청을 보냈습니다.

다음은 그 결과 로그입니다.

 

Kafka의 Producer, Consumer 로그

 

 

Kafka Producer를 통해 보낸 메시지가 Kafka Consumer 가 받은 것을 확인할 수 있었습니다. 

 

 

 

 

마무리

이번에는 Spring Boot와 Kafka 를 연동하여 서버끼리 메시지큐를 이용한 통신을 구현해 보았습니다.

Kafka는 여러 개의 서비스로 나누어진 MSA 환경에서 중요한 역할을 합니다.

여러 개의 서버가 토픽을 구독할 수 있는 점에서 여러 개의 DB가 있는 MSA 환경에서 데이터의 무결성을 지켜주는 중요한 역할을 합니다.

다음에는 Kafka Connect를 사용해 보겠습니다. 감사합니다.