[Spring] 실시간 채팅 기능 회고 - 3

일상 회고록·2024년 1월 16일

Bridge Springboot kafka 채팅

브릿지 프로젝트

목록 보기

3/4

안녕하세요!

지난 채팅회고 - 1 , 채팅회고 - 2 를 통해 WebSocket + Stomp 로 채팅 기능을 구현 후, 이벤트 브로커인 Kafka 도입 배경까지 알아봤습니다.

Kafka 관련 자료들은 굉장히 많기 때문에 Kafka로 브로커를 전환하는 과정은 순조롭게 이루어졌는데요

하지만 언제나 개발은 마음대로 흘러가지 않죠,,

그럼 트러블슈팅 시작하겠습니다!

0. 환경

일단 Kafka를 사용하기 위해서 Producer 와 Consumer 설정을 해주어야 합니다.

Producer

```java
@EnableKafka
@Configuration
public class ProducerConfig {

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

    @Bean
    public Map<String, Object> producerConfigurations() {
        return ImmutableMap.<String, Object>builder()
                .put(BOOTSTRAP_SERVERS_CONFIG, Constant.BOOTSTRAP_SERVER)
                .put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class)
                .put(VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class)
                .build();
    }

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

Consumer

@EnableKafka
@Configuration
public class ConsumerConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, ChatMessageRequest> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, ChatMessageRequest> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<String, ChatMessageRequest> consumerFactory() {
        JsonDeserializer<ChatMessageRequest> deserializer = new JsonDeserializer<>();
        deserializer.addTrustedPackages("*");

        Map<String, Object> consumerConfigurations =
                ImmutableMap.<String, Object>builder()
                        .put(BOOTSTRAP_SERVERS_CONFIG, Constant.BOOTSTRAP_SERVER)
                        .put(GROUP_ID_CONFIG, Constant.GROUP_ID)
                        .put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
                        .put(VALUE_DESERIALIZER_CLASS_CONFIG, deserializer)
                        .put(AUTO_OFFSET_RESET_CONFIG, "latest")
                        .build();

        return new DefaultKafkaConsumerFactory<>(consumerConfigurations, new StringDeserializer(), deserializer);
    }
}

위와 같이 설정해주고,

만들어진 Bean들을 이용해 실질적으로 메세지(프레임)을 받고, 보내주는 메소드를 작성합니다.

MessageSender

@Service
@RequiredArgsConstructor
public class MessageSender {

    private final KafkaTemplate<String, ChatMessageRequest> kafkaTemplate;

    public void send(String topic, ChatMessageRequest message) {

        // Kafka Template 을 사용하여 메세지를 지정된 토픽으로 전송
        kafkaTemplate.send(topic, message);
    }
}

MessageReceiver

@Service
@RequiredArgsConstructor
public class MessageReceiver {

    private final SimpMessagingTemplate template;

    @KafkaListener(topics = Constant.KAFKA_TOPIC, groupId = Constant.GROUP_ID)
    public void receiveMessage(ChatMessageRequest message) {
       
        // 메세지객체 내부의 채팅방 ID 참조 -> 구독자에게 메세지 발송
        template.convertAndSend("/sub/chat/room/" + message.getChatRoomId(), message);
    }
}