~ 6/17 한이음 프로젝트 공모전 sprint 1🔥

Kafka

• Spring 에서 Kafka를 사용하려 하다 보니, 관련 설정들을 해주어야 했다. 해당 설정들에 대해서 살펴보도록 하겠다!

ConsumerConfig

package com.hanium.diarist.common.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.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;

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

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "diary");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        config.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        config.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "java.util.HashMap");

        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(),
                new ErrorHandlingDeserializer<>(new JsonDeserializer<>()));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

• Consumer 에 대한 설정을 map을 통해 할 수 있었다. 그룹ID, serializer 방식, 등등의 설정을 ConsumerFactory를 통해 하였다.

ProducerConfig

package com.hanium.diarist.common.config;

import com.hanium.diarist.domain.diary.exception.KafkaConnectException;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Configuration
public class KafkaProducerConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
                ProducerListener.super.onSuccess(producerRecord, recordMetadata);
            }

            @Override
            public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                if (exception instanceof TimeoutException) {
                    throw new KafkaConnectException();
                }
        }

});
        return kafkaTemplate;
    }
}

• Producer 에 대한 설정으로, bootstrap-server, 직렬화 방식등에 대한 설정을 해주었다.
• 이후 KafkaTemplate를 통해 데이터를 보낼 템플릿을 설정하였다.

ProducerService

package com.hanium.diarist.domain.diary.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hanium.diarist.domain.diary.domain.Diary;
import com.hanium.diarist.domain.diary.dto.CreateDiaryRequest;
import com.hanium.diarist.domain.diary.exception.JsonProcessException;
import com.hanium.diarist.domain.diary.exception.KafkaConnectException;
import com.hanium.diarist.domain.diary.repository.DiaryRepository;
import com.hanium.diarist.domain.user.domain.User;
import com.hanium.diarist.domain.user.repository.UserRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

@Service
@Slf4j
@RequiredArgsConstructor
public class CreateDiaryProducerService {

    private final KafkaTemplate<String,String> kafkaTemplate;
    private final ObjectMapper objectMapper;
    private final DiaryRepository diaryRepository;
    private final UserRepository userRepository;
    private final String CreateTopicName = "create-diary";
    private final String reCreateTopicName = "re-create-diary";

    public boolean sendCreateDiaryMessageWithAd(CreateDiaryRequest createDiaryRequest) {
        LocalDate date = createDiaryRequest.getDiaryDate();// 일기 작성 날짜 가져옴
        User user = userRepository.findById(createDiaryRequest.getUserId()).orElseThrow();// 해당 유저 찾음
        Optional<Diary> existingDiary = diaryRepository.findByUserAndDiaryDate(user, date);// 해당 유저의 일기 작성 날짜에 일기가 있는지 확인

        try{
            String message = objectMapper.writeValueAsString(createDiaryRequest);
            if (existingDiary.isPresent()|| !date.isEqual(LocalDate.now())) { // 당시 날짜의 일기가 있거나, 과거의 일기를 작성할 경우
                // 광고 시청 필수
                sendKafkaMessage(reCreateTopicName, message);
                watchAd();
                return true;
            }else{// 과거 날짜 일기 작성
                 sendKafkaMessage(CreateTopicName, message);
                 return false;
            }
        } catch (KafkaProducerException e){
            throw new KafkaConnectException();
        } catch (JsonProcessingException e) {
            throw new JsonProcessException();
        }
    }

    private void sendKafkaMessage(String topic,String message) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.thenAccept(result -> {
            // 성공 시 로직
            log.info("Message sent successfully to topic {} with offset {}", result.getRecordMetadata().topic(), result.getRecordMetadata().offset());
        }).exceptionally(ex -> {
            // 실패 시 예외 처리
            if (ex.getCause() instanceof TimeoutException) {
                throw new KafkaConnectException();
            }
            return null;
        });
    }

    @Async
    public void watchAd() {
        System.out.println("광고 시청 중");
    }

}

• 날짜를 작성하는 조건 ( 오늘 첫 작성, 오늘 재 작성, 과거 일기 작성) 에 따라 로직이 달라지기 때문에 토픽을 나누어주었다.
• 일기를 보낼 경우 카프카 브로커에 들어가는 것을 확인할 수 있었다!

ConsumerService

package com.hanium.diarist.domain.diary.service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hanium.diarist.domain.diary.domain.Diary;
import com.hanium.diarist.domain.diary.dto.CreateDiaryResponse;
import com.hanium.diarist.domain.diary.exception.DiaryIdInvalidException;
import com.hanium.diarist.domain.diary.exception.DiaryNotFoundException;
import com.hanium.diarist.domain.diary.exception.JsonProcessException;
import com.hanium.diarist.domain.diary.exception.SseException;
import com.hanium.diarist.domain.diary.repository.DiaryRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.CopyOnWriteArrayList;

@Service
@RequiredArgsConstructor
public class CreateDiaryConsumerService {

    private final DiaryRepository diaryRepository;
    private final ObjectMapper objectMapper;
    private final CopyOnWriteArrayList<SseEmitter> emitters = new CopyOnWriteArrayList<>();// 클라이언트와 sse 연결을 관리하는 리스트

    // 클라이언트가 sse 연결을 요청할 때 새로운 sseEmitter를 생성해 반환함
    public SseEmitter addEmitter() {
        SseEmitter emitter = new SseEmitter(60000L);// 타임아웃 시간 60초
        this.emitters.add(emitter);
        emitter.onCompletion(() -> this.emitters.remove(emitter));// 연결 완료시 리스트에서 제거
        emitter.onTimeout(() -> this.emitters.remove(emitter));// 타임아웃시 리스트에서 제거
        emitter.onError(e -> this.emitters.remove(emitter));// 에러시 리스트에서 제거

        return emitter;
    }

    // 모든 클라이언트에게 메세지를 전송함.
    private void sendEmitters(CreateDiaryResponse diaryResponse){
        for (SseEmitter emitter : emitters) {
            try {
                String jsonResponse = objectMapper.writeValueAsString(diaryResponse);
                //System.out.println("Sending to emitter: " + jsonResponse);
                emitter.send(SseEmitter.event().data(jsonResponse));
            }catch (JsonProcessingException e) {
                throw new JsonProcessException();
            } catch (IOException e) {
                this.emitters.remove(emitter);
                throw new SseException();
            }
        }
    }

    // kafka 의 create-diary-response 토픽에서 메세지를 수신해서 처리함
    @KafkaListener(topics = "create-diary-response", groupId = "diary")
    public void consumeCreateDiaryResponse(HashMap<String,Integer> message) {
        try {
            long diaryId = Long.parseLong(String.valueOf(message.get("diaryId")));
            Diary diary = diaryRepository.findByDiaryIdWithDetails(diaryId).orElseThrow(() -> new DiaryNotFoundException());
            CreateDiaryResponse createDiaryResponse = new CreateDiaryResponse(diary.getEmotion().getEmotionName(), diary.getEmotion().getEmotionPicture(), diary.getContent(), diary.getArtist().getArtistName(), diary.getArtist().getArtistPicture(), diary.getDiaryDate(), diary.getImage().getImageUrl());
//            System.out.println(createDiaryResponse);
            sendEmitters(createDiaryResponse);
        } catch (NumberFormatException e) {
            throw new DiaryIdInvalidException();
        }
    }

}

• 컨슈머를 구현할 때 한가지 생각해보야 할 일이 있었다.

graph LR
	a[app] --> b[server]
	b --> c([kafka])
	c--> d[model]
	d-->c
	c-->b

	

• 다음과 같이 app→ 서버로 요청, 서버에서 카프카 , 모델에서 카프카의 메세지를 구독해서 읽어온다.
• 이후 모델에서 카프카로 메세지큐를 보내고 서버가 해당 메세지를 구독을 한다는 개념까지는 가능하였다.

<aside> 💡 그렇다면 어떻게 서버에서 app으로 일기가 작성 완료되었다는 것을 알려줄 수 있을까?

</aside>

• 이 부분에 대해서 생각해볼 필요가 있었다.

→ 생각할 수 있는 방법은 몇가지가 있었다

1. 웹소켓 (WebSocket)

• 장점:
  • 양방향 통신: 클라이언트와 서버가 지속적으로 데이터를 주고받을 수 있습니다.
  • 저지연: 실시간 통신에 적합하며, 즉각적인 상호작용이 필요할 때 유용합니다.
  • 표준 프로토콜로 널리 지원됩니다.
• 단점:
  • 연결 유지 관리: 서버는 각 클라이언트와 지속적으로 연결을 유지해야 하며, 이는 리소스를 상당히 사용할 수 있습니다.
  • 스케일링 문제: 많은 수의 동시 연결을 관리하기 위해서는 추가 인프라스트럭처와 복잡성이 요구됩니다.

2. 서버-전송 이벤트 (SSE)

• 장점:
  • 서버에서 클라이언트로의 단방향 통신: 서버가 클라이언트에게 데이터를 푸시할 수 있습니다.
  • 저렴한 유지 비용: 웹소켓에 비해 서버 부담이 적습니다.
  • 자동 재연결 및 간단한 메시지 이벤트 처리.
• 단점:
  • 단방향 통신만 가능: 클라이언트에서 서버로는 직접적인 메시지 전송이 불가능합니다.
  • 브라우저 호환성 문제: 일부 오래된 브라우저에서는 지원되지 않을 수 있습니다.

3. 파이어베이스 클라우드 메시징 (FCM)

• 장점:
  • 광범위한 기기 지원: 모바일 기기와 데스크톱을 포함해 다양한 플랫폼에서 작동합니다.
  • 높은 확장성 및 신뢰성: 구글의 인프라를 활용하여 대규모 메시지를 효과적으로 전송할 수 있습니다.
  • 오프라인 메시지 처리: 기기가 오프라인일 때 메시지를 저장했다가 기기가 온라인 상태가 되면 전송합니다.
• 단점:
  • 외부 서비스 의존성: FCM은 구글의 서비스이므로 이를 통한 메시지 전송은 구글의 정책과 서비스 상태에 크게 의존합니다.
  • 추가 비용 발생 가능성: 무료 사용량을 초과할 경우 비용이 발생할 수 있습니다.

최적의 선택

• 웹소켓은 양방향 통신이 필요하거나 사용자의 실시간 상호작용이 중요한 경우 가장 적합합니다.
• SSE는 서버에서 클라이언트로 주기적이거나 지속적인 데이터 스트림이 필요할 때 유용하며, 리소스 사용이 덜합니다.
• FCM은 다양한 기기와 플랫폼 간의 메시지 전송이 중요하고, 외부 서비스에 의존하는 것이 문제가 되지 않는 경우 적합합니다.

카프카에서 데이터를 받아 클라이언트에게 전송하는 경우, 서버와 클라이언트 간의 통신 방식, 지원해야 하는 클라이언트 유형, 필요한 인프라 등을 고려해 가장 적합한 기술을 선택하는 것이 중요합니다. 예를 들어, 높은 실시간 반응성과 양방향 통신이 필요한 경우 웹소켓을, 비교적 간단한 알림이나 데이터 업데이트 전송에는 SSE나 FCM을 고려해볼 수 있습니다.

• 다음의 이유를 고려해 SSE로 구현하기로 하였다.

---

1. 클라이언트가 SSE로 연결을 요청한다. ( 타임아웃시간 1분 )
2. 해당 1분 내에 카프카의 컨슈머로 일기 작성이 완료되었다는 메세지를 읽으면 해당 데이터를 가져온다,.
3. 일기의 데이터를 SSE로 클라이언트에게 전송한다.

• 다음과 같은 로직의 순서로 서버에서 클라이언트까지 데이터를 전송 할 수 있었다.