Spring Boot - Server Sent Event를 사용한 실시간 알림

💡 Server Sent Event란?

• HTTP 스트리밍을 통해 서버에서 클라이언트에게 단방향으로 알림을 전송할 수 있는 HTML5 표준 기술입니다.
• EventStream의 최대 개수는 HTTP/1.1 사용시 6개, 간단히 말해 크롬탭을 6개까지 이용가능하며 HTTP/2 사용시 최대 100개까지 유지할 수 있다고 합니다.
• JavaScript의 EventSource를 사용하여 커넥션을 맺을 수 있으며, 접속에 문제가 발생한 경우 자동으로 재시도를 하는 특징을 가지고 있습니다.
• IE에서는 EventSource를 기본적으로 제공하고 있지 않지만 polyfill이라는 것을 사용하여 구현할 수 있지만 이제 IE는 떠났으며 걱정하지 않아도 괜찮지 않을까? 합니다.
• 클라이언트에서 페이지를 닫아도 서버가 이를 체킹하기 어렵습니다. 소켓을 사용하면 disconnect 이벤트를 알 수 있지만 SSE는 어렵습니다.

 

💡 구현하고자 하는 전체적인 그림

• 상황은 간단합니다. 사용자 A가 글을 등록했을 경우 사용자 B 또는 SseEmitter에 등록된 전체 사용자에게 알림을 보내는 것입니다.

전체적인 구조

 

💡 Spring boot Source

 

컨트롤러 

• 클라이언트에서 EventSource를 보낼 때 요청을 받아들이기 위한 컨트롤러가 필요합니다. 
• sse 통신을 하기위해서는 MIME 타입을 text/event-stream으로 설정을 해야합니다.
• Last-Event-ID란 sse의 커넥션 만료 시간 및 네트워크 오류로 인해 연결이 끊어졌을 경우 클라이언트는 그 시간동안 데이터가 유실될 수 있습니다. 그로인해 클라이언트가 받은 마지막 아이디를 기준으로 그동안의 유실된 데이터를 다시 보낼 수 있습니다.

@RestController
@RequestMapping("/notification")
@RequiredArgsConstructor
public class NotificationApi {

    private final NotificationService notificationService;

    @CrossOrigin("*")
    @GetMapping(value = "/subscribe", produces = "text/event-stream")
    public SseEmitter subscribe(@RequestParam(name = "userId") Long userId,
                                @RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "") String lastEventId) {
        return notificationService.subscribe(userId, lastEventId);
    }
}

 

서비스

• userId를 사용하여 emitterId를 만드는데 만드는 형식은 userId + currentTimeMillis 입니다. 이렇게 만드는 이유는 위에서 언급한 
  Last-Event-ID와 관련이 있으며 Last-Event-ID는 클라이언트가 마지막으로 받은 이벤트의 아이디입니다. 이렇게 두는 이유는 아래에서 설명하겠씁니다.
• 나머지 코드는 천천히 살펴보면 이해가 될것입니다.

// Last-Event-Id가 중복이 되면 어떤 데이터가 마지막으로 보낸 데이터인지 확인할 수 없습니다.
Last-Event-Id = 10
{10, data1}
{10, data3}
{10, data2}

// Last-Event-Id를 시간과 함께 섞어 설정하면 어떤 데이터를 보냈는지 순서를 파악할 수 있습니다.
Last-Event-Id = 10_1631593143664

{10_2931593143664, data1}
{10_4031593143664, data3}
{10_1831593143664, data2}

 

@Getter
@AllArgsConstructor
public class NotificationDTO {

    private String text;
}

@Service
@RequiredArgsConstructor
public class NotificationService {

    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    private final Map<String, NotificationDTO> eventCaches = new ConcurrentHashMap<>();
    private static final Long EMITTER_SESSION_TIME = (60 * 60 * 1000) * 3L; // 3시간

    // 브라우저에서 eventsource로 연결할때마다 emitter 객체 생성하여 등록
    public SseEmitter subscribe(Long userId, String lastEventId) {
        String emitterId = makeIdByUserId(userId);
        SseEmitter emitter = new SseEmitter(EMITTER_SESSION_TIME);

        // SseEmitter의 시간초과 및 네트워크 오류를 포함한 비동기 요청이 정상적으로 동작할 수 없다면 저장해둔 SseEmitter 삭제
        emitter.onCompletion(() -> deleteByEmitterId(emitterId));
        emitter.onTimeout(() -> deleteByEmitterId(emitterId));
        emitter.onError((e) -> deleteByEmitterId(emitterId));
        saveNewEmitter(emitterId, emitter);

        // 연결 후 아무런 응답값을 보내주지 않으면 오류 발생하므로 더미 데이터 반환
        // 503 error를 방지하기 위한 더미 이벤트 전송
        String eventId = makeIdByUserId(userId);
        sendNotification(emitter, eventId, new NotificationDTO("더미 데이터"));

        // 클라이언트가 미수신한 Event가 있을 경우 전송하여 Event 유실 방지
        // 즉 연결이 끊기고 재 연결되었을 경우 lastEventId로 마지막 이벤트 찾음
        if (hasLatestNotification(lastEventId)) {
            sendLatestNotification(emitter, lastEventId, emitterId);
        }
        return emitter;
    }

    // 특정 클라이언트에게 알림 발송
    public void send(Long userId, NotificationDTO notificationDTO) {
        Map<String, SseEmitter> emitters = findEmitterByEmitterId(userId.toString());
        emitters.forEach((key, value) -> {
            saveEventCache(key, notificationDTO);
            sendNotification(value, key, notificationDTO);
        });
    }

    // 모든 클라이언트에게 알림 발송
    public void sendBroadcast(NotificationDTO notificationDTO) {
        Map<String, SseEmitter> broadcast = findAllEmitterBroadcast();
        broadcast.forEach((key, value) -> {
            saveEventCache(key, notificationDTO);
            sendNotification(value, key, notificationDTO);
        });
    }

    // 기존에 등록되어 있던 emitter 삭제 후 등록
    private void saveNewEmitter(String emitterId, SseEmitter emitter) {
        deleteAllEmitterByEmitterId(emitterId);
        emitters.put(emitterId, emitter);
    }

    // 캐시 저장
    private void saveEventCache(String eventId, NotificationDTO event) {
        eventCaches.put(eventId, event);
    }

    private Map<String, SseEmitter> findEmitterByEmitterId(String emitterId) {
        return emitters.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(emitterId))
                .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    private Map<String, NotificationDTO> findAllEventCacheByEmitterId(String emitterId) {
        return eventCaches.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(emitterId))
                .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    // 등록된 모든 클라이언트 반환
    private Map<String, SseEmitter> findAllEmitterBroadcast() {
        return emitters.entrySet().stream()
                .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    // 등록된 Emitter 삭제
    private void deleteByEmitterId(String emitterId) {
        emitters.remove(emitterId);
    }

    // 유저와 관련된 모든 Emitter 삭제
    private void deleteAllEmitterByEmitterId(String emitterId) {
        emitters.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(emitterId))
                .forEach(entry -> emitters.remove(entry.getKey()));
    }

    // 유저와 관련된 캐시 삭제
    private void deleteAllEventCacheUserId(String eventId) {
        eventCaches.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(eventId))
                .forEach(entry -> eventCaches.remove(entry.getKey()));
    }

    private void sendLatestNotification(SseEmitter emitter, String lastEventId, String emitterId) {
        Map<String, NotificationDTO> eventCaches = findAllEventCacheByEmitterId(emitterId);
        eventCaches.entrySet().stream()
                .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                .forEach(entry -> sendNotification(emitter, emitterId, entry.getValue()));
        deleteAllEventCacheUserId(lastEventId);
    }

    private void sendNotification(SseEmitter emitter, String emitterId, NotificationDTO data) {
        try {
            emitter.send(SseEmitter
                    .event()
                    .id(emitterId)
                    .data(data)
            );
        } catch (IOException e) {
            deleteByEmitterId(emitterId);
        }
    }

    private String makeIdByUserId(Long userId) {
        return userId + "_" + System.currentTimeMillis();
    }

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

 

💡 Client Source

• JavaScript에서는 EventSource를 사용하여 서버와 연결을 맺고 데이터를 받을 수 있습니다.

<script type="text/javaScript">
    const userId = 10;
    const eventSource = new EventSource(`http://localhost:8080/notification/subscribe?userId=${userId}`);
    eventSource.onmessage = function (event) {
        console.log(event.data);
    };

    eventSource.onerror = (e) => {
        console.log(e);
    }
</script>

연결 후 응답 결과

 

👍 중간 정리

• 여기까지 Server-Sent-Event를 사용하여 클라이언트와 서버가 연결을 맺고 알림을 받을 수 있다!는 것까지는 알게되었습니다.
  지금부터는 글 작성시 알림을 받을 수 있도록하는 상황을 가정하겠습니다.
• 여기서부터는 클라이언트 소스는 동일하니 서버 코드를 살펴보겠습니다.

 

💡 글 작성 시 이벤트 발송(서버 코드)

• 필요한 작업
  1. 글 작성시 예외가 발생하면 알림을 보내지 않습니다.
  2. 글 작성이 COMMIT되면 알림을 보냅니다.
  3. 비동기로 알림을 발송합니다.

 

글 작성

• 아래는 글 작성 후 이벤트가 발생하게 됩니다.

public enum ActionType {
    UNICAST,
    BROADCAST
}

@Service
@Transactional
@RequiredArgsConstructor
public class PostsService {

    private final PostsRepository postsRepository;
    private final ApplicationEventPublisher applicationEventPublisher;

    public Long create() {
        Posts posts = Posts.builder()
                .title("제목은?")
                .contents("내용은?")
                .build();
        postsRepository.save(posts);
        applicationEventPublisher.publishEvent(ActionType.BROADCAST); // 이벤트 발생
        return posts.getId();
    }
}

 

이벤트 리스너

• 사용자가 글을 작성했을 경우 예외가 발생하지 않는다면 이벤트 리스너가 호출됩니다. @TransactionEventListener 어노테이션을 사용하여 @Transactional 어노테이션이 선언된 클래스 레벨 혹은 메서드 레벨에서 커밋이 된 후에 해당 이벤트가 발생하게 됩니다.
  여기서 @EventListener 어노테이션을 사용하게 된다면 Posts 등록 시 예외가 발생하더라도 이벤트가 발생하게 됩니다.
• 이벤트의 액션 타입에 맞게 특정 클라이언트에게 발송할건지 또는 모든 클라이언트에게 발송할건지 분기 처리합니다.
• 알림 발송 시 모든 사용자에게 알림을 보내는 과정을 비동기로 처리할 수 있도록 @Async 어노테이션을 사용합니다.

@Component
@RequiredArgsConstructor
public class SseEventListener {

    private static final String TEXT = "새로운 알림이 도착했습니다.";
    private final NotificationService notificationService;

    @Async
    @TransactionalEventListener
    public void eventHandler(ActionType actionType) {
        if (actionType == ActionType.UNICAST) {
            notificationService.send(15L, new NotificationDTO(TEXT));
        }
        if (actionType == ActionType.BROADCAST) {
            notificationService.sendBroadcast(new NotificationDTO(TEXT));
        }
    }
}

 

비동기 처리 Config

• 알림을 비동기 방식으로 발송하기 위해 Async를 설정해줍니다.
• 비동기에 대해 자세한 내용은 아래 링크를 첨부하겠습니다.
• 비동기의 동작 방식  링크를 참고해주세요!
• @Async 설정 방법  링크를 참고해주세요!

@Configuration
@EnableAsync
public class AsyncConfig extends AsyncConfigurerSupport {

    private static final int CORE_POOL_SIZE = 10;
    private static final int MAX_POOL_SIZE = 90;
    private static final int QUEUE_CAPACITY = 10;
    private static final boolean TASKS_TO_COMPLETE_ON_SHUTDOWN = true;
    private static final int AWAIT_TERMINATION_SECONDS = 60;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setThreadNamePrefix("async-");
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setWaitForTasksToCompleteOnShutdown(TASKS_TO_COMPLETE_ON_SHUTDOWN);
        executor.setAwaitTerminationSeconds(AWAIT_TERMINATION_SECONDS);
        executor.setRejectedExecutionHandler((r, e) -> {
            throw new IllegalStateException("동시 요청수가 많습니다. 잠시후에 다시 요청해주세요.");
        });
        executor.initialize();
        return executor;
    }
}

 

 

😂 구현 후 알게된 Server-Sent-Event의 사실

• AWS API Gateway를 사용한다면 SSE를 적용할 수 없습니다. 구현 후 개발 서버에 배포하면서 계속하여 504 에러가 발생했는데 
  구글링 결과 지원을 하지 않는다고하여 rabbitMQ를 적용하여 개발하고 있습니다..
  https://github.com/streamdata-serverless/streamdata-io-basic-demo-stockmarket-prices/issues/1
• 다음편에는 rabbitMQ와 MQTT를 사용하여 알림을 발송하는 편으로 돌아오겠습니다.!