SpringBoot 주문,알림 프로세스 성능개선

공덕이형·2023년 12월 8일

FCM Message Queue Springboot kafka zookeeper 성능개선

배달 플랫폼 프로젝트를 진행하며 알림 프로세스 성능 개선을 진행했던 과정을 작성해보고자 한다.

FCM 사용 이유

우선 배달 플랫폼의 주문 프로세스를 생각해보자

고객이 해당 가게에 물품을 담는다.
결제를 진행한다.
해당 가게에서 조리를 시작한다.

여기서 결제는 카카오 페이 API를 활용했고, 해당 가게의 어드민에게 주문이 들어왔다는 사실을 알려주어야 한다.
또한, 결제과 완료되는 순간 데이터베이스에 복잡한 주문 내역 DTO를 Insert 해야 하고, 어드민에게 알림을 전송해야 했다.

주문 알림 어떤걸로 할까?

우선 알림을 어떻게 전송할지에 대해 고민이 가장 많았다.

생각해본 기술들은 다음과 같다.

웹소켓
- 채팅을 구현할 수 있다는 점에서 아이디어를 얻었다.
- 그러나, 웹소켓은 클라이언트와 서버가 지속적으로 연결되어야만 한다.
- 즉, 연결을 지속하는데 자원이 필요하고 혹여나 웹소켓 연결이 끊어지게 된다면?
메세지 큐
- MQ는 server to server에 사용된다. 클라이언트에게 바로 메세지를 전달하는데 적합하지 않을 것이라 판단했다.
NaverCloud Simple & Notification Service
- SMS를 통해 어드민에게 알림을 보낸다?
- 굳이 웹 어플리케이션으로 만들었는데 모바일로 알림을 알려줘야만 할까?
- 바쁜 상황에 모바일을 볼 수 없는 상황이라면?
FCM
- 웹 어플리케이션이니 웹 푸시 알림을 보내는 것이 가장 적합하다 생각했다.

FCM 어떻게 개선해야 할까?

배달 플랫폼 특성상 사용자들이 서비스를 경험하는 곳은 주문 과정이라 생각하게 됐다.
만약 사용자의 과다 유입으로 인해 주문 프로세스에 대용량 트래픽이 들어오게 된다면 문제가 생길것이 당연했다.

또한, FCM이라는 외부 API에 의존하게 될 경우 서버의 요청과정이 하나 더 늘어나게 되는 것이 문제가 될 것이였다.

FCM 처리 속도

현재 주문 프로세스 과정은 다음과 같다.

사용자가 결제를 완료한다.
결제가 완료됨과 동시에 ORDERS, ORDER_DETAILS, ORDER_MENU, ORDER_MENU_SUB 총 4개의 테이블에 INSERT가 시행된다.
INSERT가 완료된 후, FCM을 통해 사장님 웹 어플리케이션으로 웹푸시가 발송됨

이 과정에서 Jmeter를 통해 100개의 스레드, 100번 루프, 1초당 1번의 요청으로 테스트를 진행해 보았다.

예외까지 발생하게 되며, 이를 처리해줄 수 있는 방법이 없다.

FCM @Async 적용

이를 해결하기 위해선 운영체제를 학습하며 배웠던 비동기 개념을 적용시키기로 했다.

@Async를 적용하여 비동기 작동으로 변경한 후

스레드 풀을 사용해서 스레드를 관리할 수 있도록 설정했다. 서비스의 규모에 따라서 최대 스레드 수를 변경하면 되도록 설정했다. 테스트 과정에서는 100개의 스레드를 사용할 것이므로 최대 스레드 수를 100으로 설정했다.

그러나, 여기서 한 가지 의문점이 생겼다. 만약 FCM에서 Exception 발생시엔 알림이 전송되지 않았다는 것 아닌가? 그렇다면 비동기로 동작한 FCM을 어떻게 관리해줘야 할까?

FCM Kafka

위의 문제점을 해결하기 위해 생각해 낸 것이 Kafka였다. Kafka란 대량의 데이터를 수집 및 처리하기 위해 사용되며 브로커들이 하나의 클러스터로 구성되어 파티션을 추가하여 수평적으로 확장할 수 있다는 장점을 가졌다. 또한 ErrorHandler를 통해 예외가 발생한 메세지를 따로 처리할 수 있다는 장점이 있다.

이러한 장점을 토대로 Kafka를 선택하게 됐고 이를 적용해보았다.

우선 Kafka의 토픽으로 정상 알림인 babpool-fcm과 예외발생한 알림인 babpool-fcm-r로 토픽을 분류했다.

먼저 FCM Producer를 구현해보겠다. 프로듀서는 따로 어노테이션으로 구현하지 않고 KafkaTemplate 클래스의 send 메서드를 통해 구현한다.

@Service
@RequiredArgsConstructor
public class KafkaProducer {

    @Value("${spring.kafka.topic}")
    private String TOPIC;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    public void sendMessage(FCMNotificationRequestDto dto) throws IOException {
        try {
            String jsonDto = objectMapper.writeValueAsString(dto);
            this.kafkaTemplate.send(TOPIC, jsonDto);
        } catch (Exception e) {
            throw new IOException();
        }
    }
}

또한 Consumer를 FCM 전송 메서드로 지정해 준다.

    @KafkaListener(topics = "babpool-fcm", groupId = "babpool", errorHandler = "fcmErrorHandler")
    public void sendNotificationByToken(String jsonDto) throws JsonProcessingException {
        try {
            processAndSendMessage(jsonDto);
        } catch (FirebaseMessagingException e) {
            throw new FcmException(e.getMessagingErrorCode());
        }
    }

만약 예외 발생시엔 재전송이 가능하도록 RetryTemplate과 ErrorHandler를 구현하도록 하자
RetryTemplate의 경우 예외 재발송 포함 최대 3회까지 시도하도록 구현했다.

@Configuration
@EnableKafka
public class KafkaConfig {
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
        fixedBackOffPolicy.setBackOffPeriod(1000L);
        retryTemplate.setBackOffPolicy(fixedBackOffPolicy);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(2);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }
}

@Component
@Slf4j
@RequiredArgsConstructor
public class FcmErrorHandler implements KafkaListenerErrorHandler {

    private final KafkaTemplate<String, String> kafkaTemplate;
    @Value("${spring.kafka.r-topic}")
    private String DEAD_TOPIC;

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
        return null;
    }

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
        String fcmError = "[KafkaErrorHandler] kafkaMessage=[" + message.getPayload() + "], errorMessage=[" + exception.getCause() + "]";
        log.error(fcmError);
        FcmException fcmException = (FcmException) exception.getCause();
        MessagingErrorCode status = fcmException.getStatus();

        if (status == MessagingErrorCode.UNAVAILABLE || status == MessagingErrorCode.INTERNAL) {
            kafkaTemplate.send(DEAD_TOPIC, (String) (message.getPayload()));
        } else {
            return null;
        }
        return null;
    }
}

재전송 시에는 DEAD_TOPIC으로 토픽을 변경한 후, DEAD_TOPIC을 처리할 수 있도록 컨슈머를 하나 더 구현한다.

    @KafkaListener(topics = "babpool-fcm-r", groupId = "babpool", errorHandler = "reFcmErrorHandler")
    public void reSendNotificationByToken(String jsonDto) {
        try {
            retryTemplate.execute(context -> {
                try {
                    processAndSendMessage(jsonDto);
                    return null;
                } catch (JsonProcessingException ex) {
                    throw new RuntimeException(ex);
                } catch (FirebaseMessagingException ex) {
                    throw new FcmException(ex.getMessagingErrorCode());
                }
            });
        } catch (RetryException ex) {
            log.error("재시도 횟수를 초과했습니다.", ex);
        }
    }

재전송은 앞에서도 설명했듯이 최대 3회 실행한다. 만약 3회 이상시에도 전송하지 못한 경우 RetryException으로 log를 남기는 것으로 구현했다.

ErrorHandler

에러 핸들러를 정확하게 타는지 확인해보기 위해 우선 FCM API문서에서 예외코드를 찾아보았다.

이 문서에서 UNAVAILABLE과 같이 FCM 서버에서 문제가 생길 경우엔 ErrorHandler를 거칠 수 있도록 해야하지만, 지금 테스트환경에서는 내가 FCM서버를 중단시킬 수 없기에 잘못된 토큰을 보내 INVALID_ARGUMENT를 탈 수 있도록 구성해서 테스트를 진행해봤다.