SSE + Redis pub/sub

기존에 SSE로만 구현한 알림 기능은 단일 서버에서 문제가 없지만 분산 서버에서 한 가지 문제가 발생한다.

사용자가 "서버1"에서 구독을 하면 SSE 연결 정보를 "서버1"의 인스턴스 메모리에 저장한다.

따라서 서버2에서 이벤트를 발행해도 서버2에는 연결 정보가 없기 때문에 제 기능을 하지 못한다.

pub/sub과 변경된 알림 기능 동작 방식

redis pup/sub 시스템은 채널에 구독 신청을 한 모든 sub에게 메시지를 전달한다.

pub/sub 내부 구조

여기서도 간단한게 설명했지만 다시 설명하자면,

publisher는 메시지를 redis 서버로 발행하고 해당 topic에 구독자(sbscriber)가 존재하면 발행자가 메시지를 발행할 경우 구독자에게 메시지가 전송된다.

즉, sender(publisher)들은 receiver(subscriber)에게 직접 전달하는 것이 아니라 해당 채널에 메시지를 전달하면 그 메시지를 구독하고 있는 구독자가 이를 수신하는 것이다.

따라서 pub/sub으로 구현된 알림 flow는 다음과 같다.

클라이언트가 SSE 연결 요청을 하면 서버는 user_id로 emitter 객체와 channel을 생성하고 구독한다.
알림이 있는 기능(리뷰에 댓글 달기 등)이 요청으로 들어오면 요청 정보(dto)를 기반으로 channel을 생성하고 전송할 메시지와 channel을 publish 한다.
개발자가 구현한 publish()가 동작하면 구현한 onMessage()를 호출하면서 구독한 subscriber에게 메시지를 전송한다. 그러면 client에게 알림이 온다.

구현 코드

기존 코드는 수정하지 않고 v2를 생성했다.

[변경 코드]

notification
---- controller
---- facadeService
---- notificationListener
---- sendNotificationService
---- subscriberService

config
---- redisPublisher
---- redisSubscriber

이벤트가 발행되면 구현한 publish() 메소드를 호출하게 된다.

@Component
@RequiredArgsConstructor
public class RedisPublisher {

    private final RedisTemplate<String, Object> redisTemplate;

    // 서킷에서 사용하는 publish
    public void publish(ChannelTopic topic, String message) {
        redisTemplate.convertAndSend(topic.getTopic(), message);
    }

    public void publish(ChannelTopic topic, NotificationResult message) {
        redisTemplate.convertAndSend(topic.getTopic(), message);
    }
}

MessageListener를 구현한 구현체이다. publish()가 호출되면 여기서 구현한 onMessage()가 호출된다.

기존에는 service에서 해당 emitterr 객체의 정보를 가지고 클라이언트에게 알림을 보냈지만,

onMessage()를 통해서 이전 기능이 동작한다.

@Service
@RequiredArgsConstructor
public class RedisSubscriber implements MessageListener {

    private final EmitterRepository emitterRepository;
    private final ObjectMapper objectMapper;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            String userId = new String(message.getChannel());
            var notificationResult
                    = objectMapper.readValue(message.getBody(), NotificationResult.class);

            emitterRepository
                    .findAllEmitterStartWithByUserId(userId)
                    .forEach(
                            (key, emitter) -> {
                                emitterRepository.saveEventCache(key, notificationResult);
                                try {
                                    emitter.send(SseEmitter.event().
                                            id(key)
                                            .data(SendNotificationResult.from(notificationResult),
                                                    MediaType.APPLICATION_JSON));
                                } catch (IOException e) {
                                    emitterRepository.deleteEmitterByEmitterId(key);
                                    emitter.completeWithError(e);
                                }
                            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

이전에는 SSE 연결만을 담당했지만 pub/sub을 적용한 코드는 SSE 연결과 동시에 채널을 생성하고 구독한다.

@Service
public class SubscribeServiceV2 implements SubscribeUseCase {

  private final EmitterRepository emitterRepository;
  private final RedisSubscriber subscriber;
  private final RedisMessageListenerContainer container;
  ...

  @Override
  public SseEmitter subscribe(final String emitterId, final String lastEventId) {
    final Long DEFAULT_TIMEOUT = 60L * 1000L * 60L;
    var emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));
    
    // 채널을 생성하고 구독하는 로직
    ChannelTopic topic = new ChannelTopic(emitterId.split("_")[0]);
    container.addMessageListener(this.subscriber, topic);

    emitter.onCompletion(() -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    emitter.onTimeout(() -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    emitter.onError((ec) -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    return emitter;
  }
}

SSE만 사용할 때는 send() 메소드에서 클라이언트에게 알림을 보내는 책임을 담당했지만

지금은 RedisSubscriber의 onMessage()에서 담당한다.

@Service
public class SendNotificationServiceV2 implements SendNotificationUseCase {

  private final EmitterRepository emitterRepository;
  private final RedisPublisher redisPublisher;

  public SendNotificationServiceV2(EmitterRepository emitterRepository, RedisPublisher redisPublisher) {
    this.emitterRepository = emitterRepository;
    this.redisPublisher = redisPublisher;
  }

  @Override
  public void send(NotificationResult notificationResult) {
    // publisher event 전송
    ChannelTopic topic = new ChannelTopic(notificationResult.receiveUserId().toString());
    redisPublisher.publish(topic, notificationResult);
  }
  
	 ...
	 ...
 }

결과 확인

두 개의 서버 인스턴스를 올리고 알림 기능이 적용된 리뷰에 댓글을 작성하는 기능을 수행했다.

[SSE]

리뷰와 알림 데이터는 잘 생성되어 DB에 반영 되었지만, Client에게 알림이 도착하지 않는다.

[SSE + redis pub/sub]

pub/sub을 적용한것은 리뷰와 알림 데이터가 생성되어 DB에 반영되고 Client에게도 알림이 잘 도착했다.

Summary

하나의 서버에서는 pub/sub 없이도 잘 동작하지만 서버의 인스턴스가 늘어나고 별도의 로드밸런싱 없이는 사용자에게 알림이 도착하는 것을 보장할 수 없었다. 하지만 redis가 제공하는 pub/sub 기능을 활용해서 여러 서버를 운영하는 서비스에서 알림이 도착하는 것을 확인할 수 있었다. pub/sub을 제공하는 다른 미들웨어들도 존재하지만 이미 리프레시 토큰을 저장하는 redis 서버와 캐싱을 담당하는 redis 서버가 각각 1개씩 존재했기 때문에 redis로 활용해봤다.

Last updated 1 year ago

hashtagpub/sub과 변경된 알림 기능 동작 방식

hashtag구현 코드

hashtag결과 확인

hashtagSummary

pub/sub과 변경된 알림 기능 동작 방식

구현 코드

결과 확인

Summary