diff --git a/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisExecutorConfig.java b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisExecutorConfig.java new file mode 100644 index 0000000..dc5b196 --- /dev/null +++ b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisExecutorConfig.java @@ -0,0 +1,41 @@ +package com.momogo.api.notification.redis; + +import java.util.concurrent.ThreadPoolExecutor; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/* + * 알림 Redis Pub/Sub 전용 스레드 풀 설정. + */ +@Configuration +public class NotificationRedisExecutorConfig { + + // 개별 Pub/Sub 메시지 비동기 처리 전용 스레드 풀 + @Bean(name = "notificationRedisTaskExecutor") + public ThreadPoolTaskExecutor notificationRedisTaskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(10); + executor.setMaxPoolSize(50); + executor.setQueueCapacity(500); + executor.setThreadNamePrefix("notification-redis-task-"); + // 큐+최대스레드가 모두 꽉 찼을 때 기본 정책(AbortPolicy)은 메시지를 버린다. + // CallerRunsPolicy로 바꿔서, 꽉 찼을 땐 발행자 스레드가 대신 처리하게 해 + // 처리 속도가 느려지더라도 메시지가 유실되지 않도록 한다. + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + executor.initialize(); + return executor; + } + + // Redis SUBSCRIBE 커넥션 유지 전용 독립 스레드 풀 + @Bean(name = "notificationRedisSubscriptionExecutor") + public ThreadPoolTaskExecutor notificationRedisSubscriptionExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(5); + executor.setQueueCapacity(10); + executor.setThreadNamePrefix("notification-redis-sub-"); + executor.initialize(); + return executor; + } +} diff --git a/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisPubSubConfig.java b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisPubSubConfig.java new file mode 100644 index 0000000..e853b8d --- /dev/null +++ b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisPubSubConfig.java @@ -0,0 +1,61 @@ +package com.momogo.api.notification.redis; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.concurrent.Executor; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; +import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; + +/* + * 알림 Redis Pub/Sub 채널/리스너/컨테이너 설정. + */ +@Slf4j +@Configuration +@RequiredArgsConstructor +public class NotificationRedisPubSubConfig { + + // 알림 SSE 중계에 사용할 Redis 채널 토픽 정의 + @Bean + public ChannelTopic notificationTopic() { + return new ChannelTopic("notification-sse-channel"); + } + + // Redis 메시지를 수신할 어댑터 설정. Jackson2JsonRedisSerializer로 역직렬화를 위임해 + // subscriber가 수동으로 JSON을 파싱하지 않고 NotificationSseMessage를 바로 받도록 함. + // Jackson2JsonRedisSerializer(Class)는 내부적으로 순정 ObjectMapper를 새로 만들어서 + // JavaTimeModule이 없어 OffsetDateTime(NotificationResponse.createdAt)을 못 읽는다. + // 앱이 이미 쓰는(JavaTimeModule 등록된) ObjectMapper 빈을 그대로 주입해 사용한다. + @Bean + public MessageListenerAdapter notificationListenerAdapter( + NotificationRedisSubscriber subscriber, ObjectMapper objectMapper) { + MessageListenerAdapter adapter = new MessageListenerAdapter(subscriber, "handleMessage"); + adapter.setSerializer(new Jackson2JsonRedisSerializer<>(objectMapper, NotificationSseMessage.class)); + return adapter; + } + + @Bean + public RedisMessageListenerContainer notificationRedisMessageListenerContainer( + RedisConnectionFactory connectionFactory, + MessageListenerAdapter notificationListenerAdapter, + ChannelTopic notificationTopic, + @Qualifier("notificationRedisTaskExecutor") Executor taskExecutor, + @Qualifier("notificationRedisSubscriptionExecutor") Executor subscriptionExecutor + ) { + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + container.addMessageListener(notificationListenerAdapter, notificationTopic); + container.setTaskExecutor(taskExecutor); + container.setSubscriptionExecutor(subscriptionExecutor); + container.setErrorHandler(e -> + log.error("[Notification Redis Pub/Sub Error] 비동기 메시지 수신/처리 중 오류 발생", e) + ); + return container; + } +} diff --git a/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisSubscriber.java b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisSubscriber.java new file mode 100644 index 0000000..2c8b066 --- /dev/null +++ b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationRedisSubscriber.java @@ -0,0 +1,56 @@ +package com.momogo.api.notification.redis; + +import com.momogo.api.notification.registry.NotificationEmitterRegistry; +import com.momogo.core.domain.notification.dto.response.NotificationResponse; +import java.io.IOException; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/* + * NotificationRedisPubSubConfig의 MessageListenerAdapter가 호출하는 구독자. + * Redis 채널로 발행된 알림을 받아, 이 인스턴스에 실제로 연결된 유저가 있으면 SSE로 전송 + * 연결된 유저가 없으면(다른 인스턴스에 연결된 유저면) 조용히 무시 + * + * 역직렬화는 MessageListenerAdapter에 등록한 Jackson2JsonRedisSerializer가 대신 처리하므로 + * 여기서는 이미 NotificationSseMessage 객체로 전달받는다. 역직렬화 자체가 실패하면 + * RedisMessageListenerContainer의 errorHandler가 잡아서 로그만 남긴다. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class NotificationRedisSubscriber { + + private final NotificationEmitterRegistry emitterRegistry; + + public void handleMessage(NotificationSseMessage message) { + if (message == null || message.userId() == null || message.notification() == null + || message.notification().id() == null) { + log.warn("[NotificationRedisSubscriber] 필수 필드가 없는 Redis 메시지를 무시합니다 - message: {}", message); + return; + } + + NotificationResponse notification = message.notification(); + var emitters = emitterRegistry.findAllByUserId(message.userId()); + if (emitters.isEmpty()) { + log.info("[NotificationRedisSubscriber] Redis 메시지 수신 - userId: {} (이 인스턴스엔 연결 없음, 무시)", + message.userId()); + return; + } + log.info("[NotificationRedisSubscriber] Redis 메시지 수신 - userId: {}, 이 인스턴스의 연결 수: {} -> SSE 전송", + message.userId(), emitters.size()); + + for (SseEmitter emitter : emitters) { + try { + emitter.send(SseEmitter.event() + .id(notification.id().toString()) + .name("notifications") + .data(notification)); + } catch (IOException e) { + log.warn("[NotificationRedisSubscriber] SSE 알림 전송 실패 - userId: {}", message.userId(), e); + emitter.completeWithError(e); + } + } + } +} diff --git a/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationSseMessage.java b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationSseMessage.java new file mode 100644 index 0000000..9f18852 --- /dev/null +++ b/momogo-api/src/main/java/com/momogo/api/notification/redis/NotificationSseMessage.java @@ -0,0 +1,14 @@ +package com.momogo.api.notification.redis; + +import com.momogo.core.domain.notification.dto.response.NotificationResponse; +import java.util.UUID; + +/* + * SSE 알림을 인스턴스 간에 중계하기 위해 Redis 채널에 실어보내는 메시지. + * userId로 "누구에게 보낼지"를, notification으로 "무엇을 보낼지"를 담는다. + */ +public record NotificationSseMessage( + UUID userId, + NotificationResponse notification +) { +} diff --git a/momogo-api/src/main/java/com/momogo/api/notification/service/NotificationSseServiceImpl.java b/momogo-api/src/main/java/com/momogo/api/notification/service/NotificationSseServiceImpl.java index ffba1f3..219d0ef 100644 --- a/momogo-api/src/main/java/com/momogo/api/notification/service/NotificationSseServiceImpl.java +++ b/momogo-api/src/main/java/com/momogo/api/notification/service/NotificationSseServiceImpl.java @@ -1,17 +1,29 @@ package com.momogo.api.notification.service; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.momogo.api.notification.redis.NotificationSseMessage; import com.momogo.api.notification.registry.NotificationEmitterRegistry; +import com.momogo.core.common.exception.BusinessException; +import com.momogo.core.common.exception.GlobalErrorCode; import com.momogo.core.domain.notification.dto.response.NotificationResponse; import com.momogo.core.domain.notification.sse.NotificationSseService; import java.io.IOException; import java.util.UUID; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; /* * NotificationSseService(publish) + NotificationSseConnector(connect) 구현체. + * + * publish()는 로컬 emitter에 직접 쓰지 않고 Redis 채널로 발행만 한다. 실제 로컬 전송은 + * NotificationRedisSubscriber가 이 채널을 구독해서 담당한다 - 이 인스턴스에 연결된 유저가 + * 아니면 조용히 무시되므로, 어느 인스턴스가 publish()를 호출하든 상관없이 그 유저가 + * 실제로 연결된 인스턴스에서 SSE가 전송된다. */ @Slf4j @Service @@ -22,6 +34,9 @@ public class NotificationSseServiceImpl implements NotificationSseService, Notif private static final long TIMEOUT = 30L * 60 * 1000; // 30분 private final NotificationEmitterRegistry emitterRegistry; + private final StringRedisTemplate stringRedisTemplate; + private final ChannelTopic notificationTopic; + private final ObjectMapper objectMapper; // 클라이언트의 최초 SSE 연결 요청을 처리 @Override @@ -41,20 +56,21 @@ public SseEmitter connect(UUID userId) { return emitter; } - // 실제 알림을 SSE로 전송 + // 알림을 Redis 채널로 발행 (실제 SSE 전송은 NotificationRedisSubscriber가 수행) + // 발행 실패는 여기서 삼키지 않고 그대로 던진다. 호출자(NotificationSsePublishListener, + // NotificationEventListener)가 이미 유저 단위로 try/catch하고 있어, 한 명의 발행 실패가 + // 다른 유저들의 알림 처리를 막지 않으면서도 실패를 호출자가 인지할 수 있다. @Override public void publish(UUID userId, NotificationResponse notification) { - for (SseEmitter emitter : emitterRegistry.findAllByUserId(userId)) { - try { - emitter.send(SseEmitter.event() - .id(notification.id().toString()) - .name("notifications") - .data(notification)); - } catch (IOException e) { - log.warn("SSE 알림 전송 실패 - userId: {}", userId, e); - // completeWithError()가 emitter의 onError 콜백을 트리거 -> registry 정리 - emitter.completeWithError(e); - } + String json; + try { + json = objectMapper.writeValueAsString(new NotificationSseMessage(userId, notification)); + } catch (JsonProcessingException e) { + throw new BusinessException(GlobalErrorCode.SSE_PUBLISH_FAILED, "SSE 메시지 직렬화 실패 - userId: " + userId, e); } + + Long receivers = stringRedisTemplate.convertAndSend(notificationTopic.getTopic(), json); + log.info("[NotificationSseServiceImpl] Redis 채널({}) 발행 완료 - userId: {}, 구독 중인 인스턴스 수: {}", + notificationTopic.getTopic(), userId, receivers); } } diff --git a/momogo-core/src/main/java/com/momogo/core/common/exception/GlobalErrorCode.java b/momogo-core/src/main/java/com/momogo/core/common/exception/GlobalErrorCode.java index a8c739c..3b80bb4 100644 --- a/momogo-core/src/main/java/com/momogo/core/common/exception/GlobalErrorCode.java +++ b/momogo-core/src/main/java/com/momogo/core/common/exception/GlobalErrorCode.java @@ -24,6 +24,7 @@ public enum GlobalErrorCode implements ErrorCode { OAUTH2_ALREADY_LINKED_OTHER_SOCIAL(9013, "OAUTH2_ALREADY_LINKED_OTHER_SOCIAL", HttpStatus.CONFLICT, "이미 다른 소셜 계정으로 가입된 이메일입니다."), OAUTH2_SOCIAL_ID_MISMATCH(9014, "OAUTH2_SOCIAL_ID_MISMATCH", HttpStatus.CONFLICT, "기존 계정 정보와 일치하지 않습니다."), OAUTH2_MISSING_ATTRIBUTE(9015, "OAUTH2_MISSING_ATTRIBUTE", HttpStatus.BAD_REQUEST, "OAuth2 사용자 정보에 필수 값이 없습니다."), + SSE_PUBLISH_FAILED(9016, "SSE_PUBLISH_FAILED", HttpStatus.INTERNAL_SERVER_ERROR, "SSE 메시지 발행에 실패했습니다."), INTERNAL_SERVER_ERROR(9999, "SERVER_ERROR", HttpStatus.INTERNAL_SERVER_ERROR, "서버 내부 오류가 발생했습니다.");