알림 기능

어떤 이벤트가 발생 했을 때, 사용자에게 알림이 가도록하는 알림 기능을 추가하려고 한다.

사용할 수 있는 프로토콜

1. Polling

주기적으로 서버에 요청하여 변경 내용을 확인하는 폴링 방식. 하지만 실시간 통신이 필요할 경우 비효율적이다. 실시간 요청이 아니고 정해진 시간(N)을 기준으로 (요청 -> 응답)을 반복한다.

• Short Polling 클라이언트가 서버로부터 정기적인 정보를 받기 위한 프로토콜이다. 과정은 다음과 같다. 1) 클라이언트가 서버에 새로운 정보에 대한 HTTP 요청을 보낸다. 2) 서버는 새로운 정보를 반환한다. 3) 클라이언트는 설정한 주기(time)로 요청을 반복한다. 주기적으로 HTTP Reqeust를 보내는 방식으로, 실시간성이 떨어지고 서버에 불필요한 부하를 줄 수 있다. TCP의 커넥션을 맺고 끊는 것 자체가 무겁기 때문에 실시간으로 변환되는 빠른 정보의 응답을 기대하기 힘들다.

• Long Polling Short Polling 보다는 더 효율적인 방식이다. 이벤트가 발생할 때까지 연결을 유지하는 방식으로 여전히 새로운 연결을 자주 설정해야 한다. 1) 클라이언트가 서버에 요청을 보낸다. 서버에 새로운 이벤트가 발생할 때까지 기다린다. 2) 서버는 이벤트 정보를 반환한다. 3) 클라이언트는 응답을 받고 요청한다. Shor Polling에 비해 필요한 요청 수를 줄일 수 있다. 하지만 서버로부터 응답을 받고나면 다시 연결 요청을 해야하기 때문에 빈번하게 연결 요청을 하게되면 서버에 부담이 가는것은 동일하다. 따라서 이 방식은 서버의 상태가 주기적으로 변하지 않는 경우에 적합하다.

2. Web Socket

웹 소켓은 OSI 4계층 프로토콜인 TCP에 기반한 양방향 메시지 전달 프로토콜이다. 기존의 HTTP 요청-응답 구조와 달리, 지속적으로 연결을 유지하며 서버와 클라이너트 간 즉각적인 데이터 교환을 가능하게 한다. 이로인해 HTTP보다 오버헤드가 적고 데이터 전송 속도가 빠르다.

웹소켓은 실시간 양방향 통신이 필요한 애플리케이션에 적합하며, 특히 멀티플레이어 게임, 실시간 협업 도구, 라이브 채팅 등과 같이 지속적이고 빠른 데이터 업데이트가 요구되는 경우에 유용하다. Polling이나 Server-Sent Events(SSE)가 충분하지 않은 상황에서 웹소켓 사용을 고려할 수 있다. 하지만 구현이 상대적으로 복잡할 수 있다.

3. SSE (Server Sent Events)

SSE는 서버에서 클라이언트로의 단방향 실시간 통신을 위한 기술로 HTTP 프로토콜을 사용하여 구현한다. 추가적인 라이브러리나 프토콜이 필요하지는 않다. 연결이 끊어지면 자동 재연결 기능을 제공하며, 구현이 간단하고 대부분의 브라우저에서도 지원한다. 또한 HTTP의 Persistence Connection을 이용한 기술이기 때문에 여러 HTTP 요청과 응답을 한 번의 TCP 연결로 처리할 수 있다. (지속적으로 데이터를 스트리밍 할 수 있음)

과정은 다음과 같다. 1) 클라이언트는 서버를 구독한다. (SSE Connection을 맺음) 2) 서버는 이벤트가 발생할 때마다 구독된 클라이언트에게 데이터를 전송한다.

SSE는 단방향 통신(서버 -> 클라이언트)이기 때문에 양방향이 필요하면 WebSocket을 고려해본다.

SSE 채택 이유

채택한 가장 큰 이유는 단방향 통신이기 때문이다. 구현하려는 알림 기능은 서버에서 클라이언트라는 특성이 있어서, 클라이언트가 서버에 요청을 보낼 필요가 없어 실시간 메시지 전송에는 매우 적합하다. 이러한 특성으로 현 상황에서는 Web Sockect 보다는 SSE가 더 적합하고 판단했다.

하지만 SSE에도 여러 단점들은 존재한다. 클라이언트와 HTTP 연결을 지속적으로 유지해야 하므로, 동시에 많은 ㅇ녀결이 발생할 경우 서버의 부하 관리가 문제가 될 수 있다. 또한, 대량의 클라이언트에게 동시에 데이터를 전송하는 경우에도 성능 문제도 고민해 봐야하는 요소 중 하나일 것이다.

SseEmitter 객체

구현에 앞서 Spring Framework 4.2부터 SSE 통신을 지원하는 SSeEmitter 클래스가 추가되고, Spring 5부터는 webflux를 이용해서 SSE 통신도 할 수 있다.

동작 Flow

1. 먼저 유저를 subscribe(Long userId) 구독 해야한다.

2. 컨트롤러에서 header("lastEventId")를 받고 Service의 subscribe() 인자로 전달한다.

3. SseEmitter 객체를 생성해서 저장소에 저장시키고 필요할 때마다 유저와 매칭되는 SseEmitter를 불러와서 이벤트에 대한 응답을 전송해준다.

구현

✅ EmitterRepositoryImpl

이벤트가 발생하면 userId로 SseEmitter 객체를 생성한다. 그것을 저장하고 관리하기 위한 저장소이다. (알림 조회, 삭제..etc)

@Repository
public class EmitterRepositoryImpl implements EmitterRepository {

  private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

  private final Map<String, Object> eventCache = new ConcurrentHashMap<>();

  @Override
  public SseEmitter save(String emitterId, SseEmitter sseEmitter) {
    emitters.put(emitterId, sseEmitter);
    return sseEmitter;
  }

  @Override
  public void saveEventCache(String eventCacheId, Object event) {
    eventCache.put(eventCacheId, event);
  }

  @Override
  public Map<String, SseEmitter> findAllEmitterStartWithByUserId(String userId) {
    return emitters.entrySet().stream()
        .filter(entry -> entry.getKey().startsWith(userId))
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
  }

  @Override
  public Map<String, Object> findAllEventCacheStartWithByUserId(String userId) {
    return eventCache.entrySet().stream()
        .filter(entry -> entry.getKey().startsWith(userId))
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
  }

  @Override
  public void deleteEmitterByEmitterId(String emitterId) {
    emitters.remove(emitterId);
  }

  @Override
  public void deleteAllEmitterStartWithId(String userId) {
    emitters.forEach(
        (key, emitter) -> {
          if (key.startsWith(userId)) {
            emitters.remove(key);
          }
        });
  }

  @Override
  public void deleteCacheById(String id) {
    eventCache.remove(id);
  }

  @Override
  public void deleteAllEventCacheStartWithId(String userId) {
    eventCache.forEach(
        (key, emitter) -> {
          if (key.startsWith(userId)) {
            eventCache.remove(key);
          }
        });
  }
}

➡️ ConcurrentHashMap Thread-safe한 Map, 동시에 여러 스레드가 접근하더라도 데이터를 조작할 수 있도록 보장한다. [참고] https://pplenty.tistory.com/17

➡️ saveEventCache 혹시나 연결이 끊어졌을 경우 이벤트가 발생해서 클라이언트에게 전송이 가지 않는다면 여기에 저장하고 다시 구독할 때 전송할 수 있도록 한다. SseEmitter 객체의 유실을 막기 위함이다.

✅ SubscribeService

@Service
public class SubscribeService implements SubscribeUseCase {

  private final EmitterRepository emitterRepository;

  public SubscribeService(EmitterRepository emitterRepository) {
    this.emitterRepository = emitterRepository;
  }

  @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));
    emitter.onCompletion(() -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    emitter.onTimeout(() -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    emitter.onError((ec) -> emitterRepository.deleteEmitterByEmitterId(emitterId));
    return emitter;
  }
}

// facade Service
public SseEmitter subscribe(long userId, String lastEventId) {
  String emitterId = createNotificationService.makeTimeIncludeId(userId);
  SseEmitter emitter = subScribeService.subscribe(emitterId, lastEventId);
  sendNotificationService.connectNotification(emitter, emitterId, userId);
  sendNotificationService.sendLostData(lastEventId, userId, emitterId, emitter);
  return emitter;
}

public String makeTimeIncludeId(Long userId) {
  final String UNDER_BAR = "_";
  return userId + UNDER_BAR + System.currentTimeMillis();
}

• onCompletion() 메소드는 SSE 통신이 성공적으로 완료되면 호출된다. 즉, 클라이언트와의 연결이 정상적으로 종료됨을 의미한다. 통신이 종료된 후 추가적인 작업을 수행하거나 리소스를 정리할 수 있다.

• onTimeout()은 위에서 설정한 시간, SSE 통신이 타임아웃 되었을 때 호출된다. 타임아웃은 클라이언트가 일정 시간 동안 응답을 받지 못한 경우 연결이 종료된 상태를 얘기한다. 즉, 연결 시간이 만료되면 추가 작업을 수행하거나 예외 처리를 할 수 있다.

• onCompletion(), onError()메소드는 종료 상태의 트리거이다. 둘 중 하나가 트리거 된 후에는 알림이 전송되지 않는다.

➡️ Map에 저장할 Key(String) 값은 왜 구분자와 시간으로 조합할까? Last-Event-ID는 헤더에 담겨져 오는 값으로 이전에 받지 못한 이벤트가 존재하는 경우 (SSE 연결에 대한 시간 만료 혹은 종료)나 받은 마지막 이벤트 ID 값을 넘겨 그 이후의 데이터(받지 못한 데이터) 부터 받을 수 있게 할 때 필요한 값이다.

➡️ connectNotification() SseEmitter의 유효 시간(클라-서버 연결 시간) 동안 어느 데이터도 전송되지 않는다면 503 상태 코드를 반환한다. 이에 대한 방안으로 맨 처음 연결하게 되면 Connection Message(더미 데이터)를 보내서 이를 방지한다.

✅ SendNotificationService

@Service
public class SendNotificationService implements SendNotificationUseCase {

  private final EmitterRepository emitterRepository;

  public SendNotificationService(EmitterRepository emitterRepository) {
    this.emitterRepository = emitterRepository;
  }

  @Override
  public void send(NotificationResult notificationResult) {
    emitterRepository
        .findAllEmitterStartWithByUserId(String.valueOf(notificationResult.receiveUserId()))
        .forEach(
            (key, emitter) -> {
              emitterRepository.saveEventCache(key, notificationResult);
              sendNotification(emitter, key, SendNotificationResult.from(notificationResult));
            });
  }

  @Override
  public void sendLostData(String lastEventId, Long userId, String emitterId, SseEmitter emitter) {
    if (hasLostData(lastEventId)) {
      emitterRepository
          .findAllEventCacheStartWithByUserId(String.valueOf(userId))
          .entrySet()
          .stream()
          .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
          .forEach(entry -> sendNotification(emitter, entry.getKey(), entry.getValue()));
    }
  }

  @Override
  public void connectNotification(SseEmitter emitter, String emitterId, Long userId) {
    sendNotification(emitter, emitterId, "Connected to Server. [userId=" + userId + "]");
  }

  private void sendNotification(SseEmitter emitter, String emitterId, Object data) {
    try {
      emitter.send(SseEmitter.event().id(emitterId).data(data, MediaType.APPLICATION_JSON));
    } catch (Exception ex) {
      emitterRepository.deleteEmitterByEmitterId(emitterId);
      emitter.completeWithError(ex);
    }
  }

  private boolean hasLostData(String lastEventId) {
    return !lastEventId.isEmpty();
  }
}

➡️ sendLostData() 미수신한 데이터를 전송하는 메소드. 클라이언트가 미수신한 이벤트 목록이 존재할 경우, subScribe 할 때 클라이언트에게 전송시켜 유실을 방지한다.

이 외에도 알림을 읽는 서비스, 삭제, 전체 조회하는 서비스들을 추가로 구현했다.

✅ Controller

@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public ResponseEntity<SseEmitter> subscriber(
    @AuthenticationPrincipal User user,
    @RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "")
        String lastEventId,
    HttpServletResponse response) {
  response.addHeader("X-Accel-Buffering", "no");
  var userId = Long.parseLong(user.getUsername());
  var resEmitter = notificationService.subscribe(userId, lastEventId);
  return ResponseEntity.status(HttpStatus.OK).body(resEmitter);
}

생각해야 할 부분

1. Repository에서 Map으로 SseEmitter 객체를 저장할 때 scale-out된 환경에서 잘 동작할까?

2. Map에 저장된 자료를 entrySet()으로 돌면서 빈번하게 이벤트가 발생한다면 매번 O(N)으로 탐색을 진행해야 하는데 성능적인 문제가 발생하지 않을까?

3. HTTP 2.0을 적용 하는것은 어떨까?

4. Nginx에서 발생하는 문제점들