[unispace] 실시간 예약 시스템 구현(5) - ConcurrentHashMap과 웹 소켓 연결 클라이언트 개수 관리

Deeeep Breath·2024년 8월 6일

unispace

목록 보기

9/12

사용자가 현재 자신이 접속한 예약페이지에 몇명이 접속했는지 확인할 수 있도록 하는 기능을 추가하려고 한다.

클라이언트가 예약 페이지에 접속할 때 웹 소켓이 연결되고, /topic/reservations/${roomId} 를 구독하게 된다. 이 과정에서 서버로 메시지를 전송하여 해당 주소를 구독하는 클라이언트의 수를 저장하도록 했다.

구현

먼저 클라이언트의 접속 현황을 저장하는 클래스를 만들었다.

@Component
@RequiredArgsConstructor
public class ReservationWebSocketConnectionService {
    private final ConcurrentHashMap<Long, AtomicInteger> clientCount = new ConcurrentHashMap<>();
    
    // 구독 메서드: 클라이언트가 예약 페이지에 접속할 때 호출
    public int subscribe(Long roomId) {
        clientCount.putIfAbsent(roomId, new AtomicInteger(0)); // 방 ID가 없으면 초기화
        return clientCount.get(roomId).incrementAndGet(); // 클라이언트 수 증가 후 반환
    }

    // 구독 해제 메서드: 클라이언트가 예약 페이지를 이탈할 때 호출
    public int unsubscribe(Long roomId) {
        AtomicInteger currentCount = clientCount.get(roomId);
        if (currentCount != null) {
            int newCount = currentCount.decrementAndGet(); // 클라이언트 수 감소

            // 클라이언트 수가 0이 되면 방에서 제거
            if (newCount <= 0) {
                clientCount.remove(roomId);
            }
            return newCount; // 감소된 클라이언트 수 반환
        } else {
            // 방 ID가 존재하지 않으면 0 반환
            return 0;
        }
    }
    
    // 예약 페이지에 접속한 클라이언트의 수를 반환
    public int getCurrentClientCount(Long roomId) {
        AtomicInteger count =  clientCount.get(roomId);
        return (count != null) ? count.get() : 0;
    }
}

여러 사용자가 동시에 예약 페이지에 접속하거나 이탈하는 상황에서 안전하게 클라이언트 수를 관리하기 위해서 동시성 문제를 효과적으로 처리하면서도 다른 자료구조들에 비해 높은 성능을 제공하는 ConcurrentHashMap을 사용했다.

HashMap과 CurrentHashMap 비교

실제 HashMap과 CurrentHashMap이 동시성 처리와 관련해서 실제로 차이가 있는지 궁금해서 테스트 코드를 작성했다.

CurrentHashMap대신 HashMap을 이용해서 클라이언트의 접속 현황을 저장하는 클래스를 만들었다.

@Service
public class ReservationWebSocketConnectionServiceWithHashMap {
    private final Map<Long, Integer> clientCount = new HashMap<>();

    // 구독 메서드: 클라이언트가 예약 페이지에 접속할 때 호출
    public int subscribe(Long roomId) {
        clientCount.putIfAbsent(roomId, 0);
        int newCount = clientCount.get(roomId) + 1;
        clientCount.put(roomId, newCount);
        return newCount; // 클라이언트 수 반환
    }

    // 예약 페이지에 접속한 클라이언트의 수를 반환
    public int getCurrentClientCount(Long roomId) {
        return clientCount.getOrDefault(roomId, 0);
    }
}

이를 바탕으로 테스트 코드를 작성했다.

@SpringBootTest
public class ReservationWebSocketConnectionServiceTest {
    @Autowired
    private ReservationWebSocketConnectionServiceWithHashMap serviceWithHashMap;

    @Autowired
    private ReservationWebSocketConnectionService serviceWithConcurrentHashMap;

    @Test
    public void testHashMap() {
        Long roomId = 1L;

        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                serviceWithHashMap.subscribe(roomId);
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                serviceWithHashMap.subscribe(roomId);
            }
        });

        thread1.start();
        thread2.start();
        
        thread1.join();
        thread2.join();
        
        int actualCount = serviceWithHashMap.getCurrentClientCount(roomId);
        int expectedCount = 20000;
        
        System.out.println("Expected Value: 20000");
        System.out.println("Final client value: " + actualCount);
        
        assertTrue(actualCount < expectedCount);
    }

testHashMap의 각 스레드는 subscribe()를 10000번씩 호출한다.
따라서 동시성 문제가 발생하지 않는다면 각 스레드가 모두 실행 된 뒤 clientCount는 20000이 되어야 한다.

그러나 HashMap은 동시성 처리를 지원하지 않기 때문에 기대한 값에 훨씬 못미치는 결과가 나온 것을 볼 수 있다.

이번에는 ConcurrentHashMap을 사용하는 테스트 코드를 작성했다.

@Test
    public void testConcurrentHashMap() {
        Long roomId = 2L;

        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                serviceWithConcurrentHashMap.subscribe(roomId);
            }
        });

        Thread thread2 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                serviceWithConcurrentHashMap.subscribe(roomId);
            }
        });

        thread1.start();
        thread2.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        int actualCount = serviceWithConcurrentHashMap.getCurrentClientCount(roomId);
        int expectedCount = 20000;

        System.out.println("Expected Value: 20000");
        System.out.println("Final client value: " + actualCount);

        assertEquals(expectedCount, actualCount); // ConcurrentHashMap의 경우, 실제 카운트가 예상과 일치해야 함
    }

ConcurrentHashMap은 버킷 락, CAS, 트리 기반 구조 등을 사용하여 동시성 문제를 효율적으로 관리한다. 따라서 Thread-safe하지 못한 HashMap과는 다르게 기대했던 값과 동일한 테스트 결과를 얻을 수 있다.

++ 추가
HashMap을 사용하더라도 subscribe 메서드에 synchronized 키워드를 사용하면 동시성 문제가 적절하게 처리되었다.

HashMap + synchronized

private final Map<Long, Integer> clientCount = new HashMap<>();

// 구독 메서드: 클라이언트가 예약 페이지에 접속할 때 호출
public synchronized int subscribe(Long roomId) {
    clientCount.putIfAbsent(roomId, 0);
    int newCount = clientCount.get(roomId) + 1;
    clientCount.put(roomId, newCount);
    return newCount; // 클라이언트 수 반환
}

성능 비교

HashMap을 단독으로 사용할 경우 동시성 이슈 발생
HashMap + AtomicInteger의 경우 가장 빠른 실행 시간을 보여줬지만 테스트를 여러번 진행해 보니 동시성 문제가 발생하는 경우가 있었음
ConcurrentHashMap의 성능이 확실히 HashMap + synchronized 보다 뛰어나다는 것을 확인할 수 있었다.

ConcurrentHashMap

ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

ConcurrentHashMap은 16개의 세그먼트로 나뉘어져 각 세그먼트가 자체적인 잠금을 가지고 있는 구조다. 이는 한 세그먼트에 대한 쓰기 작업이 다른 세그먼트의 읽기/쓰기 작업을 방해하지 않음을 의미한다.

예를 들어 스레드 A가 ConcurrentHashMap에 ("apple", 1)을 삽입하고 동시에 스레드 B가 ("banana", 2)를 삽입 하려 할 때, 해시 함수를 통해 계산된 "apple"의 세그먼트 인덱스와 "banana"의 인덱스가 다르다면 스레드 A와 B는 서로 다른 세그먼트에 접근하므로 동시에 작업을 수행할 수 있다.

동시에 스레드 C가 ("cherry", 3)를 삽입하는데 "cherry"의 계산된 세그먼트 인덱스가 "apple"의 인덱스와 같다면, 스레드 C의 작업 수행은 스레드 A의 작업이 완료되고 세그먼트의 잠금이 해제된 이후에 가능하다.

Java8 이후 ConcurrentHashMap의 구현 방식이 변경되었다.

기존의 세그먼트 기반 잠금 방식에서 노드 기반의 접근 방식을 도입했다.
ConcurrentHashMap에서 각 버킷은 Node 객체의 연결 리스트로 구현된다.
각 Node는 키, 값, 해시 그리고 다음 노드에 대한 참조를 포함한다.
특정 버킷의 노드 수가 일정 임계값을 초과하면(전체 HashMap의 크기가 64를 초과한다면) 해당 버킷은 레드-블랙 트리 구조를 사용하는 TreeBin으로 변환된다.

레드-블랙 트리를 사용함에 따라 검색 성능이 O(log n)으로 향상된다.
동시성을 관리하는 매커니즘에도 변화가 생겼다. 세그먼트 기반 잠금 개념을 제거하고 노드 레벨에서 내부적으로 CAS 연산을 사용하여 동시성을 관리한다.

ConcurrentHashMap에서의 CAS(Compare-and-Swap) 연산

ConcurrentHashMap의 1번 버킷에 저장된 노드에 "apple" 이라는 키가 있고 이 키의 현재 값이 10이라고 하자.

각 스레드 A와 B가 이 노드에 동시에 접근하여 서로 다른 값으로 변경하려고 할 때 ConcurrentHashMap은 CAS연산을 이용해서 동시성을 관리한다.

boolean compareAndSet(expectedValue, newValue) {
    if (현재값 == expectedValue) {
        현재값 = newValue;
        return true;
    } else {
        return false;
    }
}

먼저 스레드 A가 CAS 연산을 진행한다고 하자. CAS연산을 수행하기 위해서 전달된 기대값과 노드의 현재 값이 일치하므로 스레드 A는 노드의 값을 20으로 설정하는 데 성공했다.

스레드 A와 스레드 B가 동시에 CAS 연산을 준비할 때, 만약 스레드 A가 먼저 CAS 연산을 성공적으로 수행하면 "apple"의 값이 20으로 업데이트된다. 그럼 스레드 B의 CAS 연산은 어떻게 될까?

스레드 A가 CAS 연산을 성공하였으므로 스레드 B는 CAS 연산에 실패하게 된다. 스레드 B는 그 후 기대값을 새로 설정하여 CAS 연산을 다시 시도하게 된다.

이러한 CAS 연산은 하드웨어 수준에서 지원되므로 매우 효율적이다.

다시 코드로 돌아와서, 벡엔드에서 웹 소켓 엔드포인트를 추가해 주었다.

	private final ReservationWebSocketConnectionService connectionManager;
    
    @MessageMapping("/reservations/{roomId}/subscribe")
    @SendTo("/topic/reservations/{roomId}")
    public CountUpdateMessage subscribeRoom(@DestinationVariable Long roomId) {
        int count = connectionManager.subscribe(roomId);
        System.out.println("count = " + count);
        return new CountUpdateMessage(roomId, count);
    }

    @MessageMapping("/reservations/{roomId}/leave")
    @SendTo("/topic/reservations/{roomId}")
    public CountUpdateMessage leaveRoom(@DestinationVariable Long roomId) {
        int count = connectionManager.unsubscribe(roomId);
        System.out.println("count = " + count);
        return new CountUpdateMessage(roomId, count);
    }
    
    @Data
    @NoArgsConstructor
    public static class CountUpdateMessage {
        private String type;
        private Long roomId;
        private Integer count;

        public CountUpdateMessage(Long roomId, Integer count) {
            this.type = "COUNT_UPDATE";
            this.roomId = roomId;
            this.count = count;
        }
    }

클라이언트에서 /topic/reservations/${roomId} 을 구독하려고 할 때 위 엔드포인트로 메시지를 전송하면, ConcurrentHashMap에 의해 구독한 클라이언트의 개수가 관리된다.

const subscribeToReservationUpdates = () => {
        stompClient.current.subscribe(`/topic/reservations/${roomId}`, (message) => {
            const response = JSON.parse(message.body);
            handleReservationUpdate(response);
        });

        stompClient.current.publish({
            destination: `/app/reservations/${roomId}/subscribe`,
            body: JSON.stringify({ action: 'join' })
        });
        console.log("subscribeToReservationUpdates")
    };

마찬가지로 웹 소켓 연결이 종료될 때도 메시지를 전송한다.

useEffect(() => {
        fetchReservationSchedule();
        fetchFriends();
        connectWebSocket();

        return () => {
            if (stompClient.current && stompClient.current.active) {
                stompClient.current.deactivate();
                stompClient.current.publish({
                    destination: `/app/reservations/${roomId}/leave`,
                    body: JSON.stringify({ action: 'leave' })
                });
            }
        };
    }, []);

결과

이번 기능 구현을 통해 동시성 문제를 적절히 처리하지 않으면 데이터의 일관성이 깨지고 예기치 않은 오류가 발생할 수 있다는 사실을 실감하게 되었다.

여러 사용자가 동시에 예약 페이지에 접속하고 이탈하는 상황에서 클라이언트 수를 안전하게 관리하기 위해서 HashMap과 ConcurrentHashMap의 테스트 코드를 직접 작성해보고 비교해 본 것이 큰 도움이 되었다. ConcurrentHashMap의 내부 동작 원리에 공부한 것도 어려웠지만 도움이 되었던 것 같다.

참고

Deeeep Breath

안녕하세요!

이전 포스트

[unispace] 실시간 예약 시스템 구현(4) - 예약 정보 선점 기능 웹소켓으로 리팩터링

다음 포스트