[Spring] 웹소켓/Stomp 8

다중 서버에서의 Redis Pub & Sub 

 

다중 서버 상황에서 서버 메모리에 의존적인 기능들은 대부분 문제가 생길 가능성이 크다.

 

문제

A 서버와 B 서버가 있을 때 클라이언트1과 2가 실시간 채팅하는 상황에서 1은 A 서버에 발행, 2는 B 서버를 구독하게 되면

A 서버는 클라이언트 2에 대한 정보를 알고 있지 않기 때문에 실시간 채팅 기능이 정상적으로 동작하지 않는다.

 

 

해결 - 서드 파티 솔루션으로 Redis Pub Sub 기능 사용

주로 Redis 또는 Kafka를 사용

Redis는 pub sub 기능 외에도 캐싱, 리프레시 토큰 관리 등 여러 역할을 많이 수행한다.

 

Redis Pub Sub : Redis가 제공하는 메시징 기능 

- 실시간 통신에서 메시지를 전파해주는 목적으로 사용한다.

- Redis Pub Sub에서는 구독, 발행하는 주체는 클라이언트가 아닌 서버가 된다.

 

user a는 Server A를 구독 & user b는 Server B를 구독

Server A, B는 모두 Redis를 Subscribe

 

 

1. user a가 Server A에게 publish

2. Server A는 메시지를 받아 Redis에게 Publish

3. Redis는 메시지를 받아 Redis를 구독 중인 Server A, B에게 전달

4. Server A, B는 해당 메시지를 받고, Server B를 구독 중인 user b는 Redis에서 받아온 메시지를 실시간으로 확인하게 된다.

5. user a는 본인이 발행한 메시지를 Server A를 구독하고 있기 때문에 다시 전달받는다.

 

Redis Pub & Sub 기능을 사용하기 위해서는 스프링 서버와 Redis 서버를 연결하는 객체를 만들어야 한다.

- Redis에게 publish하는 역할의 객체

- Redis에게 subscribe하는 역할의 객체

 

@Configuration
public class RedisConfig {
    @Value("${spring.data.redis.host}")
    private String host;

    @Value("${spring.data.redis.port}")
    private int port;

    @Bean // 연결 기본 객체
    @Qualifier("chatPubSub") // @Qualifier("db1") Qualifier : 같은 타입 bean이 여러 개 있을 때 어느 것을 주입할지 지정하는 식별자
    public RedisConnectionFactory chatPubSubFactory() {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(host);
        configuration.setPort(port);
        // configuration.setDatabase(0); Redis pub sub 기능은 특정 데이터베이스에 의존적이지 않는다.

        return new LettuceConnectionFactory(configuration);
    }

    /* @Qualifier("chatPubSub") : 연결 객체가 여러 개일 경우 의미가 있는 어노테이션, 현재 상황에서는 큰 의미가 없다.
     "chatPubSub"라는 RedisConnectionFactory를 사용하는 StringRedisTemplate Bean 만들기
     일반적으로 StringRedisTemplate가 아니라 RedisTemplate<키 데이터 타입, 값 데이터 타입>을 쓴다. (캐싱 상황)
     StringRedisTemplate : String 형태로 publish
     */
    @Bean // publish 객체
    @Qualifier("chatPubSub")
    public StringRedisTemplate redisTemplate(@Qualifier("chatPubSub") RedisConnectionFactory factory) {
        return new StringRedisTemplate(factory);
    }

    @Bean // subscribe 객체
    public RedisMessageListenerContainer container(@Qualifier("chatPubSub") RedisConnectionFactory factory,
                                                   MessageListenerAdapter messageListenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(messageListenerAdapter, new PatternTopic("chat")); // topic : chat에 대해서만 pub, sub
        return container;
    }

    @Bean //redis 에서 수신된 메시지를 처리하는 객체 생성
    public MessageListenerAdapter messageListenerAdapter(RedisPubSubService service) {
        // RedisPubSubService의 특정 메서드가 수신된 메시지를 처리할 수 있도록 지정
        // RedisPubSubService 속 onMessage 메서드가 동작
        return new MessageListenerAdapter(service, "onMessage");

    }

}

 

@Service
@RequiredArgsConstructor
public class RedisPubSubService implements MessageListener {

    private final StringRedisTemplate template;
    private final SimpMessageSendingOperations messageTemplate;

    // onMessage 메서드 구현, subscribe 시 동작
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String payload = new String(message.getBody());
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            ChatMessageDto chatMessageDto = objectMapper.readValue(payload, ChatMessageDto.class);
            messageTemplate.convertAndSend("/topic/" + chatMessageDto.getRoomId(), chatMessageDto);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }

        // message : redis를 subscribe할 때 받아오는 실제 메시지
        // pattern : topic 이름의 패턴이 들어가있고, 현재는 사용하지 않지만 여러 동적 기능 구현 가능
    }

    public void publish(String channel, String message) {
        messageTemplate.convertAndSend(channel, message);
    }
}

 

@RequiredArgsConstructor
@Controller
@Slf4j
public class StompController {

    // 1. messageMapping(수신), sendTo(송신) 한 번에 처리 : 유연성이 떨어진다.
//    @MessageMapping("/{roomId}") // 클라이언트에서 특정 publish/roomId 형태로 메시지 발행 시 해당 요청을 수신
//    @SendTo("/topic/{roomId}") // 해당 roomId에 메시지를 발행하여 구독 중인 클라이언트에게 메시지 전송
//    // @DestinationVariable : @MessageMapping과 함께 사용, @PathVariable의 Stomp version
//    public String sendMessage(@DestinationVariable Long roomId, String message) {
//        log.info("{}", message);
//        return message;
//    }

    private final SimpMessageSendingOperations messageTemplate;
    private final ChatService chatService;
    private final RedisPubSubService redisPubSubService;

    // 2. messageMapping만 활용 (sendTo 사용 x) : 유연성 증가
    @MessageMapping("/{roomId}") // 클라이언트에서 특정 publish/roomId 형태로 메시지 발행 시 해당 요청을 수신
    public void sendMessage(@DestinationVariable Long roomId, ChatMessageDto chatMessageDto) throws JsonProcessingException { // modelAttribute가 아니라 @Payload로 객체 등 값을 컨트롤러에서 받는다.

        log.info("message : {}, sender : {}",
                chatMessageDto.getMessage(), chatMessageDto.getSenderEmail());

        chatService.saveMessage(roomId, chatMessageDto); // 메시지를 DB에 저장

        // messageTemplate.convertAndSend("/topic/" + roomId, chatMessageDto);

        // redis pub sub 으로 동작, redisPubSubService 에 해당 코드 작성
        // 서버로 pub sub 을 하기 위해서는 아래 코드를 주석 처리, 위 코드 주석을 해제하면 된다.

        chatMessageDto.setRoomId(roomId);

        ObjectMapper objectMapper = new ObjectMapper();
        String message = objectMapper.writeValueAsString(chatMessageDto);
        redisPubSubService.publish("chat", message);
    }
}

 

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ChatMessageDto {
    // 요청, 응답 모두 사용되는 dto
    private Long roomId; // roomId 추가
    private String message;
    private String senderEmail;
}

 

Redis Config

- Lettuce 기반의 ConnectionFactory 객체

- String 메시지 발행을 위한 StringRedisTemplate 객체

- RedisMessageListenerContainer 객체 생성 => Redis 특정 채널 구독

- MessageListenerAdapter => 수신된 메시지 처리 (PubSubService의 onMessage)

 

 

메시지 송수신 

- StompController 를 통해 send된 message를 redis로 publish

- redis의 "chat" 채널로 메시지 발행 시 해당 채널을 구독 중인 RedisMessageListenerContainer가 메시지를 수신

- 수신된 메시지는 MessageListenerAdapter를 통해 onMessage 메서드로 전달

- onMessage 메서드는 수신된 메시지를 각 서버의 topic에 메시지를 발행

- 해당 topic을 subscribe 중인 클라이언트에게 채팅 메시지 전달

 

Redis pub sub 대신 Kafka를 사용할 수도 있다.

Redis : 빠른 성능(메모리, ram에 저장), pub sub 과정에서 메시지를 저장하지 않기 때문에 listen하는 서버가 없을 경우 메시지 유실

Kafka : 디스크(ssd, hdd)에 메시지 저장, 안정적인 메시징 처리 (메시지를 저장)