채팅방 읽음 카운트 전파 방식에 대한 구현 방법

[unikal1]

기존에 한번 논의했던 주제이고 그 당시의 결론은 읽음 메시지를 다시 전체에게 브로드캐스팅하자... 였는데 아무리 생각해도 오버헤드가 굉장히 클 듯 하여 고민을 좀 많이 해봤습니다.
상기된 방식의 경우 N 명이 참여한 채팅방에 모두가 한번 씩 채팅을 치면 N^3 의 메시지가 돌아다닐 것이고 이거는 서버나 클라이언트 입장에서도 감당하기 힘들 듯 합니다. 따라서 일단 다음과 같은 방식을 구성하였습니다.

1. 각 유저는 메시지를 받으면 해당 메시지를 수신하였음을 서버에 전송하여야 한다.,

2. 서버는 이런 "읽음 완료" 메시지를 받으면 redis 에 바로 갱신한다.
   
   이는 각 채팅방 별 key 를 선택하고, 내부에는 userId - messageId 의 Set이 위치한다.

3. 동시에 자체적인 concurrentHashMap을 사용하여 roomId - Set의 map 을 관리한다. 이는 timeDelay(보통 0.5초) 마다 해당 기간 동안 채팅방(map 의 id)에서 변경된 사항 즉, 0.5초라는 기간 동안 어떤 유저가 어떤 글을 마지막으로 읽었는지에 대한 변경 사항을 저장하는 것이다.

• 여기에 messageId 를 넣지 않고 key 값들인 roomId, userId 만 넣은 까닭은, 추후 다클라 - 분산 시스템으로의 확장을 고려한 것이다. 하나의 유저가 컴퓨터 - 핸드폰으로 동시에 같은 채팅방에 접속하였으나, 웹소켓이 연결된 서버 인스턴스가 각기 다를 때, messageId 를 각 서버 인스턴스에서 관리하게 되면 클라이언트는 해당 유저가 "가장 마지막으로 읽은 메시지 아이디"가 중복되어 다르게 올 위험이 생긴다. 물론 현재 방식도 문제가 있는게, 결국 중복되어 동일한 메시지가 여러번 생성될 수도 있고 ShceduledExecutorService 에 의해(완전히 동일한 시간에 동시에 실행되는 것이 아니기에) 아주 짧은 간격으로 서로 다른 messageId 를 담은 응답이 올 수도 있지만, 분산 시스템으로의 완전한 확장 시 다음과 같은 방법을 고려하면 된다고 생각한다.
• redis의 pub/sub을 사용하여 각 서버 인스턴스는 자신이 처리한 roomId-userId 를 다른 서버 인스턴스에 브로드캐스팅하여, 해당 내용을 반영하였음을 전파하거나 / 위 hash 자료구조에서 check 필드를 추가하여, 갱신한 데이터의 경우 true 등을 설정하도록 하는 방법
• 물론 현재는 모노 서버 아키텍처라 딱히 고려할 필요가 없다고 생각한다.

[코드 예시]

@Service
@RequiredArgsConstructor
public class ChatReadCountServiceImpl {

    private final RedisTemplate<String, String> redisTemplate;
    private final SimpMessageSendingOperations template;
    private final Map<String, Set<Long>> readMap = new ConcurrentHashMap<>();
    private final Set<String> scheduled = ConcurrentHashMap.newKeySet();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);

    public void onUserRead(String roomId, Long userId, String messageId) {
        redisTemplate.opsForHash().put("room:" + roomId + ":reads",
                String.valueOf(userId), messageId);

        readMap.computeIfAbsent(roomId, k -> ConcurrentHashMap.newKeySet()).add(userId);

        if(!scheduled.add(roomId)) return;

        scheduler.schedule(() -> {
            try {
                broadcast(roomId);
            } finally {
                scheduled.remove(roomId);
                readMap.remove(roomId);
            }
        }, 500, TimeUnit.MILLISECONDS);


    }
    private void broadcast(String roomId) {


        Set<Long> changedUserIds = readMap.getOrDefault(roomId, Set.of());
        if (changedUserIds.isEmpty()) return;

        Map<Long, String> filtered = chatCursorRespository.getUserToMsgMapViaLua(roomId, changedUserIds);

        // 브로드캐스트
        ReadCursorBroadcastDto payload = new ReadCursorBroadcastDto(roomId, filtered);
        template.convertAndSend("/sub/chat/room/" + roomId, payload);
    }
        }

        ReadCursorBroadcastDto payload = new ReadCursorBroadcastDto(roomId, userToMsgId);

        template.convertAndSend("/sub/chat/room/", payload);
    }

}

@Repository
@RequiredArgsConstructor
public class ChatCursorRespository {
    private final RedisTemplate<String, String> redisTemplate;
    private final RedisScript<List> readCursorScript;



    public Map<Long, String> getUserToMsgMapViaLua(String roomId, Set<Long> userIds) {
        if(userIds.isEmpty()) return Map.of();

        String hasKey = "room:" + roomId + ":reads";

        List<String> keys = userIds.stream()
                .map(String::valueOf)
                .toList();

        List<String> args = List.of(hasKey);

        @SuppressWarnings("unchecked")
        List<Object> rawResult = redisTemplate.execute(readCursorScript, keys, args.toArray());

        Map<Long, String> result = new HashMap<>();
        if(rawResult == null) return result;

        for(int i = 0; i < rawResult.size(); i += 2) {
            Long userId = Long.valueOf(rawResult.get(i).toString());
            String messageId = rawResult.get(i + 1).toString();
            result.put(userId, messageId);
        }
        return result;
    }
}

4. 실제 코드에 따른 처리 순서는 다음과 같다.

• controller 로부터 roomId, userId, messageId 를 받아, (앞서 이야기 한 대로) redis에 해당 내용을 업데이트한다.
• readMap 에 해당 내용을 갱신 혹은 추가한다.
• 만약 scheduled 에 해당 roomId 가 이미 존재한다면, 그 즉시 해당 메서드를 종료한다.
• 만약 해당 roomId 가 이미 존재하지 않는다면, scheduledExecutorService 에 해당 채팅방에 대한 실행을 예약한다. 일단 0.5 초(500ms)로 설정하였고, 따라서 0.5초 간 해당 채팅방에 대한 정보를 수집한 뒤 이를 한번에 채팅방에게 전파한다.
• scheduledExecutorService(이후 scheduler) 가 실행되면, readMap 에서 해당 roomId에 따른 userId Set 을 가져온다.
• roomId 와 userId 를 기반으로 redis에서 messageId 에 대한 리스트를 받아온다. 이는 일단 LuaScript 를 사용하였으며 userId - messageId 로 매핑한다.
• 이후 해당 내용을 해당하는 채팅방 접속 중인 유저에게 브로드캐스팅한다.
• 클라이언트는 주기적으로 오는 해당 읽음 카운트 갱신 정보를 바탕으로, 각 메시지 별 읽음 카운트를 관리하여야 한다.

---

해당 코드는 일단 구현에 집중하여 역할 분리나 상수 처리 등이 제대로 이뤄지지 않았다는 점을 감안해주시고, 결론적으로 지향하는 바는,
순간 많은 양의 메시지가 서버에 오더라도, 최대 0.5초의 딜레이 후 값을 합하여 채팅방 별로 주기적으로 뿌려주는 것입니다.
각 채팅방이 아닌, 모든 채팅방에 대해 0.5초의 배치처리를 할까 생각은 해봤지만 순간 부하량이 높을 듯 하여 각 채팅방 별로 처리하도록 하였습니다.

하나 더 생각하는게 소규모 채팅방, 가령 10명 내의 채팅방의 경우는 처리 방법을 달리하여, 기존에 상의한 방법대로, 읽음 메시지를 다시 전체에게 브로드캐스팅 하는 것도 나쁘지는 않을 것 같네요.

해당 방법에 대해 추가적인 개선 방안이나, 제안하고자 할 사항이 있을까요? 사실 아직도 이게 맞는건지가 좀 헷갈리네요.