실시간 푸쉬 기능 구현

비교

sse comparison

WebSocket

실시간 양방향 구조라는 장점을 가지지만, 클라이언트가 서버에서 push받으면 되는구조에서는, 불필요하다고 판단.

Pooling

주기적으로 새로고침해서, 전체 페이지를 다시 그려내면 되지만, 서버의 부하가 심하다고 판단.

Server-sent event

Spring이 이미 가지고있는 라이브러리이며, eventstream이 연결되면, 서버-> 클라이언트 단방향 통신이 형성되어, 필요한 정보만 간략하게 받을 수 있음.

구현

Client Side

알림을 받으려는 페이지에 javaScript만 작성 javaScript로 구현하며, EventSource를 이용

const evtSource = new EventSource("/connect");
evtSource.onopen = function(event) {
console.log("서버 연결 완료, EventStream 생성")
}
evtSource.addEventListener("sse", function(event){
console.log(event.data);
alert(event.data);
})
evtSource.onerror = function() {
console.log("EventStream 연결 애러")
evtSource.removeEventListener("sse", function(event){
    console.log("EventStream 연결 해제");
});
evtSource.close();
}

Server Side

Controller-Service-Repository로 구성

@RequiredArgsConstructor
@Repository
public class StoreSseRepository {
private final Map<Integer, SseEmitter> emitters = new ConcurrentHashMap<>();

public void save(int storeId, SseEmitter emitter) {
    emitters.put(storeId, emitter);
}

public Optional<SseEmitter> findById(int storeId) {
    return Optional.ofNullable(emitters.get(storeId));
}

public void deleteById(int storeId) {
    emitters.remove(storeId);
}
}

HashMap도 멀티스레딩을 지원하기는 하지만, @Synchronized를 지원하지 않음. 매번 Lock이 걸리므로, 병목현상도 발생
기본적인 HashMap구조이면서 Thread-safe한 ConcurrentHashMap을 사용하도록 함.
HashMap구조라, <매장PK, Emitter>로 엮이어 있다.
매장별로 오직 하나의 Emitter를 가진다.
findById를 Optional로 받아서, 서버가 오프라인인 경우, emitter가 없을 수 있는 상황을 대비한다.
delete의 경우, emitter가 timeout되었거나, 모든 데이터가 전송된 경우, emitter를 삭제하도록 한다.(TODO: 이경우 재고 필요)

@RequiredArgsConstructor
@Service
public class StoreSseService {
    private final StoreSseRepository storeSSERepository;
    private static final long TIMEOUT = 60 * 1000;
    private static final long RECONNECTION_TIMEOUT = 1000L;
    // OnCompletion: Emitter 가 완료될 때(모든 데이터가 성공적으로 전송된 상태)
    // OnTimeout: Emitter 가 타임아웃 되었을 때(지정된 시간동안 어떠한 이벤트도 전송되지 않았을 때)

    public SseEmitter createConnection(int storeId) {
        SseEmitter emitter = new SseEmitter(TIMEOUT);
        storeSSERepository.save(storeId, emitter);
        createEvent(storeId, "실시간 Push 서비스 연결 완료"); // EventStream 생성
        return emitter;
    }

    public void createOrderNotification(int orderId, int storeId) {
        if (storeSSERepository.findById(storeId).isPresent()) {
            createEvent(storeId, orderId + "번 주문 알림. 새로고침을 눌러 주문을 확인해주세요.");
        }
    }

    public void createEvent(int storeId, String data) {
        Optional<SseEmitter> opEmitter = storeSSERepository.findById(storeId);
        if (opEmitter.isPresent()) {
            SseEmitter emitter = opEmitter.get();
            SseEmitter.SseEventBuilder event = SseEmitter.event()
                    .name("sse")
                    .data(data)
                    .reconnectTime(RECONNECTION_TIMEOUT);
            try {
                emitter.send(event);
                emitter.onTimeout(() -> storeSSERepository.deleteById(storeId));
                emitter.onCompletion(() -> storeSSERepository.deleteById(storeId));
            } catch (Exception e) {
                storeSSERepository.deleteById(storeId);
                emitter.completeWithError(e);
            }
        }
    }
}

eventStream이 연결되면, createConnection메서드가 매장PK를 가져와서, emitter 생성
생성이 되면, createEvent메서드가, event를 생성후, 해당 매장pk의 emitter에 send()한다.
emitter는 event가 감지되면, emitter에 연결된 매장에 수신된 event를 전송한다.

@RequiredArgsConstructor
@RestController
public class StoreSseController {
    private final HttpSession session;
    private final StoreSseService storeSSEService;

    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect() {
        SessionStore sessionStore = (SessionStore) session.getAttribute("sessionStore");
        if (sessionStore == null) {
            return null;
        } else {
            return storeSSEService.createConnection(sessionStore.getId());
        }
    }
}

web 클라이언트가 로그인되면 eventStream이 연결될 controller가 하나 필요하다.

클라이언트쪽 로직이 모두 구현되었으면, 이제, 주문하기 로직에서 해당 sse Service를 건드려주면 된다.

@PostMapping("/api/users/{userId}/orders") // 주문하기
public ResponseEntity<?> order(@PathVariable String userId, @RequestBody UserRequest.OrderDTO reqDTO) {
    UserResponse.OrderDTO respDTO = userService.createOrder(reqDTO);
    storeSseService.createOrderNotification(respDTO.getId(), respDTO.getStoreId()); // 이 부분

    return ResponseEntity.ok(new ApiUtil<>(respDTO));
}

9oj0e / pathorder_server

feat : 실시간 주문 push #87