개요

HTTP 통신의 특성

반드시 사용자 → 서버의 흐름으로 진행된다.
즉, 사용자가 서버에게 GET/POST 요청을 하면 서버는 “요청받은 것”만 처리해서 응답한다.

그런데 실시간 기술(통신)은 사용자의 추가 요청이 없는 상태에서도, 서버가 화면에 데이터를 꽂아주는 경우가 있다.
이런 실시간 구현 방식은 크게 SSE, WebSocket이 있다.

SSE vs WebSocket

채팅처럼 매우 빈번한 실시간 통신은 보통 WebSocket이 필요하다.
SSE는 간단한 알림(서버→클라이언트 푸시)에 적합하고, HTTP 기반으로 구현된다.
WebSocket은 양방향 통신을 위해 별도 프로토콜로 연결을 유지하는 방식이라 SSE보다 구현 난이도가 올라가지만, 실시간 상호작용에는 더 적합하다.

---

SSE

SSE(Server-Sent Events)는 서버가 클라이언트에게 이벤트를 지속적으로 전송할 수 있게 해주는 웹 기술이다.
주로 실시간 업데이트/알림이 필요한 서비스에서 사용한다.

SSE 특징

클라이언트가 서버와 연결을 맺으면 서버는 해당 연결을 유지하면서 필요한 데이터를 계속 푸시할 수 있다.
단방향 통신(서버 → 클라이언트)이며, 클라이언트가 서버로 데이터를 보내는 것은 SSE로는 불가능하므로 별도의 HTTP 요청으로 처리해야 한다.

구현이 비교적 간단하다.
Spring MVC에서는 SseEmitter를 통해 SSE 이벤트를 전송할 수 있고, 이벤트 이름/ID/data 등을 빌더로 구성해서 send()로 밀어넣는 식으로 동작한다.

그리고 중요한 포인트가 하나 있다.
“서버가 사용자에게 푸시를 하려면” 서버가 해당 사용자의 연결 정보(정확히는 emitter 같은 연결 핸들)를 잡고 있어야 한다.
즉, 알림을 받아야 하는 클라이언트는 먼저 connect 요청으로 연결을 만들어 둬야 한다.

추가로 SSE는 브라우저가 재연결을 시도할 수 있고, 이벤트에 id:를 부여하면 재연결 시 Last-Event-ID 헤더를 통해 유실 복구 설계를 할 수 있다(서버가 그 ID 기준으로 누락분을 다시 보내주도록 구현해야 함).

---

Redis Pub/Sub 기반 SSE 동작 흐름

문제 상황(멀티 서버)

구매자 B가 상품을 구매했을 때 판매자(또는 admin) A에게 알림을 주고 싶은 상황이다.
단일 서버면 “A의 emitter 객체”를 찾아서 바로 send() 하면 끝난다.

하지만 서버가 2대로 확장되면 문제가 생긴다.
예를 들어 A는 서버1에 연결되어 emitter가 서버1 메모리에만 존재하고, B는 서버2로 주문 요청을 날릴 수 있다.
이때 서버2에는 A emitter가 없으니 서버2 단독으로는 알림을 못 보낸다.

해결책(메시지 전파기)

멀티 서버 환경에서 원활한 알림을 위해 “모든 서버에 메시지를 전파”해주는 Redis Pub/Sub을 사용한다.
여기서 Redis는 emitter를 저장하는 저장소가 아니라, 메시지를 퍼뜨리는 브로커 역할을 한다.

왜 emitter를 Redis/RDB에 넣지 않느냐?
SseEmitter 같은 객체는 단순 문자열처럼 저장했다가 꺼내 쓰는 구조가 아니고, 결국 “현재 서버 프로세스가 들고 있는 연결 핸들”이기 때문에 외부 저장소에 넣어봤자 재사용이 불가능하다.
결론적으로 emitter는 각 서버 로컬 메모리에 들고, 메시지만 Redis로 전파하는 구조가 맞다.

Pub/Sub 특징

• 채널이라는 논리적 공간을 통해 publish/subscribe 수행한다
• 발행된 메시지는 큐처럼 쌓여 저장되지 않으며, 구독 중인 클라이언트에게만 전달된다(구독자가 없거나, 다운돼서 못 받으면 유실될 수 있음).
• 장점: 구조가 단순하고 빠르다.
• 단점: 메시지 유실 가능성이 있다(중요도가 높은 메시지는 Kafka/Streams 같은 “저장 가능한” 방식 고려).

---

실습 시나리오

일반 사용자 → 주문 → admin 알림 메시지 수신 흐름이다.

• admin이 먼저 SSE connect 요청을 해서 연결(emitter)을 서버에 등록한다.
• 이후 일반 사용자가 주문을 생성하면, 서버가 admin에게 알림 메시지를 푸시한다.
• 단일 서버면 emitter를 찾아서 바로 send하면 끝난다.
• 멀티 서버면 “admin emitter가 어느 서버에 붙어있는지” 문제가 생기므로 Redis Pub/Sub로 메시지를 전 서버에 전파해서 해결한다.

1) SSE Emitter 객체 저장소

실습에서는 SseEmitterRegistry를 Map 기반 저장소로 만든다.
실제로 운영환경에서는 “서버 메모리에 emitter를 들고 있는 것” 자체가 한계가 있지만, SSE 구조 이해용으로는 이 방식이 가장 직관적이다.

package com.beyond.order_system.common.repository;

import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class SseEmitterRegistry {
    /*
     * [SSE emitter 객체]
     * - 사용자의 연결정보(정확히는 서버가 쥐고 있는 연결 핸들)를 의미
     *
     * [ConcurrentHashMap]
     * - 스레드 세이프한 구조
     * */

    private Map<Long, SseEmitter> emitterMap = new ConcurrentHashMap<>();

    public void addSseEmitter(Long id, SseEmitter sseEmitter) {
        this.emitterMap.put(id, sseEmitter);
    }

    public SseEmitter getEmitter(Long id) {
        return this.emitterMap.get(id);
    }

    public void removeSseEmitter(Long id) {
        this.emitterMap.remove(id);
    }
}

---

2) 공통 알림 메시지 DTO 설계

SSE로 전송할 “공통 메시지 포맷”을 DTO로 정의한다.
멀티 서버 환경에서는 이 DTO를 JSON 문자열로 serialize 해서 Redis 채널로 publish 하고, 다른 서버는 그걸 다시 deserialize 해서 처리한다.

package com.beyond.order_system.common.dto;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@AllArgsConstructor
@NoArgsConstructor
@Data
@Builder
public class SseMessageDto {
    private Long receiverId;
    private Long senderId;
    private String message;
}

---

3) SseController 연결확인 API(+ disconnect 처리)

클라이언트가 /sse/connect로 연결을 맺으면 서버는 SseEmitter를 만들고 registry에 저장한다.
그리고 브라우저를 끄거나 네트워크가 끊겼는데 emitter를 정리하지 않으면 emitter 객체가 계속 남아서 서버 부하가 생길 수 있으므로, onCompletion/onTimeout/onError에서 반드시 remove 해준다.

package com.beyond.order_system.common.controller;

import com.beyond.order_system.common.repository.SseEmitterRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

@RestController
@RequestMapping("/sse")
public class SseController {

    private final SseEmitterRegistry sseEmitterRegistry;

    @Autowired
    public SseController(SseEmitterRegistry sseEmitterRegistry) {
        this.sseEmitterRegistry = sseEmitterRegistry;
    }

    @GetMapping("/connect")
    public SseEmitter connect(@AuthenticationPrincipal String principal) throws IOException {
        Long id = Long.parseLong(principal);

        // 유효시간: 1시간
        SseEmitter sseEmitter = new SseEmitter(60 * 60 * 1000L);
        sseEmitterRegistry.addSseEmitter(id, sseEmitter);

        // disconnect 처리 (정리 훅)
        sseEmitter.onCompletion(() -> sseEmitterRegistry.removeSseEmitter(id)); // 정상 종료 [web:38]
        sseEmitter.onTimeout(() -> {
            sseEmitterRegistry.removeSseEmitter(id);
            sseEmitter.complete(); // 타임아웃 종료 [web:38]
        });
        sseEmitter.onError((e) -> {
            sseEmitterRegistry.removeSseEmitter(id);
            sseEmitter.completeWithError(e); // 에러 종료 [web:38]
        });

        // 연결 확인용 더미 이벤트
        sseEmitter.send(SseEmitter.event().name("connect").data("연결완료"));
        return sseEmitter;
    }
}

---

4) Redis Pub/Sub로 멀티 서버 알림 개선

멀티 서버 환경에서의 알림 문제

서버가 2대로 확장된 상황에서, admin(A)의 SSE 연결은 서버1에 붙어 emitter가 서버1 메모리에만 생성되어 있을 수 있다.
그런데 일반 사용자(B)가 서버2로 주문 요청을 넣으면, 서버2는 admin emitter를 찾으려고 해도 서버2 메모리에는 emitter가 없다.

이 문제는 “메모리에만 상태(emitter)를 들고 있기 때문에 생기는 문제”다.
그래서 해결책은 “메시지를 모든 서버에 전파”해주는 Pub/Sub 구조를 도입하는 것이다.

Redis Pub/Sub 특징

• 채널이라는 가상의 루트를 통해 SUB/PUB 수행한다.
• DB 선택 로직이 없다(저장 기반이 아니라 브로드캐스트 성격이기 때문).
• 한번 발송된 메시지는 저장되지 않는다(즉, 그 시점에 구독 중인 서버만 받는다).
• 장점: 빠르다.
• 단점: 메시지 유실 가능성이 있다(구독 서버가 다운이면 놓친다).

---

5) RedisConfig(pub/sub 세팅 포함)

각 서버는 특정 채널(예: order-channel)을 subscribe 하고, 주문 이벤트가 발생한 서버는 publish 한다.
그리고 수신한 서버는 “내 서버에 receiver emitter가 있으면” 그때 send 하는 구조다.

package com.beyond.order_system.common.config;

import com.beyond.order_system.common.service.SseAlarmService;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    /*
     * [연결 빈객체 : RedisConnectionFactory]
     * - redis에 대한 연결 정보(Host, Port, DB 번호)
     *
     * [템플릿 빈객체 : RedisTemplate]
     * - redis 자료구조 접근을 위한 템플릿
     *
     * [@Qualifier]
     * - 같은 타입 Bean이 여러개 있을 경우 구분하기 위한 장치
     * */

    /* *********************** Redis 연결 (일반 저장용) *********************** */
    @Bean
    @Qualifier("rtInventory")
    public RedisConnectionFactory redisConnectionFactory() {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(host);
        configuration.setPort(port);
        configuration.setDatabase(0);
        return new LettuceConnectionFactory(configuration);
    }

    @Bean
    @Qualifier("rtInventory")
    public RedisTemplate<String, String> redisTemplate(
            @Qualifier("rtInventory") RedisConnectionFactory redisConnectionFactory
    ) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        return redisTemplate;
    }

    @Bean
    @Qualifier("stockInventory")
    public RedisConnectionFactory redisStockConnectionFactory() {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(host);
        configuration.setPort(port);
        configuration.setDatabase(1);
        return new LettuceConnectionFactory(configuration);
    }

    @Bean
    @Qualifier("stockInventory")
    public RedisTemplate<String, String> redisStockTemplate(
            @Qualifier("stockInventory") RedisConnectionFactory redisConnectionFactory
    ) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        return redisTemplate;
    }

    /* *********************** SSE PUB/SUB 세팅 *********************** */
    @Bean
    @Qualifier("ssePubSub")
    public RedisConnectionFactory ssePubSubConnectionFactory() {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(host);
        configuration.setPort(port);
        // pub/sub은 저장 목적이 아니라 메시지 전파 목적
        return new LettuceConnectionFactory(configuration);
    }

    @Bean
    @Qualifier("ssePubSub")
    public RedisTemplate<String, String> redisSsePubSubTemplate(
            @Qualifier("ssePubSub") RedisConnectionFactory redisConnectionFactory
    ) {
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        return redisTemplate;
    }

    // RedisMessageListenerContainer -> MessageListenerAdapter -> SseAlarmService(onMessage 위임)
    @Bean
    @Qualifier("ssePubSub")
    public RedisMessageListenerContainer redisMessageListenerContainer(
            @Qualifier("ssePubSub") RedisConnectionFactory redisConnectionFactory,
            @Qualifier("ssePubSub") MessageListenerAdapter messageListenerAdapter
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(messageListenerAdapter, new PatternTopic("order-channel"));
        return container;
    }

    @Bean
    @Qualifier("ssePubSub")
    public MessageListenerAdapter messageListenerAdapter(SseAlarmService sseAlarmService) {
        return new MessageListenerAdapter(sseAlarmService, "onMessage");
    }
}

---

6) 메시지 송신 절차(Order → SSE Alarm)

주문이 완료되면 admin에게 알림을 보내고 싶다.
즉, 주문 로직이 끝나는 순간 sseAlarmService.sendMessage(adminId, memberId, message)를 호출한다.

// 주문 성공시 admin 유저에게 알림메시지 발송
String message = order.getId() + "번 주문이 들어왔습니다.";
sseAlarmService.sendMessage(1L, memberId, message);

---

7) SseAlarmService(로컬 send → 없으면 Redis publish)

핵심 아이디어는 이거다.

• receiver emitter가 현재 서버에 있으면: 바로 send()로 알림 발송
• receiver emitter가 현재 서버에 없으면: Redis 채널로 publish
• 모든 서버는 subscribe 중이므로 메시지를 받고, receiver emitter가 있는 서버만 최종 send

Spring에서 SseEmitter는 send(...), complete(...), onTimeout(...) 같은 API로 이벤트 스트리밍을 구현한다.
그리고 Redis Pub/Sub은 메시지를 저장하지 않기 때문에, 그 순간 구독 중인 서버에게만 퍼진다.

package com.beyond.order_system.common.service;

import com.beyond.order_system.common.dto.SseMessageDto;
import com.beyond.order_system.common.repository.SseEmitterRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;

@Component
public class SseAlarmService implements MessageListener {

    private final SseEmitterRegistry sseEmitterRegistry;
    private final ObjectMapper objectMapper;
    private final RedisTemplate<String, String> redisTemplate;

    @Autowired
    public SseAlarmService(SseEmitterRegistry sseEmitterRegistry,
                           ObjectMapper objectMapper,
                           @Qualifier("ssePubSub") RedisTemplate<String, String> redisTemplate) {
        this.sseEmitterRegistry = sseEmitterRegistry;
        this.objectMapper = objectMapper;
        this.redisTemplate = redisTemplate;
    }

    public void sendMessage(Long receiverId, Long senderId, String message) {
        SseMessageDto dto = SseMessageDto.builder()
                .receiverId(receiverId)
                .senderId(senderId)
                .message(message)
                .build();

        try {
            String data = objectMapper.writeValueAsString(dto);

            // 1) 현재 서버에 receiver emitter가 있으면 바로 send
            SseEmitter sseEmitter = sseEmitterRegistry.getEmitter(receiverId);
            if (sseEmitter != null) {
                sseEmitter.send(SseEmitter.event().name("ordered").data(data));
            } else {
                // 2) 없으면 Redis Pub/Sub 채널로 publish (전 서버로 전파)
                redisTemplate.convertAndSend("order-channel", data);
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    // 3) Redis subscribe 메시지 수신 시 처리
    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            SseMessageDto dto = objectMapper.readValue(message.getBody(), SseMessageDto.class);

            // 내 서버에 receiver emitter가 있으면 최종 send
            SseEmitter sseEmitter = sseEmitterRegistry.getEmitter(dto.getReceiverId());
            if (sseEmitter != null) {
                String data = objectMapper.writeValueAsString(dto);
                sseEmitter.send(SseEmitter.event().name("ordered").data(data));
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

---

8) 메시지 수신 절차(정리)

각 서버에서 Redis Pub/Sub의 특정 채널을 구독하기 위한 RedisMessageListenerContainer 객체를 생성한다.
MessageListenerAdapter가 컨테이너의 리스너로 등록되고, 메시지가 들어오면 SseAlarmService.onMessage로 위임된다.
그리고 최종적으로 “해당 서버에 receiver emitter가 있으면” 그 receiver에게 SSE 이벤트를 send 한다.