[싹틔움] 04/06 개발일지 Kafka 도입

💡 목표

싹틔움 서버에 Kafka 를 도입해 메인서버와 알림서버 사이에서 데이터들이 성공적으로 주고 받을 수 있고자 했다.

📚 메인서버 Kafka 설정

KafkaConfig

@Configuration
@EnableKafka
public class KafkaConfig {

    @Value("${app.kafka-brokers}")
    private String kafkaBrokers;

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

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> producerConfig = new HashMap<>();
        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokers);
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        producerConfig.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.apache.kafka.clients.producer.UniformStickyPartitioner");

        return new DefaultKafkaProducerFactory<>(producerConfig);
    }
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> myConfig = new HashMap<>();
        myConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokers);
        myConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        myConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        myConfig.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
        myConfig.put(JsonDeserializer.TRUSTED_PACKAGES, "*");

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

KafkaConfig 파일을 통해 주고 받은 데이터들을 직렬화, 역직렬화로 변환해준다.

 

NotificationProducer

@Service
@RequiredArgsConstructor
public class NotificationProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void sendNotification(NotificationMessage message) {
        kafkaTemplate.send("notifications", message);
    }
}

 

topic 명을 notifications 로 설정하고 message 를 전달해준다.

 

NotificationMessage

@Getter
@NoArgsConstructor
@AllArgsConstructor
public class NotificationMessage {
    private Long userId;
    private EventType eventType;
    private String message;
}

 

message 는 위처럼 설정해주었다.

 

testService

public String testKafka(Long userId) {
    notificationProducer.sendNotification(
            new NotificationMessage(userId,
                    EventType.FRIEND_REQUESTED,
                    "test"));
    return "카프카 테스트";
}

 

/v1/test 엔드포인트로 위와 같은 요청을 보낼 수 있게 해 연결 테스트를 해봤다.

---

Notification

@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaListenerHandler {

    private final SseEmitterHandler sseEmitterHandler;
    private final NotificationRepository notificationRepository;

    @KafkaListener(topics = "notifications", groupId = "notification-group")
    public void consume(NotificationMessage message) {
        log.info("📥 Received Kafka message: {}", message);
        String topic = "notifications-" + message.getUserId();
        String data = "EventType: " + message.getEventType() + ", Message: " + message.getMessage();
        Notification notification = new Notification(message.getUserId(), message.getEventType(), message.getMessage());
        log.info("notification DB 저장");
        notificationRepository.save(notification);
        sseEmitterHandler.broadcast(topic, data);
    }
}

 

알림서버에서 마찬가지로 KafkaConfig 를 설정해주고, topic 명 notifications , 컨슈머 그룹 id 는 notification-group 으로 설정을 해줬다. 이렇게 되면 메인 서버에서 Kafka 의 notifications topic 에 데이터가 들어가면 자동으로 그 데이터를 불러온다.

위와 같이 요청을 보냈다. ( 8080 포트 )

요청 완료

확인을 위해 이를 notification Repository DB 에 저장시켰고 위와 같이 확인이 가능했다.