SpringBoot + STOMP 웹소켓 고도화

문제점 식별

웹소켓 STOMP을 통해 pub/sub 방식으로 메시지를 발행하고 JPA을 접목하여 메시지를 저장할 수 있었지만 큰 문제점이 있었다.
채팅방을 만들고 채팅을 보낼 때 유저가 현재 웹사이트에 들어오지 않았다면 유저에게 메세지를 읽지 않았다는 알림을 보내고 싶었다.

처음 생각은 이러했다.

1. 세션에다가 정보를 담으면 되지 않을까? → 진행하는 프로젝트에서는 세션을 사용하지 않고 jwt을 통해 인증을 한다 또한 STOMP 방식은 기존 웹소켓 방식은 단순하고 편리하게 사용하는 게 목적이기에 세션을 사용하는 건 좋은 방법이 아니라고 생각했다.
2. 유저가 웹소켓이 끊겼다는 정보와 유저가 무슨 채팅방들을 가지고 있는지 DB에 저장하는 방식은? → 가능은 하겠지만 너무 많은 리소스를 차지한다 좋지 않은 방법이라고 생각하였다.

해결방법

팀원들과 많은 회의를 통해 좋은 방법을 갈구해보았다.
우리가 선택한 방법은 Redis에서 지원하는 pub/sub 방식이었다. 이 방법을 사용한다면 유저의 정보와 현재 채팅방에 있는지 존재유무를 판단할 수 있었다.
결론적으로 STOMP pub/sub 방식과 Redis pub/sub 방식을 혼용하여 사용하였다.

의존성 주입

    // redis를 JPA Repository를 이용하듯 인터페이스를 제공하는 모듈
    // 스프링에서 제공하는 Redis
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'

    // 로컬/개발 환경에서 테스트
    implementation 'it.ozimov:embedded-redis:0.7.2'

Redis Config 설정

@Configuration
public class RedisConfig {
    /*
    redis의 pub/sub 기능을 이용하기 위해 MessageListener 설정 추가
    메시지 발행이 오면 Listener가 처리함
     */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(
            RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter,
            ChannelTopic channelTopic
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // RedisMessageListenerContainer 에 Bean 으로 등록한 listenerAdapter, channelTopic 추가
        container.addMessageListener(listenerAdapter, channelTopic);
        return container;
    }

    /*
    pub/sub 통신에 사용할 redisTemplate 설정
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(String.class));
        return redisTemplate;
    }

    /*
      실제 메시지를 처리하는 subscriber 설정 추가
      sendMessage 라는 메소드가 -> RedisSubscriber 클래스 안에 오버라이딩 된 onMessage 로 처리하도록 함
     */
    @Bean
    public MessageListenerAdapter listenerAdapter(RedisSubscriber subscriber) {
        return new MessageListenerAdapter(subscriber, "sendMessage");
    }

    /*
      단일 Topic 사용을 위한 Bean 설정
      사실 중요한 부분인지 모르겠음
      계속해서 생성할 필요가 없어 보여 빈으로 등록함
     */
    @Bean
    public ChannelTopic channelTopic() {
        return new ChannelTopic("chatroom");
    }
}

RedisRepository

Redis에 정보를 저장하고 가져오기 위해 구현해 주었다.

@RequiredArgsConstructor
@Service
public class RedisRepository {
    private static final String ENTER_INFO = "ENTER_INFO";
    private static final String USER_INFO = "USER_INFO";

    /**
     * "ENTER_INFO", roomId, userId (유저가 입장한 채팅방 정보)
     */
    @Resource(name = "redisTemplate")
    private HashOperations<String, Long, Long> chatRoomInfo;

    /**
     * 채팅방 마다 유저가 읽지 않은 메세지 갯수 저장
     * 1:1 채팅에서만 사용 O, 그륩채팅에서 사용 X
     * roomId, userId, 안 읽은 메세지 갯수
     */
    @Resource(name = "redisTemplate")
    private HashOperations<String, Long, Integer> chatRoomUnReadMessageInfo;

    /**
     * 상대 정보는 sessionId 로 저장, 나의 정보는 userId에 저장
     * "USER_INFO", sessionId, userId
     */
    @Resource(name = "redisTemplate")
    private HashOperations<String, String, Long> userInfo;

    @Resource(name = "redisTemplate")
    private HashOperations<String, Long, Long> test;

RedisSubscriber

@Slf4j
@RequiredArgsConstructor
@Service
public class RedisSubscriber implements MessageListener {
    private final ObjectMapper objectMapper;
    private final RedisTemplate redisTemplate;
    private final SimpMessageSendingOperations messagingTemplate;

    /**
     * Redis에서 메시지가 발행되면 대기하고 있던 onMessage가 메시지를 받아 messagingTemplate를 이용하여 websocket 클라이언트들에게 메시지 전달
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            // redis에서 발행된 데이터를 받아 역직렬화
            String publishMessage = (String) redisTemplate.getStringSerializer().deserialize(message.getBody());

            // 만약 ChatMessageRequest 클래스로 넘어왔다면
            if (objectMapper.canSerialize(ChatMessageRequest.class)) {
                // ChatMessage 객체로 맵핑
                ChatMessageRequest roomMessage = objectMapper.readValue(publishMessage, ChatMessageRequest.class);

                GetChatMessageResponse chatMessageResponse = new GetChatMessageResponse(roomMessage);

                // Websocket 구독자에게 채팅 메시지 전송
                messagingTemplate.convertAndSend("/sub/chat/room/" + roomMessage.getRoomId(), chatMessageResponse);

            } else {   // 만약 AlarmRequest 클래스로 넘어왔다면
                AlarmRequest alarmRequest = objectMapper.readValue(publishMessage, AlarmRequest.class);
                messagingTemplate.convertAndSend("/sub/chat/room/" + alarmRequest.getOtherUserId(), AlarmResponse.toDto(alarmRequest));
            }

        } catch (Exception e) {
            throw new ChatMessageNotFoundException();
        }
    }
}

MessageListener 인터페이스를 통해 오버라이딩 된 onMesage 메서드를 통해 처리한다. (RedisConfig 클래스에서 Bean으로 등록된 listenerAdapter 을 살펴보면 작동방식을 알 수 있다.

메시지 객체

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatMessageRequest implements Serializable {

    // 메시지 타입 : 채팅
    public enum MessageType {
        TALK, // 일대일 채팅
        GROUP_TALK, // 그륩 채팅
        UNREAD_MESSAGE_COUNT_ALARM // 안읽은 메세지
    }

    private Long messageId;
    private String roomTitle; // 그룹채팅일 경우 스터디방 제목
    private MessageType type; // 메시지 타입
    private String nickName;
    private Long roomId; // 공통으로 만들어진 방 번호
    private Set<Long> otherUserIds; // 상대방
    private String message; // 메시지
    private Long userId;
    private int count;
}

Controller

@RequiredArgsConstructor
@Controller
public class ChatMessageController {

    private final ChatMessageService chatMessageService;
    private final UserRepository userRepository;

    /**
     * websocket "/pub/chat/enter"로 들어오는 메시징을 처리한다.
     * 채팅방에 입장했을 경우
     */
    @MessageMapping("/chat/enter")
    public void enter(ChatMessageRequest chatMessageRequest) {
           User user = userRepository.findById(chatMessageRequest.getUserId()).orElseThrow(UserNotFoundException::new);

           chatMessageService.enter(user.getId(), chatMessageRequest.getRoomId());
    }

    /**
     * websocket "/pub/chat/message"로 들어오는 메시징을 처리한다.
     */
    @MessageMapping("/chat/message")
    public void message(ChatMessageRequest chatMessageRequest) {
           User user = userRepository.findById(chatMessageRequest.getUserId()).orElseThrow(UserNotFoundException::new);

           chatMessageService.sendMessage(chatMessageRequest, user);
    }
}

@MessageMapping("/chat/enter") 을 통해 유저의 정보를 저장해 주었다.

@MessageMapping("/chat/message") 을 통해 유저는 chatMessageRequest 객체에 원하는 방 주소로 메시지를 보낼 수 있다.

Service

@Service
@RequiredArgsConstructor
public class ChatMessageService {

    private final RedisRepository redisRepository;
    private final RedisTemplate redisTemplate;
    private final ChannelTopic channelTopic;
    private final ChatMessageRepository chatMessageRepository;
    private final ChatRoomRepository chatRoomRepository;
    private final UserRepository userRepository;

    // 채팅방 입장
    public void enter(Long userId, Long roomId) {

        // 그룹채팅은 해시코드가 존재하지 않고 일대일 채팅은 해시코드가 존재한다.
        ChatRoom chatRoom = chatRoomRepository.findById(roomId).orElseThrow(ChatRoomNotFoundException::new);

        // 채팅방에 들어온 정보를 Redis 저장
        redisRepository.userEnterRoomInfo(userId, roomId);

        // 그룹채팅은 해시코드가 존재하지 않고 일대일 채팅은 해시코드가 존재한다.
        if (chatRoom.getRoomHashCode() != 0) {
            redisRepository.initChatRoomMessageInfo(chatRoom.getId()+"", userId);
        }
    }

    //채팅
    @Transactional
    public void sendMessage(ChatMessageRequest chatMessageRequest, User user) {
        ChatRoom chatRoom = chatRoomRepository.findById(chatMessageRequest.getRoomId()).orElseThrow(ChatRoomNotFoundException::new);

        //채팅 생성 및 저장
        ChatMessage chatMessage = ChatMessage.builder()
                .chatRoom(chatRoom)
                .user(user)
                .message(chatMessageRequest.getMessage())
                .build();

        chatMessageRepository.save(chatMessage);
        String topic = channelTopic.getTopic();

        // ChatMessageRequest에 유저정보
        chatMessageRequest.setNickName(user.getNickname());
        chatMessageRequest.setUserId(user.getId());

        if (chatMessageRequest.getType() == ChatMessageRequest.MessageType.TALK) {
            // 일대일 채팅일 경우
            redisTemplate.convertAndSend(topic, chatMessageRequest);
            updateUnReadMessageCount(chatMessageRequest);
        } else {
            // 그륩 채팅일 경우
            redisTemplate.convertAndSend(topic, chatMessageRequest);
            redisTemplate.opsForHash();
        }
    }

    //안읽은 메세지 업데이트
    private void updateUnReadMessageCount(ChatMessageRequest chatMessageRequest) {
        Long otherUserId = chatMessageRequest.getOtherUserIds().stream().collect(Collectors.toList()).get(0);
        String roomId = String.valueOf(chatMessageRequest.getRoomId());

        if (!redisRepository.existChatRoomUserInfo(otherUserId) || !redisRepository.getUserEnterRoomId(otherUserId).equals(chatMessageRequest.getRoomId())) {

            redisRepository.addChatRoomMessageCount(roomId, otherUserId);
            int unReadMessageCount = redisRepository.getChatRoomMessageCount(roomId+"", otherUserId);

            String topic = channelTopic.getTopic();

            ChatMessageRequest messageRequest = new ChatMessageRequest(chatMessageRequest, unReadMessageCount);

            redisTemplate.convertAndSend(topic, messageRequest);
        }
    }

redisTemplate.convertAndSend(topic, chatMessageRequest); 통해 메세지를 Redis pub/sub 서버로 보내는 것이다. 그리고 sendMessage 부분이 빈으로 등록된 listenerAdapter 통해 → RedisSubscriber 클래스에 OnMessage 메서드와 연결이 된다.

로컬 환경에서 테스트하기 위한 테스트

@Profile("local") // profile이 로컬 환경일경우 내장 Redis 실행
@Configuration
public class EmbeddedRedisConfig {

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

    private RedisServer redisServer;

    @PostConstruct
    public void redisServer() {
        redisServer = new RedisServer(redisPort);
        redisServer.start();
    }

    @PreDestroy
    public void stopRedis() {
        if (redisServer != null) {
            redisServer.stop();
        }
    }
}

spring:
  redis:
    port: 6379
    host: localhost

  datasource:
    url: jdbc:h2:tcp://localhost/~/test
    username: sa
    password: 12
    driver-class-name: org.h2.Driver

  jpa:
    hibernate:
      ddl-auto: none
    properties:
      hibernate:
        format_sql: true

logging.level:
  org.hibernate.SQL: debug

server:
  port: 8089

부족했던 점

Redis에 대한 기본적인 이해도가 부족했다. 많은 기능 중 Redis pub/sub 방법만 사용해 보았기에 아쉬웠다.
캐시 메모리 데이터베이스라는 것만 알고만 있었고 실제 사용해 본 거 이번이 처음이었다. 하지만 실무에서 많이 쓰기 때문에 Redis는 다시 한번 공부해 볼 필요가 있다고 느꼈다.

채팅 파트를 구현하는데 대략 일주일정도 했던 거 같았다. 웹소켓대한 지식도 Redis에 대한 지식도 없는 상태로 시작했기에 일주일이라는 시간은 구현하는데 급급했던 거 같다.
기회가 된다면 웹소켓과 Redis 설정하는 부분을 체크해 보고 나중에 적용할 기회가 된다면 더 좋은 방식으로 구현해보고 싶다는 욕심이 생겼다.