SpringBoot Redis를 활용한 Socket 통신(WebSocket, Stomp)

프로젝트를 하면서 유저들간 실시간 채팅방이 필요해 WebSocket,Stomp,Redis를 활용하여 채팅방을 구현하였습니다.

 

간단한 구조를 먼저 설명하자면 프론트쪽에서 HTTP를 통해 소켓연결을 하면 소켓 프로토콜이 연결되면서 실시간으로 서버와 클라이언트가 연결되게 된다. 클라이언트는 Request가 없어도 Response를 받을 수 있는 구조를 가지게 된다.

간단히 순서를 알아보면

1. 엔드포인트에 프론트가 소켓통신을 위한 신호를 보내 서버가 OK신호를 보내면 소켓 connect가 된다.
2. connect가 되면 프론트가 subscribe destination과 유저정보를 가진 요청을 보내고 데이터가 올떄까지 기다린다.
3. 앞으로 데이터를 subscirbe destination에 받는다.
4. 클라이언트는 send를 통해 publish로 서버에 데이터를 보내면 현재 subscribe를 하고 있는 유저들에게 들어온 데이터를 보내준다.
5. 유저가 채팅방에 나가게되면 subscribe가 풀리고 disconnect가 된다.

코드를 간단히 살펴보면 

Socket 기본 세팅 

@Slf4j
@Configuration
@RequiredArgsConstructor
@EnableWebSocketMessageBroker
public class WebSockConfig implements WebSocketMessageBrokerConfigurer {


    //서버와 처음 연결해주는 부분
    //WebSocket Open!
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        log.info("SOCKET 연결!");
        registry.addEndpoint("/ws").setAllowedOrigins("http://localhost:3000")
                .withSockJS();

    }

    //메세지 송수신을 처리하는 부분
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        //sub 로 보내면 이곳을 한번 거쳐서 프론트에 데이터를 전달해준다.
        registry.enableSimpleBroker("/sub");
        //pub 로 데이터를 받으면 이곳을 한번 거쳐서 URI 만 MessageMapping 에 매핑이 된다.
        //ex pub/chat/message 라면 pub 를 제외하고 /chat/message 를 @MessageMapping 에 매핑한다.
        registry.setApplicationDestinationPrefixes("/pub");
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(stompHandler);
    }


}

클라이언트에서 요청이 왔을떄 데이터를 보내기전에 먼저 어떤 요청인지 확인하고 처리하는 인터셉터

@Slf4j
@RequiredArgsConstructor
@Component
public class ChattingHandler implements ChannelInterceptor {

    private final JwtAuthenticationProvider jwtAuthenticationProvider;
    private final ChatService chatService;
    private final RedisRepository redisRepository;


    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {

        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        if (accessor.getCommand() == StompCommand.CONNECT) {
            String jwtToken = accessor.getFirstNativeHeader("token");
            jwtAuthenticationProvider.validateToken(jwtToken);
            log.info("Connect = {}", jwtToken);

        } else if (accessor.getCommand() == StompCommand.SUBSCRIBE) {
            String simpleDestination = (String) message.getHeaders().get("simpDestination");
            if (simpleDestination == null) {
                throw new RoomNotFoundException("존재하지 않는 방입니다.");
            }
            log.info("simpleDestination = {}", simpleDestination );
            String roomId = chatService.getRoomId(simpleDestination);

            String simpSessionId = (String) message.getHeaders().get("simpSessionId");
            log.info("roomId ={}  simpSessionId = {}" , roomId, simpSessionId);
            redisRepository.mappingUserRoom( simpSessionId,roomId);
            String userEnterRoomId = redisRepository.getUserEnterRoomId(simpSessionId);
            String jwtToken = accessor.getFirstNativeHeader("token");
            log.info("구독성공 {}, {}", simpSessionId, userEnterRoomId);
            chatService.messageResolver(new MessageRequestDto(Long.parseLong(roomId), MessageType.ENTER, ""), jwtToken);
            log.info("SUBSCRIBE {}, {}", simpSessionId, roomId);
        }else if (StompCommand.DISCONNECT == accessor.getCommand()) { // Websocket 연결 종료

            // 연결이 종료된 클라이언트 sesssionId로 채팅방 id를 얻는다.
            String sessionId = (String) message.getHeaders().get("simpSessionId");
            //나갈떄 redis 맵에서 roomId와 sessionId의 매핑을 끊어줘야 하기때문에 roomId찾고
            String roomId = redisRepository.getUserEnterRoomId(sessionId);

            // 퇴장한 클라이언트의 roomId 맵핑 정보를 삭제한다.
            redisRepository.removeUserEnterInfo(sessionId);
            log.info("DISCONNECT {}, {}", sessionId, roomId);
        }
        return message;
    }

}

레디스 설정 EC2에 미리 깔아둔 후 yml파일을 통해 설정해 연결함

@Repository
@RequiredArgsConstructor
public class RedisRepository{

    @Resource(name = "redisTemplate")
    private HashOperations<String, String, String> userRoomMap;
    public static final String ENTER_INFO = "ENTER_INFO";

    public void mappingUserRoom(String sessionId, String roomId){
        userRoomMap.put(ENTER_INFO, sessionId, roomId);
    }

    public String getUserEnterRoomId(String sessionId) {
        return userRoomMap.get(ENTER_INFO, sessionId);
    }

    public void removeUserEnterInfo(String sessionId) {
        userRoomMap.delete(ENTER_INFO, sessionId);
    }


}

ws.send를 통해 보냈을때 SockConfig에서 pub로 잡고 MessageMapping에 보내준다

@RestController
@RequiredArgsConstructor
public class ChatController {

    private final ChatService chatService;

    @MessageMapping("/api/chat/message")
    public void message(@RequestBody MessageRequestDto messageRequestDto, @Header("token") String token){
        chatService.messageResolver(messageRequestDto,token);
    }
}

채팅 서비스에서 데이터를 1차적으로 보내준다.

@Service
@RequiredArgsConstructor
@Slf4j
@Transactional
public class ChatService {

    private final ChannelTopic channelTopic;
    private final RedisTemplate redisTemplate;
    private final JwtAuthenticationProvider jwtAuthenticationProvider;
    private final UserRepository userRepository;
    private final RoomRepository roomRepository;
    private final MessageRepository messageRepository;


    public void messageResolver(MessageRequestDto messageRequestDto, String token) {
        Long userId = Long.parseLong(jwtAuthenticationProvider.getuserId(token));
        User user = userRepository.findById(userId).orElseThrow(
                () -> new UserNotFoundException("해당 유저는 존재하지 않습니다.")
        );
        Room room = roomRepository.findById(messageRequestDto.getRoomId()).orElseThrow(
                () -> new RoomNotFoundException("해당 방은 존재하지 않습니다.")
        );
        String dateResult = getTime();
        Message message = new Message(messageRequestDto,user,room,dateResult);
        log.info(" messageResolver 여기까진 잘왔어");
        sendMessage(message);
        messageRepository.save(message);

    }

    public void sendMessage(Message message) {
        if(message.getMessageType().equals(MessageType.ENTER)){
            for (UserEnterRoom userEnterRoom : message.getUser().getUserEnterRoomList()) {
                if(userEnterRoom.getUser().getId().equals(message.getUser().getId())){
                    if(userEnterRoom.getRoomUserStatus() == RoomUserStatus.ENTER){
                        message.setMessage(message.getUser().getNickName()+"님이 입장하셨습니다.");
                        userEnterRoom.setRoomUserStatus(RoomUserStatus.CHAT);
                    }
                }

            }

        }else if(message.getMessageType().equals(MessageType.QUIT)){
            for (UserEnterRoom userEnterRoom : message.getUser().getUserEnterRoomList()) {
                if(userEnterRoom.getUser().getId().equals(message.getUser().getId())){
                    if(userEnterRoom.getRoomUserStatus() == RoomUserStatus.CHAT){
                        message.setMessage(message.getUser().getNickName()+"님이 퇴장하셨습니다.");
                        userEnterRoom.setRoomUserStatus(RoomUserStatus.QUIT);
                    }
                }

            }

        }
        log.info("sendMessage여기 까지 잘왔어");
        log.info("sendMessage = {}", message.getMessage());
        MessageResponseDto messageResponseDto = new MessageResponseDto(message);
        log.info("messageResponseDto = {}", messageResponseDto );
        redisTemplate.convertAndSend(channelTopic.getTopic(), messageResponseDto);
    }

    private String getTime() {
        SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd HH:mm");
        Calendar cal = Calendar.getInstance();
        Date date = cal.getTime();
        sdf.setTimeZone(TimeZone.getTimeZone("Asia/Seoul"));
        String dateResult = sdf.format(date);
        return dateResult;
    }
    public String getRoomId(String destination) {
        int lastIndex = destination.lastIndexOf('/');
        if (lastIndex != -1)
            return destination.substring(lastIndex + 1);
        else
            return "";
    }


    public Page<ChatMessage> getChatMessageByRoomId(String roomId, Pageable pageable) {
        int page = (pageable.getPageNumber() == 0) ? 0 : (pageable.getPageNumber() -1);
        pageable = PageRequest.of(page, 150);
        return chatMessageRepository.findByRoomId(roomId, pageable);
    }

}

redisTemplate.convertAndSend(channelTopic.getTopic(), messageResponseDto);를 보내면 

RedisConfig에서 설정한 ListnerAdpater를 통해 Subscriber에서의 sendMessage에서 실제로 데이터를 보내준다.

@Service
@Slf4j
public class RedisSubscriber {

    private final ObjectMapper objectMapper;
    private final SimpMessageSendingOperations messagingTemplate;

    @Autowired
    public RedisSubscriber(ObjectMapper objectMapper, SimpMessageSendingOperations messagingTemplate) {
        this.objectMapper = objectMapper;
        this.messagingTemplate = messagingTemplate;
    }

    public void sendMessage(String publishMessage){
        try{
            MessageResponseDto messageResponseDto = objectMapper.readValue(publishMessage,MessageResponseDto.class);
            messagingTemplate.convertAndSend("/sub/api/chat/rooms" + messageResponseDto.getRoomId(), messageResponseDto);
        }catch (Exception e){
            log.error("Exception {}", e);
        }
    }

}

github주소 : https://github.com/minkik715/naegahama/tree/message

참고한 사이트https://daddyprogrammer.org/post/series/spring-websocket-chat-server/

https://docs.spring.io/spring-framework/docs/4.3.x/spring-framework-reference/html/websocket.htm   

https://dev-gorany.tistory.com/m/212