Scale Out을 위한 Redis Pub/Sub 기반 공개 채팅 아키텍처 도입

[mete0rfish]

User description

#️⃣ 이슈

• close

🔎 작업 내용

https://meteorfish.tistory.com/9

---

PR Type

Enhancement

---

Description

• Migrates chat functionality to Redis Pub/Sub for scalability.

  • Removes RabbitMQ dependencies.
  • Implements Redis publisher and subscriber for message handling.

• Improves WebSocket handling for chat rooms.

  • Manages user connections and disconnections.
  • Stores and retrieves recent messages from Redis.

• Adds a default chat room on application start.

---

Diagram Walkthrough

flowchart LR
  A["WebSocket Client"]
  B["ChatWebSocketHandler"]
  C["RedisPublisher"]
  D["RedisSubscriber"]
  E["ChatService"]
  F["Redis"]
  G["Database"]

  A -- "Connect/Send Message" --> B
  B -- "Publish Message" --> C
  C -- "Send to Channel" --> F
  F -- "Receive Message" --> D
  D -- "Send to Sockets" --> A
  B -- "Save Message" --> E
  E -- "Save to DB" --> G

Loading

File Walkthrough

	Relevant files
Enhancement	7 files ChatController.java Updates `getChatMessages` to use `findLatestMessages`        +1/-1      ChatWebSocketHandler.java Refactors WebSocket handler to use Redis Pub/Sub                  +92/-57  ChatRecentMessageService.java Updates `getRecentMessages` to use `findLatestMessages`    +1/-1      ChatService.java Implements Redis Pub/Sub and manages chat rooms                    +82/-23  RedisPublisher.java Implements Redis publisher for sending chat messages          +32/-0    RedisSubscriber.java Implements Redis subscriber for receiving chat messages    +53/-0    DataLoader.java Adds default chat room creation                                                    +7/-4
Dependencies	6 files ChatRabbitMqNames.java Removes RabbitMQ related constants                                              +0/-12    ChatConsumerService.java Removes RabbitMQ consumer service interface                            +0/-16    ChatConsumerServiceImpl.java Removes RabbitMQ consumer service implementation                  +0/-46    ChatProducerService.java Removes RabbitMQ producer service interface                            +0/-8      ChatProducerServiceImpl.java Removes RabbitMQ producer service implementation                  +0/-23    RabbitmqConfig.java Removes RabbitMQ configuration                                                      +0/-75
Miscellaneous	3 files TestChatWebSocketHandler.java Removes test WebSocket handler                                                      +0/-23    ChatProcessService.java Removes message queue processing service                                  +0/-45    TestDataLoader.java No changes                                                                                              +4/-3
Configuration changes	4 files DummyDataLoader.java Adds conditional property for dummy data loading                  +7/-7      WebSocketConfig.java Excludes WebSocket config from test profile                            +2/-1      RedisConfig.java Configures Redis connections and message listeners              +31/-0    application-local.properties Updates local DB dialect                                                                  +2/-2
Additional files	7 files ChatControllerTest.java +0/-117  ChatWebSocketHandlerTest.java +0/-155  ChatMessageQueueTest.java +2/-10    ChatServiceTest.java +0/-71    PostgresTestContainerBase.java +0/-23    RedisTestContainer.java +0/-4      RedisTestContainerBase.java +0/-36

---

[mete0rfish]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 No relevant tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Exception Handling The handleTextMessage method catches all exceptions but doesn't rethrow or handle them in a way that informs the client. Consider sending an error message to the client via the WebSocket to indicate that their message was not processed. protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { try { ChatMessage chatMessage = chatService.createMessage(message); chatService.saveMessage(chatMessage); String roomId = chatMessage.getRoomId(); String redisKey = "chat:room:" + roomId; String messageJson = objectMapper.writeValueAsString(chatMessage); chatRedisTemplate.opsForList().leftPush(redisKey, messageJson); chatRedisTemplate.opsForList().trim(redisKey, 0, 199); try { redisPublisher.publish(chatMessage); } catch (Exception e) { log.error("Failed to publish chat message to Redis, but it is saved in DB. Message: {}", chatMessage, e); } } catch (Exception e) { log.error("Failed to process chat message. SessionId: {}", session.getId(), e); } Potential Race Condition The chatRooms map is accessed and modified by multiple WebSocket sessions concurrently. Consider using a concurrent-safe data structure like ConcurrentHashMap to avoid potential race conditions. private Map<String, ChatRoom> chatRooms; private final ChatRepository chatRepository; @Qualifier("chatRedisTemplate") private final RedisTemplate<String, String> chatRedisTemplate; // Redis Topic 관련 의존성 추가 private final RedisMessageListenerContainer redisContainer; private final MessageListenerAdapter messageListener; private final RedisConfig redisConfig; // 기본 채팅방을 위한 고정 ID private static final String DEFAULT_ROOM_ID = "00000000-0000-0000-0000-000000000001"; @PostConstruct private void init() { chatRooms = new LinkedHashMap<>(); // 애플리케이션 시작 시, 모든 인스턴스가 동일한 ID를 가진 기본 채팅방을 생성 ChatRoom defaultChatRoom = ChatRoom.builder() .roomId(DEFAULT_ROOM_ID) .name("기본 채팅방") .build(); chatRooms.put(DEFAULT_ROOM_ID, defaultChatRoom); Exception Handling The onMessage method catches IOException but throws RuntimeException. It's better to handle the exception more gracefully, possibly by logging the error and taking appropriate action, instead of crashing the entire stream processing. @Override public void onMessage(Message message, byte[] pattern) { try { String publishMessage = new String(message.getBody()); ChatMessage chatMessage = objectMapper.readValue(publishMessage, ChatMessage.class); ChatRoom room = chatService.findRoomById(chatMessage.getRoomId()); Set<WebSocketSession> sessions = room.getSessions(); sendToEachSocket(sessions, new TextMessage(objectMapper.writeValueAsString(chatMessage))); } catch (IOException e) { log.error("Error processing message", e); } }

[mete0rfish]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Avoid auto-creating chatrooms if not exist Creating a new ChatRoom if one doesn't exist might lead to unintended consequences, such as creating a large number of empty rooms. Instead, return null if the room doesn't exist, or throw an exception to indicate that the room was not found. Handle the case where the room does not exist in the calling methods. server/src/main/java/com/onetool/server/api/chat/service/ChatService.java [62-65] public ChatRoom findRoomById(String roomId) { - // 채팅방이 없으면 새로 생성 (기본 채팅방 외 다른 채팅방이 필요할 경우를 대비) - return chatRooms.computeIfAbsent(roomId, id -> ChatRoom.builder().roomId(id).build()); + return chatRooms.get(roomId); } Suggestion importance[1-10]: 8 __ Why: The suggestion to avoid auto-creating chatrooms is valuable because it prevents the creation of empty rooms. Returning null if the room doesn't exist is a good approach, and the calling methods should handle this case. This change improves the logic and prevents unintended behavior.	Medium
	Replace session ID with username Instead of using the session ID as the sender's name, which is not user-friendly, retrieve the actual username. If the username is unavailable, consider using a default name like "Guest". This will improve the readability of chat messages. server/src/main/java/com/onetool/server/api/chat/handler/ChatWebSocketHandler.java [52-58] +String username = (String) session.getAttributes().get("username"); +if (username == null) { + username = "Guest"; // 기본 이름 설정 +} + ChatMessage enterMessage = ChatMessage.builder() .type(MessageType.ENTER) .roomId(roomId) - .sender(session.getId()) - .message(session.getId() + "님이 입장했습니다.") + .sender(username) + .message(username + "님이 입장했습니다.") .build(); Suggestion importance[1-10]: 7 __ Why: Replacing the session ID with a username improves the user experience. However, the suggestion assumes that the username is stored in the session attributes, which might not always be the case. It would be better to verify that the username exists and handle the case where it doesn't.	Medium
	Use constant for Redis key prefix Consider extracting the Redis key to a constant or configuration value for better maintainability and to avoid hardcoding the key format. This makes it easier to change the key structure in the future without modifying the code in multiple places. server/src/main/java/com/onetool/server/api/chat/handler/ChatWebSocketHandler.java [73-76] -String redisKey = "chat:room:" + roomId; +private static final String CHAT_ROOM_KEY_PREFIX = "chat:room:"; + +String redisKey = CHAT_ROOM_KEY_PREFIX + roomId; String messageJson = objectMapper.writeValueAsString(chatMessage); chatRedisTemplate.opsForList().leftPush(redisKey, messageJson); chatRedisTemplate.opsForList().trim(redisKey, 0, 199); Suggestion importance[1-10]: 6 __ Why: Extracting the Redis key to a constant improves maintainability. However, the impact is not very high.	Low