SSE 적용 (2) - 이중화를 적용해보자

필요성

- 신규 구축 스테이지 대부분 API 2대 -> 스케일링 이슈 고민이 필요

- 현재 각각 API에 Subscribe, broadcast가 있으면 이벤트 발생한 쪽에 응답 누락 가능성이 있음.

 

구현 방법

- MQ를 이용

  - 기존에 사용중인 리소스

  - Redis나 Kafka를 많이 사용하나, 현재 상황에서 SSE때문에 인프라 추가는 불가한 상황

  - 다중 로그인 이렇게 심화까지 필요하지 않음 (단일 기기 로그인 환경)

  - 구현을 이것저것 보니 MQ로 충분하게 대응 가능할 것 같음

- Subscribe는 어느 서버에서 발생해도 상관이 없음, 따라서 API로 진행, Broadcast는 MQ로 적용

- 이벤트 발생 시 이벤트 ID Prefix를 확인하여 구독중인 모든 서버에 매핑되는 모든 이벤트를 동시에 Broadcast하도록 구현

- Fanout 방식으로 구현 진행

  - 해당 Exchange를 구독하고 있는 모든 큐에 broadcast를 날려 prefix 하위를 모두 broadcast하는 방식

 

구현

Batch

- 업무 Flow상 subscrie가 되어있다는 가정하에, 규정준수검사 Batch Logic 동작 중에 broadcast 수행

- broadcast 메서드 호출 시 EXCHANGE_SSE_BROADCAST으로 메시지 발행

- broadcast는 API에 구현해 놓았기 때문에 Batch에서는 Exchange 선언만 하면 됨

- 예제 코드 

//Menu1SseBroadcastStrategy

@Override
public void sseBroadcast(String cmpnId, String subId, SseExecutionStatus status) {
 
    SseBroadcastModel sseBroadcastModel = SseBroadcastModel.builder()
            .cmpnId(cmpnId)
            .menuName(getSseBroadcastMenu().getMenuName())
            .subId(subId)
            .status(status.getValue())
            .build();
 
    rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_SSE_BROADCAST, "", sseBroadcastModel);
}

 

 

API

- 실제 broadcast 메시지를 받아 처리하는 곳

- 예제 코드

//SSEConsumer

@RabbitListener(queues = RabbitMqConfig.QUEUE_SSE_BROADCAST_PROCESS, concurrency = "${consumer.sse-broadcast}")
public void sseReceiveMessage(SseBroadcastModel message) {
 
    sseService.broadcastToMappedEvent(message);
}

 

// SSEService
public void broadcastToMappedEvent(SseBroadcastModel message) {
    String idPrefix = getEventIdPrefix(message.getCmpnId(), message.getMenuName());
 
    emitters.entrySet().stream()
            .filter(entry -> entry.getKey().startsWith(idPrefix))
            .forEach(entry -> {
                SseEmitter emitter = entry.getValue();
                String eventId = entry.getKey();
                sendEventToClient(emitter, eventId, message.getMenuName(), eventId + "_" + message.getSubId()+ "_" + message.getStatus());
            });
}
 
private void sendEventToClient(SseEmitter emitter, String eventId, String menuName, String data) {
    try {
        emitter.send(
                SseEmitter.event()
                        .id(eventId)
                        .name(menuName)
                        .data(data));
    } catch (Exception e) {
        emitter.completeWithError(e);
    }
}
 
private String getEventIdPrefix(String cmpnId, String menuName) {
    StringJoiner eventIdBuilder = new StringJoiner("_");
    eventIdBuilder.add(cmpnId)
                  .add(menuName);
    return eventIdBuilder.toString();
}

 

// RabbitMQConfig

...
public static final String EXCHANGE_SSE_BROADCAST = "fanout.sse.broadcast";
public static final String QUEUE_SSE_BROADCAST_PROCESS = "sse.broadcast.process";
...
@Bean
public FanoutExchange sseBroadcastFanoutExchange() {
    return new FanoutExchange(EXCHANGE_SSE_BROADCAST);
}
 
@Bean
public Queue getSseBroadcastQueue() {
    return new Queue(QUEUE_SSE_BROADCAST_PROCESS);
}
 
@Bean
public Binding bindingSseExchangeAndQueue(Queue getSseBroadcastQueue, FanoutExchange sseBroadcastFanoutExchange) {
    return BindingBuilder.bind(getSseBroadcastQueue).to(sseBroadcastFanoutExchange);
}
...

 

Client

- API에서 broadcast가 일어나면 watch에 sseEventData response 변경이 감지되고 그때 조건에 따라 업데이트 진행

- 예제 코드

// menu1.vue
...
sseResponseData(newVal) {
    if (newVal.indexOf('_complete') !== -1) {
        this.getGridList();
    }
},
...
 
//menu1Detail.vue
...
sseResponseData(newVal) {
    if (newVal.indexOf('_complete') !== -1 && newVal.includes(this.menu1DataId)) {
        this.fetchDetail();
    }
},
...