[트러블 슈팅] 포트 분리로 인한 채팅 불가 feat Redis PubSub

이번 프로젝트를 진행하면서 채팅 + 비디오 동기화 프로젝트를 진행했었다.

여기서 채팅과 비디오 동기화 문제를 해결하기 위해서 WebSocket과 STOMP를 이용해서 클라이언트로 부터 오는 데이터를 서버쪽에서 가공 후 발행하면 구독하고 있는 주소로 데이터를 보내는 형식으로 로직을 작성했는데, 해당 방식의 문제점은 로드 밸런서를 구성한 환경에서는 8080포트 이용자와 8081포트의 이용자는 서로 메시지를 주고 받을 수 없는 환경이라는 것이다.
 
Nginx가 앞에서 모든 요청에 대해서 Round Robin 알고리즘으로 무작위로 8080포트와 8081포트로 요청을 분산시키는데 이 때 만약에 사용자1은 8080포트로 연결되고, 사용자2는 8081포트로 연결되었을 때 사용자1과 사용자2는 서로 채팅을 주고받을 수 없게 되거나 메시지 유실등의 문제를 겪게 된다.
 
이를 해결하기 위한 기술적인 해결책은 RabbitMQ와 Kafka와 같은 Message Queue와 같은 시스템을 중간에 두고 데이터를 MQ로 보냄으로써 MQ에서 발행하고 서버가 구독하는 것으로 변경하면 이 문제가 해결되는데, 프로젝트를 함에 있어서 해당 기술을 익히는데 많은 시간이 걸릴거 같아 사용 경험이 있던 Redis의 PubSub을 이용해서 처리하였다.

---

문제 상황

문제 상황은 위에 작성한 글과 같았다.
8080포트의 사용자와 8081포트의 사용자가 서로 채팅을 못한다는 것. 같은 서비스를 사용중인 유저들간의 대화가 안되는 현상이 발생한 것이다.
 

---

Redis를 메세지 브로커로써 활용하기

Redis는 인메모리 NoSQL 데이터베이스로써 빠른 성능을 제공하고 보통 백엔드 작업에서 캐싱정책에 많이 사용되는 기술이다. + 싱글 스레드 구조로 멀티 서버 환경에서 동시성 이슈를 막기 위한 해결책으로 분산락을 거는 방식으로도 사용된다.
 
현재 프로젝트에서 사용한 Redis의 PubSub환경은 클라이언트로 부터 오는 데이터를 서버에서 Redis로 발행하고 Redis가 생성한 발행 channel에 대해서 서버가 구독하는 흐름으로 연결된다. Redis를 메세제 브로커로 활용하는 것이다
 

대강적인 데이터의 흐름은 위에 사진과같다.
 
클라이언트로부터 채팅 메시지가 오게 되면 백엔드서버에서는 해당 데이터를 가지고 Redis의 channel을 생성해서 함께 보낸다. Redis는 서버로부터 건너온 channel과 데이터를 발행하고 구독하고 있는 백엔드 서버 모두에게 해당 메시지를 발행한다.
 

@Bean
@Qualifier("chatListener")
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory, @Qualifier("chatListener") MessageListener messageListener) {

  RedisMessageListenerContainer container = new RedisMessageListenerContainer();
  container.setConnectionFactory(connectionFactory);
  container.addMessageListener(messageListener, new PatternTopic("chat"));
  log.info("RedisMessageListenerContainer created");
  return container;
}

@Bean
@Qualifier("chatListener")
public MessageListener chatMessageListener(RedisChatPubSubService service) {
  log.info("ChatMessageListener created");
  return new MessageListenerAdapter(service, "onMessage");
}

Redis 설정을 해준다. 서버가 Redis로부터 메시지를 구독할 channel명(chat)과 Redis로부터 메시지를 받게 될 경우 실행 시킬 컴포넌트와 메서드명(onMessae)를 MessageListenerAdapter를 생성해주고 해당 어댑터를 MessageListenerContainer에 등록해준다.

@Slf4j
@Service
@RequiredArgsConstructor
public class RedisChatPubSubService implements MessageListener {
    private final ObjectMapper mapper;
    private final SimpMessageSendingOperations messagingTemplate;
    private final RedisTemplate<String, Object> redisTemplate;


    public void publish(String channelId, String message) {
        redisTemplate.convertAndSend(channelId, message);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        try {
            ChatMessageResponse response = mapper.readValue(body, ChatMessageResponse.class);
            messagingTemplate.convertAndSend("/sub/chat." + response.getChannelId(), response);
            log.info("Message received: " + body);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}

그럼 onMessage에서는 Redis가 발행하는 채널로부터 메시지를 받게 되면 onMessage 함수가 호출 되고 다시 클라이언트로 보내게 된다.

  @MessageMapping("/chat.{channelId}")
  public void sendMessage(@DestinationVariable Long channelId, @Payload ChatMessageRequest request) throws JsonProcessingException {
    request.allocateChannelId(channelId);
    redisChatPubSubService.publish("chat", mapper.writeValueAsString(webSocketService.sendMessage(request)));
    return response;
  }

해당 웹소켓 API 호출 시 "chat"이라는 채널에 데이터를 보낸다.
 
클라이언트 -> 서버(sendMessage 메서드를 호출하여 Redis의 channel을 생성하고 데이터를 해당 채널에 보낸다) -> Redis 발행 -> 서버 구독 -> 클라이언트의 흐름이다.
 

---

결과 및 검증

8081포트

8080포트

두 포트에서 로그를 찍어보니 Redis로부터 메시지를 구독받고 있는 모습을 볼 수 있다.
이로써 포트에 상관없이 Redis의 channel을 구독하고 있다면 모든 서버가 동일하게 메시지를 받는다