Redis Pub/Sub을 활용한 채팅 구현의 여정 - Service 레이어

서비스단 코드를 먼저 보여주고 각 메서드에 대해 설명하겠다.

@Service
@RequiredArgsConstructor
@Slf4j
public class ChatServiceImpl implements ChatService {

    private final ChatRoomRepository chatRoomRepository;
    private final RedisTemplate<String, String> redisTemplate;
    private final RedisTemplate<String, ChattingResponseDto> redisTemplateForSave;
    private final RedisTemplate<String, ChannelTopic> redisTopicTemplate;
    private final UserRepository userRepository;
    private final NotificationService notificationServiceImpl;
    private final RedisPublisher redisPublisher;
    private final WebSocketEventListener webSocketEventListener;

    @Override
    public ChatRoomCreateResponse getOrCreateChatRoom(Long sender, Long receiver, String roomTitle) {
        log.info("Getting or creating chat room. Sender: {}, Receiver: {}, RoomTitle: {}", sender, receiver, roomTitle);

        User findSender = findSender(sender);
        User findReceiver = findReceiver(receiver);

        log.info("Sender found: {}. Receiver found: {}", findSender, findReceiver);

        String findSenderProfileImage = getProfileImage(findSender);
        String findReceiverProfileImage = getProfileImage(findReceiver);

        ChatRoom findChatRoom = createChatRoom(sender, receiver, roomTitle, findSenderProfileImage, findReceiverProfileImage);

        findChatRoom = validateChatRoom(roomTitle, findChatRoom);

        log.info("Chat room created or validated: {}", findChatRoom);

        if (findChatRoom.getOwner().equals(findSender.getUuid())) {
            log.info("Returning chat room with title: {}", findChatRoom.getTitle());
            return ChatRoomCreateResponse.builder()
                    .roomTitle(findChatRoom.getTitle())
                    .roomId(findChatRoom.getRoomId())
                    .build();
        }

        log.info("Returning chat room with receiver's nickname: {}", findReceiver.getNickname());
        return ChatRoomCreateResponse.builder()
                .roomTitle(findReceiver.getNickname())
                .roomId(findChatRoom.getRoomId())
                .build();
    }

    private ChatRoom createChatRoom(Long sender, Long receiver, String roomTitle, String findSenderProfileImage, String findReceiverProfileImage) {
        log.info("Creating chat room if not exists. Sender: {}, Receiver: {}, Title: {}", sender, receiver, roomTitle);
        return chatRoomRepository.findByOwnerAndMateAndTitle(sender, receiver, roomTitle, receiver, sender, roomTitle)
                .orElseGet(() -> {
                    log.info("No existing chat room found. Creating a new one.");
                    ChatRoom newChatRoom = ChatRoom.to(receiver, sender, findSenderProfileImage, roomTitle, findReceiverProfileImage);

                    chatRoomRepository.save(newChatRoom);
                    log.info("New chat room saved: {}", newChatRoom);

                    setTopicInRedisTemplate(newChatRoom);
                    return newChatRoom;
                });
    }

    private void setTopicInRedisTemplate(ChatRoom newChatRoom) {
        log.info("Setting topic in Redis template for chat room: {}", newChatRoom.getRoomId());
        ChannelTopic topic = new ChannelTopic("/sub/chat/room/" + newChatRoom.getRoomId());
        redisTopicTemplate.opsForValue().set("chatTopic:" + newChatRoom.getRoomId(), topic);
    }

    private ChatRoom validateChatRoom(String roomTitle, ChatRoom findChatRoom) {
        log.info("Validating chat room title.");
        if (findChatRoom.getTitle().isEmpty() || roomTitle != null) {
            log.info("Updating chat room title: {}", roomTitle);
            ChatRoom updateChatRoom = findChatRoom.toBuilder()
                    .title(roomTitle)
                    .build();

            findChatRoom = chatRoomRepository.save(updateChatRoom);
        }
        return findChatRoom;
    }

    private String getProfileImage(User findSender) {
        log.info("Getting profile image for user: {}", findSender.getUuid());
        return findSender.getProfileImage().isEmpty() ? null : findSender.getProfileImage();
    }

    private User findReceiver(Long receiver) {
        log.info("Finding receiver by ID: {}", receiver);
        return userRepository.findById(receiver).orElseThrow(
                () -> {
                    log.error("Receiver not found: {}", receiver);
                    return new MemberException(ExceptionCode.NOT_FOUND_MEMBER);
                }
        );
    }

    private User findSender(Long sender) {
        log.info("Finding sender by ID: {}", sender);
        return userRepository.findById(sender).orElseThrow(
                () -> {
                    log.error("Sender not found: {}", sender);
                    return new MemberException(ExceptionCode.NOT_FOUND_MEMBER);
                }
        );
    }

    @Override
    public String calculateTimeAgo(Timestamp lastTime) {
        log.info("Calculating time ago for timestamp: {}", lastTime);
        long diffInMillis = System.currentTimeMillis() - lastTime.getTime();
        long diffInMinutes = TimeUnit.MILLISECONDS.toMinutes(diffInMillis);

        if (diffInMinutes < 60) {
            return diffInMinutes + "분 전";
        } else {
            long diffInHours = TimeUnit.MILLISECONDS.toHours(diffInMillis);
            if (diffInHours < 24) {
                return diffInHours + "시간 전";
            } else {
                long diffInDays = TimeUnit.MILLISECONDS.toDays(diffInMillis);
                return diffInDays + "일 전";
            }
        }
    }

    @Override
    public List<ChatRoomResponseDto> getChatRoomList(Long uuid) {
        log.info("Fetching chat room list for user: {}", uuid);
        List<ChatRoom> chatRooms = chatRoomRepository.findByOwnerOrMate(uuid, uuid);
        List<ChatRoomResponseDto> chatRoomList = new ArrayList<>();

        for (ChatRoom room : chatRooms) {
            log.info("Processing chat room: {}", room.getRoomId());
            User findSender = findSender(room.getOwner());
            User findReceiver = findReceiver(room.getMate());

            //채팅 가져오기
            String key = "chatRoomId:" + room.getRoomId();
            List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);

            // 가장 최근 메시지 가져오기
            ChattingResponseDto lastMessage = null;
            if (chatList != null && !chatList.isEmpty()) {
                lastMessage = chatList.get(chatList.size() - 1); // 리스트의 마지막 요소
            }

            // 로그로 확인
            if (lastMessage != null) {
               log.info("가장 최근 메시지: {}", lastMessage.getContent());
            } else {
                log.info("메시지가 없습니다.");
            }

            Timestamp lastTime = room.getLastTime();
            List<ChattingResponseDto> unreceivedMessages = getMessagesAfterLastTime(room.getRoomId(), lastTime, uuid);

            ChatRoomResponseDto chatroom = ChatRoomResponseDto.to(room, findSender, findReceiver, unreceivedMessages, lastMessage);

            chatRoomList.add(chatroom);
        }
        log.info("Chat room list fetched successfully.");
        return chatRoomList;
    }

    @Override
    public void saveChatToRedis(String roomId, ChattingResponseDto chat) {
        log.info("Saving chat to Redis. Room ID: {}, Chat: {}", roomId, chat);
        String key = "chatRoomId:" + roomId;

        List<ChattingResponseDto> chatList = loadMessageFromRedis(key);
        saveMessages(chat, chatList, key);
    }

    private void saveMessages(ChattingResponseDto chat, List<ChattingResponseDto> chatList, String key) {
        log.info("Checking for duplicate messages in Redis.");
        boolean isDuplicate = chatList.stream().anyMatch(savedChat ->
                savedChat.getLastTime().equals(chat.getLastTime()) &&
                        savedChat.getContent().equals(chat.getContent())
        );

        if (!isDuplicate) {
            log.info("Message is not a duplicate. Saving to Redis.");
            redisTemplateForSave.opsForList().rightPush(key, chat);
            redisTemplateForSave.expire(key, 3, TimeUnit.DAYS);
        } else {
            log.warn("Duplicate message detected. Not saving to Redis.");
        }
    }

    private List<ChattingResponseDto> loadMessageFromRedis(String key) {
        log.info("Fetching messages from Redis. Key: {}", key);
        List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);
        if (chatList == null) {
            log.info("No messages found in Redis for key: {}", key);
            chatList = new ArrayList<>();
        }
        return chatList;
    }

    @Override
    public void leaveRoom(Long roomId) {
        log.info("User leaving chat room: {}", roomId);
        chatRoomRepository.deleteById(roomId);
    }

    @Override
    public Timestamp getParsedLastTime(String lastTime) {
        log.info("Parsing timestamp: {}", lastTime);
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            return new Timestamp(dateFormat.parse(lastTime).getTime());
        } catch (Exception e) {
            log.error("Failed to parse timestamp: {}", lastTime, e);
            return new Timestamp(System.currentTimeMillis());
        }
    }

    @Override
    public List<ChattingResponseDto> getMessagesAfterLastTime(Long roomId, Timestamp lastTime, Long uuid) {
        log.info("Fetching messages after last time. Room ID: {}, Last Time: {}", roomId, lastTime);
        User findUser = getFindUser(uuid);

        // Redis에서 유저의 마지막 세션 종료 시간 가져오기
        log.info("유저의 마지막 세션 종료 시간 추출");
        Timestamp disconnectTime = getDisconnectTimeByUserId(lastTime, findUser);

        //채팅 가져오기
        String key = "chatRoomId:" + roomId;
        List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);

        if (chatList == null || chatList.isEmpty()) {
            log.warn("No messages found for roomId: {}", roomId);
            return new ArrayList<>();
        }

        List<ChattingResponseDto> unreceivedMessages = chatList.stream()
                .filter(message -> !message.getUserId().equals(findUser.getUuid()))
                .filter(message -> message.getLastTime().after(disconnectTime)) // 세션 종료 시간을 기준으로 필터링
                .sorted(Comparator.comparing(ChattingResponseDto::getLastTime).reversed()) //가장 최근 메세지가 위에 오도록
                .toList();

        log.info("Unreceived messages fetched for roomId: {}. Count: {}", roomId, unreceivedMessages.size());
        return unreceivedMessages;
    }

    private Timestamp getDisconnectTimeByUserId(Timestamp lastTime, User findUser) {
        String disconnectTimeKey = "session:lastDisconnected:" + findUser.getUuid();
        String disconnectTimeStr = redisTemplate.opsForValue().get(disconnectTimeKey);
        Timestamp disconnectTime;

        if (disconnectTimeStr != null) {
            disconnectTime = new Timestamp(Long.parseLong(disconnectTimeStr));
            log.info("User last disconnect time found: {}", disconnectTime);
        } else {
            log.warn("No disconnect time found for userId: {}. Using provided lastTime: {}", findUser.getUuid(), lastTime);
            disconnectTime = lastTime;
        }
        return disconnectTime;
    }

    private User getFindUser(Long uuid) {
        return userRepository.findByUuid(uuid).orElseThrow(
                () -> new MemberException(ExceptionCode.NOT_FOUND_MEMBER)
        );

    }

    @Override
    public void sendMessageToPublisher(ChattingRequestDto message) {
        log.info("{} send This Message - {}", message.getUserId(), message.getContent());
        log.info("message contain Image : {}", message.getImage());

        //1.해당 채팅방 조회
        ChatRoom findChatRoom = findChatRoom(message);

        //2.마지막 채팅 시간 추출
        Timestamp parsedLastTime = getParsedLastTime(message.getLastTime());

        // 메세지 받을 사람 지정
        // 오너가 보내면 메이트, 메이트가보내면 오너로 지정
        Long receiver = isMateOrOwner(message, findChatRoom);

        //유저가 세션에 참여중이지 않으면 fcm 알림을 보냄
        String sessionId = redisTemplate.opsForValue().get("user:session:" + receiver);
        if (sessionId == null || !webSocketEventListener.isSessionConnected(sessionId)) {
            log.warn("User {} is offline. Sending notification.", receiver);
            sendNotificationToDisConnectedUser(message, findChatRoom, parsedLastTime, receiver);
        }

        //채팅 응답 생성
        ChattingResponseDto responseDto = ChattingResponseDto.to(message, parsedLastTime);

        //응답 Redis에 저장
        saveChatToRedis(String.valueOf(message.getRoomId()), responseDto);

        //Pub에 내용을 퍼블리싱
        redisPublisher.publish(responseDto);
    }

    private static Long isMateOrOwner(ChattingRequestDto message, ChatRoom findChatRoom) {
        log.info("isMateOrOwner를 수행하여 받을 사람 지정");
        if (findChatRoom.getMate().equals(message.getUserId())) {
            log.info("receiver id = {}", findChatRoom.getOwner());
            return findChatRoom.getOwner();
        } else {
            log.info("receiver id = {}", findChatRoom.getMate());
            return findChatRoom.getMate();
        }
    }

    @Override
    public ChatRoomResponseDto getChatRoom(Long uuid, Long roomId) {
        ChatRoom chatRoom = chatRoomRepository.findById(roomId).orElseThrow(
                () -> new ChatException(ExceptionCode.NOT_FOUND_CHATROOM)
        );

        log.info("Processing chat room: {}", chatRoom.getRoomId());
        User findSender = findSender(chatRoom.getOwner());
        User findReceiver = findReceiver(chatRoom.getMate());

        //채팅 가져오기
        String key = "chatRoomId:" + roomId;
        List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);

        // 가장 최근 메시지 가져오기
        ChattingResponseDto lastMessage = null;
        if (chatList != null && !chatList.isEmpty()) {
            lastMessage = chatList.get(chatList.size() - 1); // 리스트의 마지막 요소
        }

        // 로그로 확인
        if (lastMessage != null) {
            log.info("가장 최근 메시지: " + lastMessage.getContent());
        } else {
            log.info("메시지가 없습니다.");
        }
        Timestamp lastTime = chatRoom.getLastTime();
        List<ChattingResponseDto> unreceivedMessages = getMessagesAfterLastTime(chatRoom.getRoomId(), lastTime, uuid);

        return ChatRoomResponseDto.to(chatRoom, findSender, findReceiver, unreceivedMessages, lastMessage);
    }

    private void sendNotificationToDisConnectedUser(ChattingRequestDto message, ChatRoom findChatRoom, Timestamp parsedLastTime, Long receiver) {
        log.info("Sending notification to disconnected user: {}", receiver);
        NotificationRequestDto notificationRequestDto = NotificationRequestDto.to(message, findChatRoom, parsedLastTime, receiver);
        try {
            notificationServiceImpl.sendNotification(notificationRequestDto);
            log.info("Notification sent successfully to user: {}", receiver);
        } catch (InterruptedException e) {
            log.error("Failed to send notification to user: {}", receiver, e);
            throw new ChatException(ExceptionCode.INTERRUPTION_OR_EXECUTION_ERR);
        }catch (Exception e) {
            log.error("Failed to send notification to user: {}", receiver, e);
        }
    }

    private ChatRoom findChatRoom(ChattingRequestDto message) {
        log.info("Finding chat room by ID: {}", message.getRoomId());
        return chatRoomRepository.findById(message.getRoomId()).orElseThrow(
                () -> {
                    log.error("Chat room not found: {}", message.getRoomId());
                    return new ChatException(ExceptionCode.NOT_FOUND_CHATROOM);
                }
        );
    }
}

 

1. getOrCreateChatRoom

채팅방에 접속하는 두 명의 유저와 채팅방을 생성한 기록이 있는지 확인하고 있다면 해당 채팅방을, 없다면 새로운 채팅방을 생성하는 메서드이다. 내부 메서드는 간단한 equal연산 및 유저의 정보를 찾는 쿼리를 날리는 repository 연결 메서드이므로 넘어가겠다.

@Override
public ChatRoomCreateResponse getOrCreateChatRoom(Long sender, Long receiver, String roomTitle) {
    log.info("Getting or creating chat room. Sender: {}, Receiver: {}, RoomTitle: {}", sender, receiver, roomTitle);

    User findSender = findSender(sender);
    User findReceiver = findReceiver(receiver);

    log.info("Sender found: {}. Receiver found: {}", findSender, findReceiver);

    String findSenderProfileImage = getProfileImage(findSender);
    String findReceiverProfileImage = getProfileImage(findReceiver);

    ChatRoom findChatRoom = createChatRoom(sender, receiver, roomTitle, findSenderProfileImage, findReceiverProfileImage);

    findChatRoom = validateChatRoom(roomTitle, findChatRoom);

    log.info("Chat room created or validated: {}", findChatRoom);

    if (findChatRoom.getOwner().equals(findSender.getUuid())) {
        log.info("Returning chat room with title: {}", findChatRoom.getTitle());
        return ChatRoomCreateResponse.builder()
                .roomTitle(findChatRoom.getTitle())
                .roomId(findChatRoom.getRoomId())
                .build();
    }

    log.info("Returning chat room with receiver's nickname: {}", findReceiver.getNickname());
    return ChatRoomCreateResponse.builder()
            .roomTitle(findReceiver.getNickname())
            .roomId(findChatRoom.getRoomId())
            .build();
}

 

2.  getRoomList

채팅방의 리스트를 출력하면서 최근 채팅 내역중 마지막 메세지내용과 전송 시간을 함께 담아 출력한다. 리스트와 함께 출력하는 데이터들은 구현하고자 하는 방향에 맞추면 된다.

@Override
public List<ChatRoomResponseDto> getChatRoomList(Long uuid) {
    log.info("Fetching chat room list for user: {}", uuid);
    List<ChatRoom> chatRooms = chatRoomRepository.findByOwnerOrMate(uuid, uuid);
    List<ChatRoomResponseDto> chatRoomList = new ArrayList<>();

    for (ChatRoom room : chatRooms) {
        log.info("Processing chat room: {}", room.getRoomId());
        User findSender = findSender(room.getOwner());
        User findReceiver = findReceiver(room.getMate());

        //채팅 가져오기
        String key = "chatRoomId:" + room.getRoomId();
        List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);

        // 가장 최근 메시지 가져오기
        ChattingResponseDto lastMessage = null;
        if (chatList != null && !chatList.isEmpty()) {
            lastMessage = chatList.get(chatList.size() - 1); // 리스트의 마지막 요소
        }

        // 로그로 확인
        if (lastMessage != null) {
           log.info("가장 최근 메시지: {}", lastMessage.getContent());
        } else {
            log.info("메시지가 없습니다.");
        }

        Timestamp lastTime = room.getLastTime();
        List<ChattingResponseDto> unreceivedMessages = getMessagesAfterLastTime(room.getRoomId(), lastTime, uuid);

        ChatRoomResponseDto chatroom = ChatRoomResponseDto.to(room, findSender, findReceiver, unreceivedMessages, lastMessage);

        chatRoomList.add(chatroom);
    }
    log.info("Chat room list fetched successfully.");
    return chatRoomList;
}

 

3. getMessageAfterLastTime

미수신 메세지를 가져오는 메서드이다. 나는 AWS 프리티어를 사용하기 때문에 실제 운영시 채팅 데이터를 오랫동안 보관하기 힘들다 판단하여 클라이언트단의 로컬에 채팅내용을 기록하고 라이프사이클을 관리하자고 프론트팀원과 기획과 이야기하여 DB에는 저장하지 않고 Redis에 채팅 기록을 3일간 저장하는 TTL을 설정했다. 하지만 이 방법또한 적절하지 못하다 생각하여 1일단위로 채팅을 Redis에 저장하고 DB에 채팅 기록을 저장하거나 S3를 통해 파일을 가져와 읽고 캐싱하는 방법을 사용할지 고려중이다.

어찌됐든 이 메서드는 클라이언트에서 마지막 유저의 채팅창 접근 시간을 파라미터로 받으면 해당 시간 이후의 데이터를 레디스에서 가져와서 반환해준다.

@Override
public List<ChattingResponseDto> getMessagesAfterLastTime(Long roomId, Timestamp lastTime, Long uuid) {
    log.info("Fetching messages after last time. Room ID: {}, Last Time: {}", roomId, lastTime);
    User findUser = getFindUser(uuid);

    // Redis에서 유저의 마지막 세션 종료 시간 가져오기
    log.info("유저의 마지막 세션 종료 시간 추출");
    Timestamp disconnectTime = getDisconnectTimeByUserId(lastTime, findUser);

    //채팅 가져오기
    String key = "chatRoomId:" + roomId;
    List<ChattingResponseDto> chatList = redisTemplateForSave.opsForList().range(key, 0, -1);

    if (chatList == null || chatList.isEmpty()) {
        log.warn("No messages found for roomId: {}", roomId);
        return new ArrayList<>();
    }

    List<ChattingResponseDto> unreceivedMessages = chatList.stream()
            .filter(message -> !message.getUserId().equals(findUser.getUuid()))
            .filter(message -> message.getLastTime().after(disconnectTime)) // 세션 종료 시간을 기준으로 필터링
            .sorted(Comparator.comparing(ChattingResponseDto::getLastTime).reversed()) //가장 최근 메세지가 위에 오도록
            .toList();

    log.info("Unreceived messages fetched for roomId: {}. Count: {}", roomId, unreceivedMessages.size());
    return unreceivedMessages;
}

 

4. sendMessageToPublisher

Pub/Sub중 요청한 메세지를 publisher 채널에 먼저 전달하는 메서드이다. publisher에게 메세지가 전달되면 해당 내용을 구독한 Subscirber들에게 메세지가  브로커에의해 전달되기 때문이다.

이때 미접속한 유저들은 채팅 내용을 확인할 수 없으므로 FCM을 통해 메세지를 알림으로 전달해준다.

@Override
public void sendMessageToPublisher(ChattingRequestDto message) {
    log.info("{} send This Message - {}", message.getUserId(), message.getContent());
    log.info("message contain Image : {}", message.getImage());

    //1.해당 채팅방 조회
    ChatRoom findChatRoom = findChatRoom(message);

    //2.마지막 채팅 시간 추출
    Timestamp parsedLastTime = getParsedLastTime(message.getLastTime());

    // 메세지 받을 사람 지정
    // 오너가 보내면 메이트, 메이트가보내면 오너로 지정
    Long receiver = isMateOrOwner(message, findChatRoom);

    //유저가 세션에 참여중이지 않으면 fcm 알림을 보냄
    String sessionId = redisTemplate.opsForValue().get("user:session:" + receiver);
    if (sessionId == null || !webSocketEventListener.isSessionConnected(sessionId)) {
        log.warn("User {} is offline. Sending notification.", receiver);
        sendNotificationToDisConnectedUser(message, findChatRoom, parsedLastTime, receiver);
    }

    //채팅 응답 생성
    ChattingResponseDto responseDto = ChattingResponseDto.to(message, parsedLastTime);

    //응답 Redis에 저장
    saveChatToRedis(String.valueOf(message.getRoomId()), responseDto);

    //Pub에 내용을 퍼블리싱
    redisPublisher.publish(responseDto);
}

 

 

이것들이 주요 메서드이고 그 외에 시간을 적절한 Format방식으로 변환하는 getParsedLastTime 메서드, 채팅방을 삭제하는 leaveRoom메서드가 있다.

 

다음에는 유저의 채팅방 접근을 감지하는 WebsocketListener클래스에 대해 공유하겠다.

 

채팅기능은 정상적을 잘 수행되지만 아직 채팅 저장방식에 대해 고민이 많다. 좋은 방식이 있다면 댓글로 공유 부탁드립니다!