SSE(Server Sent Events) -> 알림 발송
개념
서버 전송 이벤트(Server Sent Events)는
서버에서 클라이언트로 일방향 데이터를 지속적으로 전송하며
서버에서 이벤트가 발생했을 때 실시간으로 클라이언트에게 데이터를 넘겨주는
서버 → 클라이언트로의 실시간 단방향 통신 방법
연결 및 통신 방식
- 단방향 통신 : 서버 -> 클라이언트 로 데이터 전송
- HTTP 기반 : HTTP/1.1 기반으로 클라이언트가 서버에 연결을 요청하면 지속적으로 데이터 전송
- 자동 재연결 : 연결이 끊어지기 전에 연결 유지 위해 자동으로 재연결 시도
장점 / 단점
- 장점
- 설정이 매우 간단해 손쉽게 사용 가능
- 자동 재연결 및 이벤트 ID를 통해 간편하게 상태 복구 가능
- 단점
- 클라이언트에서 서버로의 데이터 전송이 불가능 (단방향)
- 연결 수에 제한이 있을 수 있다 (Spring의 커넥션 풀 최대 사용 시 서버 멈춤)
왜 SSE 사용해야할까?
Long Polling 방식은 서버에서 클라이언트에게 줄 데이터가 없는 경우 곧바로 응답하지 않고 커넥션을 계속 유지하며, 서버에서 줄 데이터가 생겼을 때만 응답을 보내는 방식이다.
(연결이 끊어져있다가 서버에 이벤트가 발생하면 연결해서 데이터를 응답한다.)
Long Polling도 서버와 클라이언트가 연결을 유지한다는 점에서 SSE와 같아 보이겠지만, 서버에서 이벤트가 발생해 데이터를 응답하면 연결이 끊어지고 다시 연결해야 한다.
BUT!! SSE는 서버에서 데이터를 응답하지 않아도 일정 시간 동안은 연결을 끊지 않고 계속 유지한다.
그래서 서버에서 이벤트가 자주 일어날 때 사용하기 좋다. (소셜 미디어 알람, 주식 가격 등 ..)
구현
SSE 사용이유
WebSocket은 양방향 통신에 적합한 반면 알림은 단방향 전달 중심의 구조기 때문에
과도한 리소스를 사용하는 WebSocket은 적합하지 않다.
또한 Long Polling은 서버에서 이벤트가 발생해 데이터를 응답하면 연결이 끊어지고 다시 연결해야 하기에 알림발송(이벤트가 자주 일어남)에 적합하지 않다.단방향 전송에 최적화된 SSE 방식을 채택하여 더 적은 리소스로
안정적인 실시간 알림 기능을 구현
< Controller >
클라이언트가 SSE 연결 요청한것을 처리합니다
로그인 사람만 SSE 연결을 할 수 있도록 로그인 세션 정보를 가져옵니다.
Last-Event-ID는 SSE 연결이 끊어졌을 경우 클라이언트가 수신한 마지막 ID값을 의미합니다. (알림 재전송을 위해 필요)
- 클라이언트가 GET 요청으로 SSE 연결을 요청한다.
- 세션에서 로그인한 사용자 정보(memberId)와 lastEventId를 서비스단으로 넘겨준다. (SSE 연결 위해)
- Last-Event-ID는 SSE 연결이 끊어졌을 경우 클라이언트가 수신한 마지막 ID값으로 놓친 알림을 재전송하기 위함이다.
- reservationAlarmSseService.connect(memberId, lastEventId)가 실행된다.
< 예시 >
@RequestMapping("/alarms")
@RestController
@RequiredArgsConstructor
public class ReservationAlarmSseController {
private final ReservationAlarmSseService reservationAlarmSseService;
@GetMapping("/reservations/alarms")
public SseEmitter subscribe(
HttpSession session,
@RequestHeader(value = "Last-Event-ID", required = false) String lastEventId
) {
// 세션에서 로그인한 사용자 정보 가져오기
MemberLoginResponse userDetails = (MemberLoginResponse) session.getAttribute("memberId");
if (userDetails == null) {
throw new IllegalStateException("로그인 정보가 없습니다.");
}
Long memberId = userDetails.getMemberId();
// lastEventId를 로그 또는 서비스로 넘겨서 놓친 알림 재전송할 수 있음
return reservationAlarmSseService.connect(memberId, lastEventId);
}
}
< Service >
서비스단에서 클라이언트와 SSE가 연결됩니다.
클라이언트와 SSE를 연결하기 위한 맵이 필요합니다.
Map을 통해서 사용자 기준으로 각각의 SSE 연결을 관리합니다.클라이언트와 SSE 연결 유지 시간을 설정합니다.
클라이언트와 SSE 연결 생성 및 관리를 하는 메서드를 구현합니다.
특정 회원에게 실시간으로 알림 메시지를 전송하는 메서드를 구현합니다.
클라이언트가 SSE와 연결되어 있는지 확인하는 메서드를 구현합니다.
-
클라이언트와 SSE 연결을 저장할 Map과 연결 시간을 설정한다.
- Map<Long, SseEmitter>를 통해서 사용자(memberId) 기준으로 각각의 SSE 연결을 관리한다. (누가 접속 중인지를 서버 메모리에 기억) -
연결을 위한 connect 메서드를 구현한다.
- 클라이언트와 SSE 연결 요청이 들어오면 SseEmitter 객체를 생성한다.
(TIMEOUT만큼 유효하게 설정) - 서버는 생성된 SseEmitter 객체를 통해 클라이언트와의 SSE 연결을 맺는다.
- 해당 SseEmitter 객체는 emitterMap에 memberId 키로 저장한다. - emitter.onTimeout() / emitter.onCompletion() /emitter.onError() ->
연결이 끊어지거나 에러 시 emitterMap 에서 지워서 메모리 누수를 방지한다.
- 클라이언트와 SSE 연결 요청이 들어오면 SseEmitter 객체를 생성한다.
-
알림을 특정 사용자에게 보내는 sendMessage 메서드를 구현한다.
- 특정 memberId로 등록된 SseEmitter를 찾아서 이벤트(alarm)와 데이터를 전송한다.
- emitterMap 에 해당 사용자의 emitter 가 있으면 .send() 로 전송한다.
- 없으면(연결 끊긴 상태) "SSE 연결 없음" 로그를 찍는다.
- 보내다가 오류 나면 emitterMap 에서 제거해서 다음 알림 때 불필요하게 시도하지 않도록 한다.
- 특정 memberId로 등록된 SseEmitter를 찾아서 이벤트(alarm)와 데이터를 전송한다.
-
사용자가 현재 SSE와 연결이 되고 있나 확인하는 메서드를 구현한다.
- emitterMap 에 있으면 true / 없으면 false
< 예시 >
@Slf4j
@Service
@Transactional
@RequiredArgsConstructor
public class ReservationAlarmSseServiceImpl implements ReservationAlarmSseService {
private static final Long TIMEOUT = 60L * 1000 * 60; // 60분
// 클라이언트와의 SSE 연결을 관리하기 위한 맵
private final Map<Long, SseEmitter> emitterMap = new ConcurrentHashMap<>();
// 클라이언트가 SSE 연결을 요청할 때 호출되는 메서드
@Override
public SseEmitter connect(Long memberId, String lastEventId) {
log.info("SSE 연결 요청: memberId={}, 현재 연결 수={}", memberId, emitterMap.size());
SseEmitter emitter = new SseEmitter(TIMEOUT);
emitterMap.put(memberId, emitter);
emitter.onTimeout(() -> {
log.info("SSE 연결 타임아웃: memberId={}", memberId);
emitterMap.remove(memberId);
});
emitter.onCompletion(() -> {
log.info("SSE 연결 종료: memberId={}", memberId);
emitterMap.remove(memberId);
});
emitter.onError(e -> {
log.warn("SSE 오류 발생: memberId={}, error={}", memberId, e.getMessage());
emitterMap.remove(memberId);
});
try {
emitter.send(SseEmitter.event().name("connect").data("SSE SUCCESS - memberId: " + memberId));
} catch (IOException e) {
log.error("초기 연결 메시지 전송 실패", e);
}
log.info("연결 완료: memberId={}, 현재 연결 수={}", memberId, emitterMap.size());
return emitter;
}
// 클라이언트에게 메시지를 전송하는 메서드
@Override
public void sendMessage(Long memberId, Object data) {
SseEmitter emitter = emitterMap.get(memberId);
if (emitter != null) {
try {
log.info("알림 전송 시도: memberId={}, data={}", memberId, data);
emitter.send(SseEmitter.event().name("alarm").data(data));
} catch (IOException e) {
log.warn("SSE 메시지 전송 실패: memberId={}, error={}", memberId, e.getMessage());
emitterMap.remove(memberId);
}
} else {
log.info("SSE 연결 없음: memberId={}", memberId);
}
}
// 클라이언트가 연결되어 있는지 확인하는 메서드
@Override
public boolean hasConnected(Long memberId) {
return emitterMap.containsKey(memberId);
}
}(+Service 인터페이스)
public interface ReservationAlarmSseService {
SseEmitter connect(Long memberId, String lastEventId);
void sendMessage(Long memberId, Object data);
boolean hasConnected(Long memberId);
}< Mapper >
- 상태가 CONFIRMED 이면서 24시간 지난 예약 조회 매퍼
- 상태가 CONFIRMED 이면서 30분 이내 end_time 도래 예정인 예약 조회 매퍼
- 5분 이내 CONFIRMED 상태로 바뀐 예약 조회 매퍼
- 5분 이내 CANCELLED 상태로 바뀐 예약 조회 매퍼
- Member는 도메인에 있는걸 가져와서 memberId와 nickname을 사용
(+ 여기에는 end_time(만료시간)과 state(상태) 조건을 비교해서 상황별로 알림을 보내줌)
<!--알림-->
<resultMap id="ReservationAlarm" type="com.airbng.dto.reservation.ReservationResponse">
<result property="reservationId" column="reservation_id"/>
<result property="startTime" column="start_time"/>
<result property="endTime" column="end_time"/>
<result property="state" column="state"/>
<association property="dropper" javaType="com.airbng.domain.Member">
<result property="memberId" column="dropper_id"/>
<result property="nickname" column="dropper_nickname"/>
</association>
<association property="keeper" javaType="com.airbng.domain.Member">
<result property="memberId" column="keeper_id"/>
<result property="nickname" column="keeper_nickname"/>
</association>
</resultMap>
<!-- end_time 24시간 지난 상태 CONFIRMED 예약 조회 -->
<select id="findExpiredConfirmedReservations" parameterType="java.time.LocalDateTime" resultMap="ReservationAlarm">
SELECT
r.reservation_id,
r.state,
r.start_time,
r.end_time,
d.member_id AS dropper_id,
d.email AS dropper_email,
d.name AS dropper_name,
d.phone AS dropper_phone,
d.nickname AS dropper_nickname,
d.password AS dropper_password,
d.status AS dropper_status,
k.member_id AS keeper_id,
k.email AS keeper_email,
k.name AS keeper_name,
k.phone AS keeper_phone,
k.nickname AS keeper_nickname,
k.password AS keeper_password,
k.status AS keeper_status
FROM reservation r
JOIN member d ON r.dropper_id = d.member_id
JOIN member k ON r.keeper_id = k.member_id
WHERE r.state = 'CONFIRMED'
AND r.end_time < #{deadline}
</select>
<!-- 30분 이내 end_time 도래 예정인 예약 (CONFIRMED) -->
<select id="findConfirmedNearEndTime" parameterType="java.time.LocalDateTime" resultMap="ReservationAlarm">
SELECT
r.reservation_id,
r.state,
r.start_time,
r.end_time,
d.member_id AS dropper_id,
d.email AS dropper_email,
d.name AS dropper_name,
d.phone AS dropper_phone,
d.nickname AS dropper_nickname,
d.password AS dropper_password,
d.status AS dropper_status,
k.member_id AS keeper_id,
k.email AS keeper_email,
k.name AS keeper_name,
k.phone AS keeper_phone,
k.nickname AS keeper_nickname,
k.password AS keeper_password,
k.status AS keeper_status
FROM reservation r
JOIN member d ON r.dropper_id = d.member_id
JOIN member k ON r.keeper_id = k.member_id
WHERE r.state = 'CONFIRMED'
AND r.end_time BETWEEN #{now} AND DATE_ADD(#{now}, INTERVAL 30 MINUTE)
</select>
<!-- CONFIRMED 상태로 바뀐 예약 -->
<select id="findStateChangedToConfirmed" resultMap="ReservationAlarm">
SELECT
r.reservation_id,
r.state,
r.start_time,
r.end_time,
d.member_id AS dropper_id,
d.email AS dropper_email,
d.name AS dropper_name,
d.phone AS dropper_phone,
d.nickname AS dropper_nickname,
d.password AS dropper_password,
d.status AS dropper_status,
k.member_id AS keeper_id,
k.email AS keeper_email,
k.name AS keeper_name,
k.phone AS keeper_phone,
k.nickname AS keeper_nickname,
k.password AS keeper_password,
k.status AS keeper_status
FROM reservation r
JOIN member d ON r.dropper_id = d.member_id
JOIN member k ON r.keeper_id = k.member_id
WHERE r.state = 'CONFIRMED'
</select>
<!-- CANCELLED 상태로 바뀐 예약 -->
<select id="findStateChangedToCancelled" resultMap="ReservationAlarm">
SELECT
r.reservation_id,
r.state,
r.start_time,
r.end_time,
d.member_id AS dropper_id,
d.email AS dropper_email,
d.name AS dropper_name,
d.phone AS dropper_phone,
d.nickname AS dropper_nickname,
d.password AS dropper_password,
d.status AS dropper_status,
k.member_id AS keeper_id,
k.email AS keeper_email,
k.name AS keeper_name,
k.phone AS keeper_phone,
k.nickname AS keeper_nickname,
k.password AS keeper_password,
k.status AS keeper_status
FROM reservation r
JOIN member d ON r.dropper_id = d.member_id
JOIN member k ON r.keeper_id = k.member_id
WHERE r.state = 'CANCELLED'
</select>( + Mapper 인터페이스 )
// 24시간이 지난 CONFIRMED 예약 조회 (EXPIRED 알림용)
List<ReservationResponse> findExpiredConfirmedReservations(@Param("deadline") LocalDateTime deadline);
// endTime이 30분 이내인 예약 조회 (REMINDER 알림용)
List<ReservationResponse> findConfirmedNearEndTime(@Param("now") LocalDateTime now);
// CONFIRMED로 변경된 예약 (STATE_CHANGE 알림용)
List<ReservationResponse> findStateChangedToConfirmed();
// CANCELLED로 변경된 예약 (CANCEL_NOTICE 알림용)
List<ReservationResponse> findStateChangedToCancelled();< AlertScheduler >
예약 상태 변화나 시간 조건에 따라 사용자(DROPPER, KEEPER)에게 SSE 알림을 보내줍니다.
설정한 시간만큼 예약 상태를 DB에서 조회해서 SSE로 알림을 보낸다.
-
스케줄러의 메인 메서드로 1분마다 다음을 반복해서 실행한다.
- EXPIRED 알림 (24시간 지난 예약)
- REMINDER 알림 (30분 전)
- STATE_CHANGE 알림 (확정된 예약)
- CANCEL_NOTICE 알림 (취소된 예약)
-
알림 대상이 되는 예약은 send 메서드를 통해서 SSE로 전송한다.
-
sendToBoth() : DROPPER / KEEPER 둘 다 보내야 하는 상황
-
sendToOne() : DROPPER에게만 보내는 상황
@Slf4j
@Async
@Component
@RequiredArgsConstructor
public class AlertScheduledTask {
private final ReservationMapper reservationMapper;
private final ReservationAlarmSseService sseService;
private final ReservationAlarmCacheService reservationAlarmCacheService;
@Scheduled(initialDelay = 10000, fixedRate = 1000 * 60) //대기시간 10초, 1분 주기로 스케줄러 실행
public void processReservationAlarms() {
log.info("스케줄러 실행 - 현재 시간: {}", LocalDateTime.now());
LocalDateTime now = LocalDateTime.now();
// 1. EXPIRED 알림 (24시간 지난 CONFIRMED)
List<ReservationResponse> expired = reservationMapper.findExpiredConfirmedReservations(now.minusHours(24));
for (ReservationResponse r : expired) {
sendToBoth(r, NotificationType.EXPIRED, "짐 보관이 아직 완료되지 않았어요.", "고객 짐 보관 상태가 아직 완료되지 않았습니다.");
}
// 2. REMINDER 알림 (30분 전)
List<ReservationResponse> remind = reservationMapper.findConfirmedNearEndTime(now.plusMinutes(30));
for (ReservationResponse r : remind) {
sendToOne(r.getDropper().getMemberId(), r.getReservationId(), r.getDropper().getNickname(), "DROPPER", NotificationType.REMINDER, "곧 짐을 찾아가셔야 해요.");
}
// 3. STATE_CHANGE 알림
List<ReservationResponse> confirmed = reservationMapper.findStateChangedToConfirmed();
for (ReservationResponse r : confirmed) {
sendToOne(r.getDropper().getMemberId(), r.getReservationId(), r.getDropper().getNickname(),"DROPPER", NotificationType.STATE_CHANGE, "예약이 확정되었습니다.");
}
// 4. CANCEL_NOTICE 알림
List<ReservationResponse> cancelled = reservationMapper.findStateChangedToCancelled();
for (ReservationResponse r : cancelled) {
sendToOne(r.getDropper().getMemberId(), r.getReservationId(), r.getDropper().getNickname(),"DROPPER", NotificationType.CANCEL_NOTICE, "예약이 취소되었습니다.");
}
}
private void sendToBoth(ReservationResponse r, NotificationType type, String dropperMsg, String keeperMsg) {
LocalDateTime now = LocalDateTime.now();
// DROPPER
if (sseService.hasConnected(r.getDropper().getMemberId())) {
AlarmRespose d = AlarmRespose.builder()
.reservationId(r.getReservationId())
.receiverId(r.getDropper().getMemberId())
.nickName(r.getDropper().getNickname())
.role("DROPPER")
.type(type)
.message(dropperMsg)
.sendTime(now.toString()).build();
sseService.sendMessage(r.getDropper().getMemberId(), d);
}
// KEEPER
if (sseService.hasConnected(r.getKeeper().getMemberId())) {
AlarmRespose k = AlarmRespose.builder()
.reservationId(r.getReservationId())
.receiverId(r.getKeeper().getMemberId())
.nickName(r.getKeeper().getNickname())
.role("KEEPER")
.type(type)
.message(keeperMsg)
.sendTime(now.toString()).build();
sseService.sendMessage(r.getKeeper().getMemberId(), k);
}
private void sendToOne(Long id, Long resId, String name, String role, NotificationType type, String message) {
if (!sseService.hasConnected(id)) return;
AlarmRespose dto = AlarmRespose.builder()
.reservationId(resId)
.receiverId(id)
.nickName(name)
.role(role)
.type(type)
.message(message)
.sendTime(String.valueOf(LocalDateTime.now()))
.build();
sseService.sendMessage(id, dto);
}
}< WebAppInitializer >
비동기 처리 방식을 사용하기 위해서 DispatcherServlet 레벨에서 비동기 지원을 설정
필터가 비동기 서블릿 요청을 올바르게 통과시킬 수 있도록 해줌
비동기 요청/응답도 필터 체인을 정상 통과하도록 설정
-
servletContext 에 필터를 등록하는 코드를 추가한다.
-
DispatcherServlet 자체에 대해 비동기 서블릿(Async Servlet) 기능을 허용하도록 한다. (컨트롤러가 async 리턴 타입을 처리할 수 있게 해줌)
-> dispatcher.setAsyncSupported(true); -
비동기 서블릿을 쓸 때 필터가 함께 작동할 수 있도록 허용해주는 설정
( SSE(Server-Sent Events) 나 @Async 컨트롤러가 있으면, 이 옵션을 켜줘야 필터가 비동기 처리까지 관여할 수 있음)-
SSE나 비동기 컨트롤러 사용을 위해 해준다.
(필터가 비동기 서블릿 요청을 올바르게 통과시킬 수 있도록 해줌)
-> cachingFilter.setAsyncSupported(true);<예시>
-
dispatcher.setAsyncSupported(true); // async 지원 활성화
// 필터 등록 및 async 지원 활성화
FilterRegistration.Dynamic cachingFilter = servletContext.addFilter("cachingRequestFilter", new CachingRequestFilter());
cachingFilter.addMappingForUrlPatterns(null, false, "/*");
cachingFilter.setAsyncSupported(true); // async 지원 활성화
System.out.println("CachingRequestFilter async supported enabled.");
System.out.println("WebAppInitializer completed.");+< DTO >
- AlarmRespose
- 알림 발송 타입을 enum으로 뺀 NotificationType 사용
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class AlarmRespose {
private Long reservationId;
private Long receiverId; // 실제 알림 받는 유저 ID
private String nickName; // 실제 알림 받는 유저 이름
private String role; // "KEEPER" 또는 "DROPPER"
private NotificationType type; // "EXPIRED", "STATE_CHANGE" 등
private String message; // 사용자에게 보여줄 메시지
private String sendTime; // 발송 시각
}
- ReservationResponse
- dropper / keeper는 Member 도메인을 사용한다. (타입 : Member)
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ReservationResponse {
private Long reservationId; // 예약 ID
private Member dropper;
private Member keeper;
private String startTime;
private String endTime;
private String state;
}
< enum >
- 알림 상태를 enum으로 따로 빼서 설정
public enum NotificationType {
EXPIRED, //CONFIRMED인데 endTime 24시간 지남
STATE_CHANGE, //예약 상태가 바뀜 (PENDING -> CONFIRMED / CONFIRMED -> CANCELLED )
REMINDER, //시간 다가올 때 리마인드
CANCEL_NOTICE // 예약 취소 알림
}문제점
- 알림 중복 방지가 안되는 문제가 생긴다. 이를 해결하기 위해 레디스를 사용하여 알림을 보낸 건에 대해서 해당 캐시를 저장하는 방식을 사용하였다.
레디스를 사용하여 알림 중복 방지는 아래 글을 참고
-> 레디스 적용
KIM