이번 포스팅에서는 실제로 SSE를 구현해보도록 하겠습니다!
- 이전글
2025.10.14 - [자바/스프링부트] - [JAVA / SSE] 스프링부트에서 SSE 구현하기 (SSE + Redis pub/sub) - 1
[JAVA / SSE] 스프링부트에서 SSE 구현하기 (SSE + Redis pub/sub) - 1
실시간 알림 기능은 이제 대부분의 웹 서비스에 들어가는 기술입니다. 예를 들어, 메시지 알림이나 실시간 데이터 갱신 같은 기능들이 대표적이죠. 그렇다면, 이를 구현하고자 한다면 가장 먼저
huffpuffkin.tistory.com
전체 흐름
일단 전체적인 로직 흐름을 정리해보자면 다음과 같습니다.
| 1) 클라이언트가 /api/notification/conn 엔드포인트에 접근해 emitter 생성 2) 서버 → 알림 생성 & Redis에 저장 (TTL 포함) 3) RedisPublisher → pub/sub 채널에 발행 (Publish) 4) RedisSubscriber → Redis 메시지 수신 5) SseManager → 연결된 클라이언트로 SSE 알림 전송 6) 일정 시간이 지나면 Redis key 자동 만료 7) RedisExpirationListener → 만료 이벤트 감지 & DB 업데이트 |
SSEManager
실질적으로 클라이언트와 연결되는 emitter를 관리하는 클래스입니다. 유저 별 다중 SSE 연결 관리 + 주기적 하트비트 + 에러/종료 정리의 역할을 수행합니다.
@Slf4j
@Component
@RequiredArgsConstructor
public class SseManager {
private final ScheduledThreadPoolExecutor heartbeatPool;
//userPk, emitters
private final Map<Long, CopyOnWriteArrayList<SseEmitter>> emitterList = new ConcurrentHashMap<>();
//emitter, ScheduledFuture
private final Map<SseEmitter, ScheduledFuture<?>> heartbeatTasks = new ConcurrentHashMap<>();
/* emitter 생성 */
public SseEmitter createEmitter(Long userPk) {
SseEmitter emitter = new SseEmitter(0L);
emitterList.computeIfAbsent(userPk, k -> new CopyOnWriteArrayList<>()).add(emitter);
//15초 주기 하트비트
ScheduledFuture<?> f = heartbeatPool.scheduleAtFixedRate(
() -> sendHeartBeat(userPk, emitter), 1, 15, TimeUnit.SECONDS
);
heartbeatTasks.put(emitter, f);
//실행 종료 시 실행 runnable
Runnable r = () -> cleanUpEmitter(userPk, emitter);
//연결 종료 시
emitter.onCompletion(r);
//타임아웃 시
emitter.onTimeout(r);
//에러 발생 시
emitter.onError(e -> r.run());
return emitter;
}
/* heartbeat 전송 */
public void sendHeartBeat(Long userPk, SseEmitter emitter) {
try {
emitter.send(SseEmitter.event().name("heartbeat").data("ok"));
} catch(Exception e) {
//전송 중 오류: emitter 삭제
var task = heartbeatTasks.remove(emitter);
//스레드 실행 취소
if(task != null) {
task.cancel(true);
}
//emitter 등록 취소
emitterList.getOrDefault(userPk, new CopyOnWriteArrayList<>()).remove(emitter);
//sse연결을 에러와 함께 즉시 취소
try {
// emitter.completeWithError(e);
} catch(Exception ignore) {}
}
}
/* 메시지 전송 */
public void sendNotificationToEmitter(Long userPk, PushNotificationDTO notification) {
List<SseEmitter> emitters = emitterList.getOrDefault(userPk, new CopyOnWriteArrayList<>());
if(!emitters.isEmpty()) {
List<SseEmitter> deadEmitters = new ArrayList<>();
for(SseEmitter emitter : emitters) {
try {
emitter.send(SseEmitter.event()
.name("alarm")
.data(notification));
} catch (Exception e) {
log.error("Error Sending SSE to user: {} with message: {}", userPk, notification);
deadEmitters.add(emitter);
}
}
log.info("SSE: success");
deadEmitters.forEach(i -> cleanUpEmitter(userPk, i));
} else {
log.warn("No emitter found for user: {}", userPk);
}
}
/* clean up */
public void cleanUpEmitter(Long userPk, SseEmitter emitter) {
//hearbeatTask에서 삭제
var task = heartbeatTasks.remove(emitter);
if(task != null) {task.cancel(true);}
var list = emitterList.get(userPk);
if (list != null) {
list.remove(emitter);
if (list.isEmpty()) emitterList.remove(userPk, list);
}
try { emitter.complete(); } catch (Exception ignore) {}
}
}
주요 필드
- emitterList: 한 유저가 여러 브라우저 탭/모바일에서 동시에 여러 SSE 연결을 가질 수 있으므로 리스트로 보관합니다.
- heartbeatTasks: 각 emitter마다 등록된 반복 하트비트 작업의 핸들을 보관합니다.
- heartbeatPool: 하트비트(keep-alive) 스케줄링 전용 풀입니다.
Emitter 생성 메서드
public SseEmitter createEmitter(Long userPk) {
SseEmitter emitter = new SseEmitter(0L);
emitterList.computeIfAbsent(userPk, k -> new CopyOnWriteArrayList<>()).add(emitter);
ScheduledFuture<?> f = heartbeatPool.scheduleAtFixedRate(
() -> sendHeartBeat(userPk, emitter), 1, 15, TimeUnit.SECONDS
);
heartbeatTasks.put(emitter, f);
Runnable r = () -> cleanUpEmitter(userPk, emitter);
emitter.onCompletion(r);
emitter.onTimeout(r);
emitter.onError(e -> r.run());
return emitter;
}- 타임아웃을 0L로 설정하여 SseEmitter 생성, 서버 관점에서는 연결을 무제한으로 유지.
- 하트비트 스케줄 등록: 생성 1초 뒤 시작하여 15초 주기로 하트비트 전송
- 종료 훅: Emitter 종료/타임아웃/에러 발생 시에 동일하게 정리 로직을 실행
하트비트 전송 메서드
프록시/NGINX/브라우저가 유휴 연결을 끊지 않도록 주기적으로 데이터를 보내는 메서드입니다.
public void sendHeartBeat(Long userPk, SseEmitter emitter) {
try {
emitter.send(SseEmitter.event().name("heartbeat").data("ok"));
} catch(Exception e) {
var task = heartbeatTasks.remove(emitter);
if(task != null) task.cancel(true);
emitterList.getOrDefault(userPk, new CopyOnWriteArrayList<>()).remove(emitter);
try {
// emitter.completeWithError(e);
} catch(Exception ignore) {}
}
}실패 시 해당 emitter의 하트비트 작업을 즉시 취소하고 emitterList에서 제거합니다.
알림 전송 메서드
public void sendNotificationToEmitter(Long userPk, PushNotificationDTO notification) {
List<SseEmitter> emitters = emitterList.getOrDefault(userPk, new CopyOnWriteArrayList<>());
if(!emitters.isEmpty()) {
List<SseEmitter> deadEmitters = new ArrayList<>();
for(SseEmitter emitter : emitters) {
try {
emitter.send(SseEmitter.event().name("alarm").data(notification));
} catch (Exception e) {
log.error("Error Sending SSE to user: {} with message: {}", userPk, notification);
deadEmitters.add(emitter);
}
}
deadEmitters.forEach(i -> cleanUpEmitter(userPk, i));
} else {
log.warn("No emitter found for user: {}", userPk);
}
}- 브로드캐스트 방식을 활용해 유저의 모든 활성 emitter에 동일 이벤트를 전송합니다.
- 전송에 실패한 emitter만 모아서 일괄적으로 clean up을 진행합니다.
- 이벤트 이름을 alarm으로 명시해 프론트에서 쉽게 바인딩 할 수 있도록 구현했습니다.
정리 메서드
public void cleanUpEmitter(Long userPk, SseEmitter emitter) {
var task = heartbeatTasks.remove(emitter);
if(task != null) { task.cancel(true); }
var list = emitterList.get(userPk);
if (list != null) {
list.remove(emitter);
if (list.isEmpty()) emitterList.remove(userPk, list);
}
try { emitter.complete(); } catch (Exception ignore) {}
}- 하트비트 작업 핸들 제거 및 취소
- 유저 리스트에서 해당 emitter 제거, 이후 빈 리스트가 되면 엔트리도 제거
- emitter.complete()로 서버 측 스트림 종료
- 예외는 무시하고 처리
SseConfig - SSE 하트비트 스케줄러
SSE는 기본적으로 오랜시간 동안 HTTP 연결을 유지하기 때문에 주기적으로 하트비트를 보내줘야 연결이 끊어지지 않습니다.
/*
SSE 하트비트 전송용 공용 스케줄러 풀
*/
@Configuration
public class SseConfig {
@Bean(destroyMethod = "shutdown")
public ScheduledThreadPoolExecutor heartbeatPool() {
int poolSize = Math.max(2, Runtime.getRuntime().availableProcessors() / 2);
//데몬 스레드로 설정: 백그라운드용 스레드
ScheduledThreadPoolExecutor ex = new ScheduledThreadPoolExecutor(poolSize, r -> {
Thread t = new Thread(r, "sse-heartbeat-" + System.nanoTime());
t.setDaemon(true);
return t;
});
//취소된 작업을 대기 큐에서 즉시 제거
ex.setRemoveOnCancelPolicy(true);
return ex;
}
}
하트비트를 전송하는 스레드를 데몬 스레드로 설정하고 15초 간격으로 하트비트를 전송합니다.
Redis 설정
일단 저는 Lettuce를 사용해 Redis를 설정했습니다.
@Value("${spring.data.redis.host}")
private String host;
@Value("${spring.data.redis.port}")
private int port;
/* Redis 연결 객체 생성 */
@Bean
public RedisConnectionFactory redisConnectionFactory() {
return new LettuceConnectionFactory(host, port);
}설정 파일에서 host, port를 읽어와 LettuceConnectionFactory 객체를 생성해 Redis와 연결합니다.
이후 RedisTemplate을 알림 저장용(key/value)과 Pub/Sub용, 이렇게 두 가지로 구분해 구현했습니다.
/* 공용 ObjectMapper 등록 */
@Bean
public ObjectMapper redisObjectMapper() {
return new ObjectMapper()
.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule())
.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
}일단 공용 ObjectMapper를 등록해주었습니다. ObjectMapper는 Redis에 데이터를 저장하거나 전송할 때 객체를 JSON으로 직렬화하거나 반대로 역직렬화할 대 사용합니다.
알림 저장 기능을 사용할 때 데이터를 JSON으로 직렬화해서 저장하고 가져올 수 있도록 설정합니다.
/* 공용 JSON serializer 등록 - (key,val) 저장용 */
@Bean
public Jackson2JsonRedisSerializer<PushNotificationDTO> redisJsonSerializer() {
return new Jackson2JsonRedisSerializer<>(redisObjectMapper(), PushNotificationDTO.class);
}
/* RedisTemplate 빈 생성 - (key,val) 저장용 */
@Bean
public RedisTemplate<String, PushNotificationDTO> redisTemplate() {
var t = new RedisTemplate<String, PushNotificationDTO>();
t.setConnectionFactory(redisConnectionFactory());
t.setKeySerializer(new StringRedisSerializer());
t.setValueSerializer(redisJsonSerializer());
return t;
}레디스에 키를 등록할 때 String의 형태로 key-val이 저장되기 때문에 직렬화, 역직렬화는 필수적입니다.
또한 Pub/Sub 기능을 사용할 때 데이터를 JSON으로 직렬화해서 보내고 받을 수 있도록 설정합니다.
/* 공용 JSON serializer 등록 - pub/sub용 */
@Bean
public Jackson2JsonRedisSerializer<PublishDTO> pubsubSerializer() {
return new Jackson2JsonRedisSerializer<>(redisObjectMapper(), PublishDTO.class);
}
/* Redis pub/sub 메시지 리스너 컨테이너 */
@Bean
public RedisTemplate<String, PublishDTO> pubSubTemplate() {
var t = new RedisTemplate<String, PublishDTO>();
t.setConnectionFactory(redisConnectionFactory());
t.setKeySerializer(new StringRedisSerializer());
t.setValueSerializer(pubsubSerializer());
t.afterPropertiesSet();
return t;
}pubsubSerializer()는 PublishDTO 객체(객체를 pub/sub할 때 사용하는 DTO)의 직렬화, 역직렬화를 담당합니다. Pub/Sub 메시지는 문자열 또는 byte 배열로 전송되기 때문에 DTO를 JSON으로 변환하는 과정이 필수적입니다!!
예를 들자면
PublishDTO(pnPk=1, userPk=10) → {"pnPk":1,"userPk":10}위와 같이 변경해주는 기능을 수행합니다.
이후 Pub/Sub는 메시지 구독자 설정을 진행합니다.
@Value("${spring.data.redis.notification.channel}")
private String CHANNEL;
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisSubscriber redisSubscriber) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(redisConnectionFactory());
var ser = pubsubSerializer();
//구독 설정
container.addMessageListener((message, pattern) ->{
String ch = new String(message.getChannel(), java.nio.charset.StandardCharsets.UTF_8);
Object payload = ser.deserialize(message.getBody());
redisSubscriber.onMessage(ch, (PublishDTO) payload);
},
new PatternTopic(CHANNEL));
return container;
}설정 파일에서 알림 구독에 사용할 채널명을 읽어온 다음 해당 채널을 구독하고, 그 채널에 메시지가 발행되면 자동으로 백엔드에서 수신하여 처리하도록 합니다.
마지막으로 Redis 키 TTL 만료 감지 리스너를 등록합니다.
@Bean
public RedisExpirationListener redisExpirationListener(RedisMessageListenerContainer container) {
return new RedisExpirationListener(container);
}알림은 최대 30일 캐싱하도록 구현하였기 때문에 30일이 지나면 자동으로 redis에서 키가 삭제되고 이를 감지해 DB에서 처리하도록 구현하기 위해 등록해주었습니다.
그리고 RedisExpriationListener 클래스를 생성해 만료된 키를 DB에서 처리하는 로직을 구현했습니다.
저는 30일이 지나면 자동으로 읽음 + 거절 처리를 하도록 구현했습니다.
@Slf4j
@Component
public class RedisExpirationListener extends KeyExpirationEventMessageListener {
public RedisExpirationListener(RedisMessageListenerContainer listenerContainer) { super(listenerContainer); }
@Transactional
@Override
public void onMessage(Message message, byte[] pattern) {
String expiredKey = message.toString();
log.info("Redis key expired - key: {}", expiredKey);
//...로직 구현
}
}다른 팀원이 작성한 로직도 존재하기 때문에 해당 부분은 삭제하고 간략하게 업로드하였습니다.
다음은 RedisConfig 전체 코드입니다.
@Configuration
public class RedisConfig {
@Value("${spring.data.redis.host}")
private String host;
@Value("${spring.data.redis.port}")
private int port;
@Value("${spring.data.redis.notification.channel}")
private String CHANNEL;
/* Redis 연결 객체 생성 */
@Bean
public RedisConnectionFactory redisConnectionFactory() {
return new LettuceConnectionFactory(host, port);
}
/* 공용 ObjectMapper 등록 */
@Bean
public ObjectMapper redisObjectMapper() {
return new ObjectMapper()
.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule())
.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
}
/* 공용 JSON serializer 등록 - (key,val) 저장용 */
@Bean
public Jackson2JsonRedisSerializer<PushNotificationDTO> redisJsonSerializer() {
return new Jackson2JsonRedisSerializer<>(redisObjectMapper(), PushNotificationDTO.class);
}
/* RedisTemplate 빈 생성 - (key,val) 저장용 */
@Bean
public RedisTemplate<String, PushNotificationDTO> redisTemplate() {
var t = new RedisTemplate<String, PushNotificationDTO>();
t.setConnectionFactory(redisConnectionFactory());
t.setKeySerializer(new StringRedisSerializer());
t.setValueSerializer(redisJsonSerializer());
return t;
}
/* 공용 JSON serializer 등록 - pub/sub용 */
@Bean
public Jackson2JsonRedisSerializer<PublishDTO> pubsubSerializer() {
return new Jackson2JsonRedisSerializer<>(redisObjectMapper(), PublishDTO.class);
}
/* Redis pub/sub 메시지 리스너 컨테이너 */
@Bean
public RedisTemplate<String, PublishDTO> pubSubTemplate() {
var t = new RedisTemplate<String, PublishDTO>();
t.setConnectionFactory(redisConnectionFactory());
t.setKeySerializer(new StringRedisSerializer());
t.setValueSerializer(pubsubSerializer());
t.afterPropertiesSet();
return t;
}
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisSubscriber redisSubscriber) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(redisConnectionFactory());
var ser = pubsubSerializer();
//구독 설정
container.addMessageListener((message, pattern) ->{
String ch = new String(message.getChannel(), java.nio.charset.StandardCharsets.UTF_8);
Object payload = ser.deserialize(message.getBody());
redisSubscriber.onMessage(ch, (PublishDTO) payload);
},
new PatternTopic(CHANNEL));
return container;
}
/* 이벤트 리스너 등록 */
@Bean
public RedisExpirationListener redisExpirationListener(RedisMessageListenerContainer container) {
return new RedisExpirationListener(container);
}
}
RedisPublisher
RedisPublisher에서는 알림을 저장하고 발행합니다.
@RequiredArgsConstructor
@Component
@Slf4j
public class RedisPublisher {
private final RedisTemplate<String, PublishDTO> pubsubTemplate;
private final ObjectMapper objectMapper;
private final RedisTemplate<String, PushNotificationDTO> redisTemplate;
@Value("${spring.data.redis.notification.channel}")
private String CHANNEL;
private final Long TTL = 30L;
private final String KEY_PREFIX = "pn:";
/* 메시지를 특정 채널에 발행(pnPk, userPk): userPk에 해당하는 유저에게 발행 */
public void publish(Long pnPk, Long userPk) {
PublishDTO dto = new PublishDTO(pnPk, userPk);
pubsubTemplate.convertAndSend(CHANNEL, dto);
}
/* 알림 데이터를 Redis에 저장(pnPk, notification) */
public void saveNotificationWithTTL(Long pnPk, PushNotification notification) {
redisTemplate.opsForValue().set("pn:" + pnPk, dto, 30, TimeUnit.DAYS);
}
}실제 클래스의 축약 버전입니다.
publish 메서드에서는 Pub/Sub 채널에 알림 객체를 발행해 다른 서버 인스턴스가 받도록 합니다.
saveNotificationWithTTL 메서드에서는 pn:<알림pk> 를 key로 설정해 알림 객체를 Redis에 저장합니다.
RedisSubscriber
RedisSubscriber에서는 메시지를 수신하고 SSE를 전송합니다.
@Slf4j
@RequiredArgsConstructor
@Component
public class RedisSubscriber {
private final RedisTemplate<String, PushNotificationDTO> redisTemplate;
private final ScheduledExecutorService exService = Executors.newScheduledThreadPool(10);
private final SseManager sseManager;
private final String KEY_PREFIX = "pn:";
public void onMessage(String channel, PublishDTO message) {
log.info("Received message from channel: [{}] at time: {} with message: {}", channel, Instant.now(), message.getPnPk());
//redis에서 키 조회 + sse전송
processMessage(message.getPnPk(), message.getUserPk(), 5);
}
public void processMessage(Long pnPk, Long userPk, int attempt) {
exService.submit(() -> {
try {
PushNotificationDTO pn = null;
for(int i = 0; i < attempt; i++) {
pn = (PushNotificationDTO) redisTemplate.opsForValue().get(KEY_PREFIX + String.valueOf(pnPk));
if(pn != null) {
break;
}
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
if(pn != null) {
sseManager.sendNotificationToEmitter(userPk, pn);
} else {
log.warn("No notification found in Redis for key: {}", pnPk);
}
} catch (Exception e) {
log.error("Exception while processing push notification", e);
}
});
}
}
onMessage 메서드에서 구독한 채널의 메시지를 감지해 processMessage메서드를 호출해 200ms씩 재시도하면서 Redis 값을 가져오기를 시도합니다. 가져온 후에는 SSEEmitter를 통해 사용자에게 실시간으로 전송합니다.
클라이언트는?
이제 기본적인 기능이 구현이 되었습니다.
이제 클라이언트는 /api/notification/conn 엔드포인트에 접속해 연결을 한 개 오픈합니다.
Controller
@Operation(summary = "sse 연결", description = "emitter 생성")
@GetMapping(value = "/conn", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter connectSse(@AuthenticationPrincipal JwtPrincipal jwtPrincipal) {
return notificationService.connectSse(jwtPrincipal.userPk());
}
Service
/* sse 연결 (emitter 생성) */
public SseEmitter connectSse(Long userPk) {
return sseManager.createEmitter(userPk);
}
이후 비즈니스 로직에서 알림이 발생하면 아래와 같이 NotificationService에 정의되어 있는 saveNotification메서드를 이용해 알림을 전송합니다.
notificationService.saveNotification(invitee.getUserPk(), pn);
/* 알림 DB 저장 및 publish */
@Transactional
public void saveNotification(Long userPk, PushNotification pushNotification) {
//DB 저장
notificationRepository.save(pushNotification);
//트랜잭션 성공 시 notification redis publish
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
redisPublisher.saveNotificationWithTTL(pushNotification.getPnPk(), pushNotification);
redisPublisher.publish(pushNotification.getPnPk(), userPk);
}
});
}해당 메서드에서는 일단 DB에 알림 정보를 저장한 후 트랜잭션이 성공하면 Redis에 키 저장 후 publish합니다.
잘 동작하는지 확인하는 방법
일단 SecurityConfig의 SecurityFilterChain에서 /api/notification/conn 엔드포인트에 대해 permitAll 처리를 해줍니다.
.requestMatchers("/api/notification/conn").permitAll()
이후 브라우저에서 아래처럼 접속해 하트비트가 15초마다 제대로 수신되는지 확인하면 됩니다!
이때 permitAll처리를 해주더라도 JwtPrincipal 오류가 발생할 수 있습니다. 그럴 때는 컨트롤러로 가서 /conn의 본래 파라미터인
AuthenticationPrincipal JwtPrincipal jwtPrincipal을 삭제해주고 아래처럼 1L(예시)을 넣어주면 됩니다!
@Operation(summary = "sse 연결", description = "emitter 생성")
@GetMapping(value = "/conn", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter connectSse() {
return notificationService.connectSse(1L);
}
마무리
이렇게 해서 Spring Boot SSE + Reids를 활용한 실시간 알림 기능 구현이 완료되었습니다. 백엔드 구현 자체는 오래 걸리지 않았으나 프론트에서 하루정도 시간 쏟으며 고생했던 기억이 있습니다. 다음에는 리액트로 SSE를 어떻게 처리했는지에 대해 포스팅하겠습니다:)
'자바 > 스프링부트' 카테고리의 다른 글
| [Java / Spring] Redis 이용한 대기열 시스템 구현 (0) | 2025.10.27 |
|---|---|
| [Java / Spring] Redis 이용한 좌석 선점/결제 로직 구현 (0) | 2025.10.21 |
| [JAVA / 스프링 부트] 커스텀 필터 & AuthorizationManager로 그룹별 API 접근 제어하기 (1) | 2025.10.21 |
| [JAVA / 스프링부트] UTC 저장 + KST 응답 구조로 타임존 문제 해결하기 (Spring Boot + MariaDB) (0) | 2025.10.20 |
| [JAVA / SSE] 스프링부트에서 SSE 구현하기 (SSE + Redis pub/sub) - 1 (0) | 2025.10.14 |
