[SSA] 실패한 알림에 대해서는 순차 처리 ? 병렬 처리 ?

실패한 알림을 재전송할 때 FCM 알림만 다시 쓰게 할 수도 있고,

아니면 기존의 SSE를 다시 활용하는 방식으로 다시 사용자가 실시간 접속 중인지 확인하고 SSE를 보내거나
그렇지 않다면 FCM을 전송하는 방식을 활용할 수도 있습니다.

 

 

그리고 실패한 알림을 재전송하는 데에
비동기로 처리할 수도 있고

동기로 처리할 수도 있습니다.

 

이때, 비동기로 처리했을 경우 콜백 메서드를 등록해 둔다면
또 다시 알림 전송에 실패했을 경우 스레드 정보가 달라질 수 있습니다.
그러면 처음에 DB에서 찾은 실패 알림과 컨텍스트가 달라져서
콜백 메서드가 DB에 실패한 알림을 update 하는 것이 아니라 insert 하게 됩니다.

 

 

또, 재전송하는 데에 있어서 구체적인 방식을 고민해 보자면 다음과 같습니다.

• 트랜잭션 새로 전파하고 조건에 맞게 배치 사이즈만큼 실패한 알림 가져 와서 순차 처리 한다 ?
  • 하나씩 처리하니까 트랜잭션도 하나씩 잡을 거고
  • 개별 트랜잭션에서 FCM 알림 느려지더라도 하나 잡은 상태니까 다른 기능에 영향 최소화할 수 있고
  • 어차피 이미 실패한 거 조금 느려도 된다는 생각입니다 ..
• 트랜잭션 새로 전파하고 재전송 메서드가 사용하는 스레드 풀 사이즈를 커넥션 풀 사이즈보다 작게 한다 ?
  • 다른 곳에서 커넥션 쓰면 똑같은 문제가 발생할 것입니다.
    1. 그렇다고 아예 안 쓰는 건 또 다른 이야기인 게 실제로 알림 전송이 무지막지한 시간이 걸리지 않을 가능성도 있고 ..
• 그러면 트랜잭션 시간을 짧게 가져 가도록 어노테이션을 최대한 지양한다면 ?
  • 실패한 알림 조회 시에만 SimpleJpaRepository에서 설정된 @Transacional(readOnly = true)를 사용합니다.
• 알림 전송 실패했는데 재전송할 때 혹시라도 사용자가 접속 중이라면 다시 SSE로 보낼 수도 있다 ?
  • 이거는 어떻게 해야 할지 그냥 나름 정하면 될 거 같은데요 ..
    1. 지금은 정상 작동의 경우, 요청한 쪽 말고 요청받은 쪽만 알림을 받도록 구현해 놓았습니다.
    2. 요청받은 쪽, 즉 요청 대상이 실시간 접속 중이라면 SSE 그렇지 않으면 FCM인데
    3. 솔직히 뭘 선택해도 나쁘지 않은 거 같아서 우선 FCM만 사용하는 것으로 할 거 같습니다.
    4. 근데 SSE를 통해서 다시 레디스 조회하는 것보다 그냥 어차피 실패한 거 FCM으로 바로 쏘는 것도 괜찮을 거 같습니다 ..

 

 

그래서 각 경우를 구현하자면

 

 

## 배치 사이즈만큼 순차 처리

@Scheduled(fixedDelay = 600000)
public void retryFailedNotifications() {
    // 1. 배치 사이즈만큼 가져오기, 여기서는 그냥 대충 50개 잡음. 별 이유 없음 그냥 예시 ..
    Pageable limit = PageRequest.of(0, 50);
    List<FailedNotification> targets = failedNotificationRepository.findByRetryCountLessThan(MAX_RETRY_COUNT, limit);

    if (targets.isEmpty()) {
    	return;
    }

    log.info("순차 재처리 시작. 대상: {}건", targets.size());

    // 2. DB 커넥션 1개만 사용해서 현재 스레드에서 하나씩 루프
    for (FailedNotification target : targets) {
        try {
            // 개별 트랜잭션 처리는 Service에 위임
            notificationRetryService.processSingleRetry(target.getId());
        } catch (Exception e) {
            log.error("재처리 실패 ID: {}", target.getId(), e);
        }
    }
}

 

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void processSingleRetry(Long id) {
    failedNotificationRepository.findByIdWithLock(id).ifPresent(notification -> {
        try {
            NotificationRequest request = NotificationRequest.fromFail(notification);
            // 동기 대기 (Block)
            fcmService.retryFcmNotification(request).get(5, TimeUnit.SECONDS);
            
            failedNotificationRepository.delete(notification);
        } catch (Exception e) {
            notification.incrementRetryCount();
            // Dirty Checking으로 자동 저장
        }
    });
}

 

 

## 트랜잭션 새로 전파하고 스레드 풀 사이즈 튜닝

@Configuration
public class ThreadPoolConfig {
    @Bean(name = "retryExecutor")
    public Executor retryExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // DB 커넥션 풀보다 작게 설정하여 리소스 점유 제한
        executor.setCorePoolSize(5); 
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(1000);
        executor.setThreadNamePrefix("Retry-");
        executor.initialize();
        return executor;
    }
}

 

@Scheduled(fixedDelay = 600000)
public void retryFailedNotifications() {
    List<FailedNotification> targets = failedNotificationRepository.findByRetryCountLessThan(MAX_RETRY_COUNT);

    // 커스텀 Executor("retryExecutor") 사용
    List<CompletableFuture<Void>> futures = targets.stream()
            .map(target -> CompletableFuture.runAsync(() -> {
                notificationRetryService.processSingleRetry(target.getId());
            }, retryExecutor)) 
            .toList();
    
    // 필요 시 join으로 대기
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
}

 

 

## 트랜잭션을 조회 시에만 사용하도록 따로 지정 X

@Component
@RequiredArgsConstructor
public class NotificationRetryService {

    private final FailedNotificationRepository repository;
    private final FcmService fcmService;

    // 메서드 레벨 @Transactional 제거
    // 이렇게 하면 조회와 삭제/저장 시에만 순간적으로 트랜잭션이 열리고 닫힘
    public void processSingleRetry(Long id) {
        // 1. 조회 (OSIV가 꺼져있다면 여기서 영속성 컨텍스트 종료됨)
        FailedNotification notification = repository.findById(id)
                .orElseThrow(() -> new EntityNotFoundException("Notification not found"));

        try {
            NotificationRequest request = NotificationRequest.fromFail(notification);
            
            // 2. FCM 전송 (DB 커넥션 없이 수행 - Free!)
            fcmService.retryFcmNotification(request).get();

            // 3. 성공 시 삭제 (새로운 짧은 트랜잭션)
            repository.deleteById(id);
        } catch (Exception e) {
            log.warn("재전송 실패: {}", e.getMessage());
            
            // 4. 실패 시 카운트 증가 (새로운 짧은 트랜잭션)
            // 영속성 컨텍스트가 끊겼으므로 Dirty Checking 불가
            // 직접 Update 쿼리를 날리거나, 다시 조회해서 저장해야 함
            increaseRetryCountDirectly(id);
        }
    }

    // 별도 메서드로 분리하여 트랜잭션 적용 (또는 Repository @Modifying 쿼리 사용)
    @Transactional
    protected void increaseRetryCountDirectly(Long id) {
        repository.findById(id).ifPresent(FailedNotification::incrementRetryCount);
    }
}

 

 

## 실패 알림 재전송 시에도 SSE 사용

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void processSingleRetry(Long id) {
    failedNotificationRepository.findByIdWithLock(id).ifPresent(notification -> {
        try {
            NotificationRequest request = NotificationRequest.fromFail(notification);
            boolean isSent = false;

            // 1. Redis를 통해 실시간 접속 여부 확인
            if (redisClientManager.isClientConnected(request.targetId())) {
                try {
                    // SSE 전송 시도
                    sseService.sendSseNotification(request);
                    isSent = true;
                    log.info("재시도: SSE로 전송 성공 (targetId: {})", request.targetId());
                } catch (Exception e) {
                    log.warn("재시도: SSE 연결 확인됨 BUT 전송 실패 -> FCM 전환");
                }
            }

            // 2. SSE 미접속이거나 실패 시 FCM 전송
            if (!isSent) {
                fcmService.retryFcmNotification(request).get(5, TimeUnit.SECONDS);
                log.info("재시도: FCM으로 전송 성공 (targetId: {})", request.targetId());
            }

            // 3. 성공 시 삭제
            failedNotificationRepository.delete(notification);

        } catch (Exception e) {
            // 최종 실패 처리
            notification.incrementRetryCount();
            // (Dirty Checking)
        }
    });
}

 

 

 

 

 

 

 

### 결론

결론적으로,

 

단순하고 다른 방법들에 비해 다소 안정성이 높은(?)

배치 사이즈만큼 순차 처리

방식을 선택하되, 

 

처리해야 할 실패 알림이 너무 많아져서 스케줄링에 시간이 너무 오래 걸린다거나

시간이 너무 오래 걸려서 다음 스케줄링에 영향을 주게 될 경우에는

 

다른 방식을 적용하는 걸 고민해 볼 것 같습니다.