Kafka를 이용해 Producer/Consumer 맛보기! - Configuration

토이 프로젝트를 진행하던 도중 API Gateway를 도입하게 되면서 기존 회원 API 서버 내 회원 서비스에 구현되있는 소셜 로그인 및 회원 가입 기능과 JWT 엑세스 토큰 및 리프레시 토큰을 발급해주는 기능을 인증 서버내 회원 서비스로 따로 빼게 되었다.

 

이 때, 발생한 문제점이 있다.

현재 토이 프로젝트는 소셜 로그인을 이용하지만 추가적인 정보가 필요하기 때문에 소셜 로그인 이후 해당 서비스에 처음 로그인 한 경우 (서비스에 가입되지 않은 상태)라면 추가적인 회원 정보 및 추천인과 관심 OTT 서비스를 등록해야 한다.

 

여기서 추가적인 회원 정보는 인증 서버 내 회원 서비스에서 사용하는 간단한 회원 가입 로직을 이용하면 되지만, 가입 시 입력한 추천인에게 포인트를 적립(Optional)해주거나 회원이 관심있는 OTT 서비스를 등록(Optional)하는 로직을 인증 서버의 역할로 가져가기엔 회원 API 서버와 인증서버간 경계 구분이 모호해진다는 결론이 나왔다.

 

이러한 문제점을 어떻게 해결할지 고민하던 도중 Kafka를 이용해 인증 서버 에서 회원 API 서버로 간단한 메세지 데이터를 전달해 포인트 적립 로직과 관심 OTT 서비스 등록 로직을 회원 API 서버 내 포인트 서비스와 관심 OTT 서비스에서 처리하도록 구현하기로 결정했다.

⛳ 로드맵

1. Kafka 의존성 추가 및 Producer/Consumer Java Configuration 설정
2. Producer/Consumer 구현 및 기존 로직과 비교

기본 설정

현재 프로젝트는 Gradle을 사용한다.

build.gradle 파일에 Spring kafka 의존성을 추가한다.

// Kafka 의존성 추가
dependencies {
	implementation 'org.springframework.kafka:spring-kafka'
	testImplementation 'org.springframework.kafka:spring-kafka-test'
}

Kafka관련한 설정을 application.properties로 관리하지 않고 Java Configuration를 이용해서 필요한 정보들만 설정했다.

application.properties는 Kafka의 Bootstrap Server의 host와 port를 @Value로 받아오기 위해 간단하게 작성해주었다.

# Kafka의 Bootstrap Server는 일반적으로 9092번 포트를 이용한다. 
kafka.host=localhost
kafka.port=9092 

Producer Configuration

인증 서버에서 회원 API 서버로 메세지를 보내기 때문에 Producer는 인증 서버, Consumer는 회원 API 서버가 된다.

// 인증 서버 
package com.dev.nbbang.auth.global.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
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 {
    private final String host;
    private final String port;

    // 카프라 Bootstrap Server Config 프로퍼티 파일에서 가져오기
    public KafkaProducerConfig(@Value("${kafka.host}") String host, @Value("${kafka.port}") String port) {
        this.host = host;
        this.port = port;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        // 프로퍼티 설정
        Map<String, Object> properties = new HashMap<>();

        // Bootstrap Server는 일반적으로 9092번 포트를 사용한다.
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, host + ":" + port);

        // Key, Value Serialize 프로퍼티 설정
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return new DefaultKafkaProducerFactory<>(properties);
    }
		
		// KafakaTemplate을 이용해 데이터 실제 전송, 설정한 프로퍼티를 이용
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

Consumer Configuration

메세지를 전달 받는 회원 API 서버도 동일하게 Java Configuration 설정을 해준다.

package com.dev.nbbang.member.global.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
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 {
    private final String host;
    private final String port;

    public KafkaConsumerConfig(@Value("${kafka.host}") String host, @Value("${kafka.port}") String port) {
        this.host = host;
        this.port = port;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> properties = new HashMap<>();

        //Bootstrap Server 설정
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, host + ":" + port);

        // Key, Value Deserialize 프로퍼티 설정
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 그룹 아이디 지정
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group-id");
        return new DefaultKafkaConsumerFactory<>(properties);
    }

		
		// @KafkaListener가 로드하는 컨테이너
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory
                = new ConcurrentKafkaListenerContainerFactory<>();

        kafkaListenerContainerFactory.setConsumerFactory(consumerFactory());

        return kafkaListenerContainerFactory;
    }
}