[ Spring Cloud ] Microservice들 간의 통신 구축하기 (2) - Kafka

이전 포스팅에선 Webclient를 이용하여 서비스들간의 통신을 구축했다.

 

이번에는 Kafka를 통해서 메세지 큐를 기용하여 서비스들 간의 통신 환경을 구축해보려고 한다.

 

kafka에 대한 예전 포스팅을 참고해서 구축했다.

 

현재 내 서비스는 app-service 와 user-service가 있는데, 

 

app-service를 통해 사용자가 자신의 앱을 서버에 등록할 수 있는 기능을 구현해 놓았다.

 

필자는 지금 사용자가 앱을 등록할 때, User 엔티티에 있는 appCount를 +1해주는 서비스를 구현하려고 한다.

0. Depdency

 

다음과 같이 의존성을 추가하면 kafka를 사용할 수 있다.

dependencies {
    implementation 'org.springframework.kafka:spring-kafka'
}

 

 

1. Producer

먼저 메세지를 보내는 Producer쪽을 알아보겠다.

 

1-1. Config

@EnableKafka
@Configuration
public class KafkaProducerConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        return new KafkaAdmin(configs);
    }

    // New Topic
    @Bean
    public NewTopic userTopic() {
        return new NewTopic("user_topic", 1, (short) 1);
    }


    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

 

 

 

새로운 토픽을 생성하는 방법은 여러가지가 있다.

 

public NewTopic(String name, int numPartitions, short replicationFactor) {
    ...
}

public NewTopic(String name, Optional<Integer> numPartitions, Optional<Short> replicationFactor) {
	...
}

public NewTopic(String name, Map<Integer, List<Integer>> replicasAssignments) {
	...
}

 

여러 방법으로 토픽의 이름, 파티션, 복제 인자를 전달하면 되겠다.

 

user-service가 subscribe 할 토픽의 이름을 user_topic으로 정했다.

(토픽을 subscribe하는 consumer쪽에서 생성하는게 맞는거 같다.)

 

현재 지금 app-service 외에도 user-service에 메세지를 보낼 일이 있다면, 해당 토픽으로 메세지를 publish하면 되겠다.

 

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

 

중간에 Serializer를 설정하는 부분이 있는데, 이는 Producer가 메세지를 발행할때 메세지를 직렬화하여 보내는데, 이 메세지에 대한 형태를 설정하는 것으로 보면 되겠다.

 

반대로 Consumer는 Deserializer를 사용하는데, Producer와 클래스를 맞추면 되겠다.

 

Key와 Value 모두 String으로 전달할 것이기 때문에, Serializer를 String으로 정했다.

 

- application.yml

spring:
  kafka:
    bootstrap-servers: "localhost:9092"
    consumer:
      group-id: "producerGroupId"

 

1-2. Service

@Slf4j
@RequiredArgsConstructor
@Service
public class KafkaService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessageUserTopic(String message) {

        log.info("[ KafkaProducer ] Send Message : " + message);
        kafkaTemplate.send("user_topic", message);
    }
}

 

Producer가 될 app-service에서는 user_topic으로 메세지를 발행하는 서비스를 만들었다.

 

kafkaService.sendMessageUserTopic("userAppCountPlus - " + appRequestDto.getUserId());

 

서비스 단에서 kafkaService를 호출하여 메세지를 발행하는 로직으로 구현했다.

 

KafkaTemplate의 제네릭스타입 K, V는 각각 토픽의 타입과 메세지의 타입을 명시하면 된다.

현재는 테스트용 프로젝트를 구축하는 중이므로, 간단한 String을 전송하지만 실제 프로젝트에서는 객체를 보내게될 수도 있겠다.

 

2. Consumer

이제 consumer쪽을 살펴보겠다. 내 프로젝트에서는 user-service가 되겠다.

 

2-1.  Config

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Value(value = "${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
}

 

 

ConcurrentKafkaListenerContainerFactory에서 concurrency는 생성할 consumer 수를 뜻한다.

해당 Factory는 AbstractKafkaListenerContainerFactory를 상속하는데,

해당 팩토리에는 설정할 수 있는 값들이 너무 많다. 😵‍💫

카프카는 복잡해

웬만하면 세세하게 뜯어보면서 어떤 설정들인지 하나씩 살펴보는데, 이건 살펴보기에 너무 많다. .

 

나중에 기회가 되면 깊게 파고들어 보자 🫠

 

이 AbstractKafkaListenerContainerFactory에는 ContainerProperties가 있는데, (.getContainerProperties()로 한번 가져와야 한다.) 이 클래스는 Listner의 Container에 대한 설정을 할 수 있다.

 

ContainerProperties의 일부. 여기도 뭐가 많다🤢

 

여기서 나는 consumer가 record를 기다릴 때 timeout 을 3000으로 설정해주었다.

 

그 외 IP 주소와 Deserializer는 Producer와 맞춰주었다. 

 

2-2. Listner

 

@Slf4j
@Service
public class KafkaConsumer {

    @KafkaListener(topics = "user_topic", groupId = "MSAGroupId")
    public void consumeUserTopic(@Payload String message, ConsumerRecordMetadata metadata) {
        log.info("[ KafkaConsumer] Message Received : " + message);
        log.info("[ KafkaConsumer] Partition : " + metadata.partition());

    }
}

 

Consumer는 @KafkaListner 어노테이션으로 토픽의 메세지를 받을 수 있다.

 

해당하는 토픽과 그룹id를 지정해두면 메세지를 받을 수 있다.

 

현재 나는 메세지를 확인하는 로그만 두었다.

 

 

테스트

이제 요청을 보내 메세지를 정상적으로 주고받는지 확인해보자.

 

 

app-service 에서 Producer가 정상적으로 메세지를 Publish했다.

 

user-service또한 정상적으로 메세지를 가져왔다.

 

 

개발 고수(?)들도 Kafka는 피한다는 말을 많이 들었는데, 왜 그런지 조금은 알 것 같다.

 

지금은 아주아주 기본적인 세팅들로만 구동했지만, 규모가 큰 앱에서는 세세한 설정이 많이 필요하겠다.

 

그래도 메세지큐를 사용하고, Pub/Sub구조로 메세지를 주고받는건 매력적인 것 같다.

 

여러 서비스에서 활용 방안이 정말 많을 것 같다.