채팅 서비스에 strategy Pattern 적용

1. 채팅 서비스의 Subscription 정보 관리

채팅 서비스에서 메세지를 주고 받는 과정은 특정 토픽에 대한 발행과 구독이라는 행위로 이루어 진다. 채팅 서비스를 이용하는 유저가 서버에 메세지 전송을 요청하면, 채팅 서버에서는 메세지를 해당하는 토픽에 발행한다. 발행된 메세지는 해당 토픽을 구독하고 있는 모든 유저에게 전송된다. 이러한 송신 → 수신 과정을 Message Broking 이라고 한다. 전송이 요청된 메세지를 올바르게 전달하기 위해서는 어떤 유저가 웹소켓 서버와 연결되어 있고, 또 해당 페이지를 구독하고 있는지를 알고 있어야 하기 때문에 채팅서버에서는 구독 정보를 저장해 두어야 한다.

 

스프링 웹소켓은 자체적인 Message Broker 가 내장되어 있어서 이러한 구독 정보 관리를 관리한다.(UserRegistry). 구독 정보는 실시간으로 빈번하게 변경이 일어나며, 여러 스레드에서 동시에 접근해 오기 때문에, 구독 정보를 관리하는 자료구조는 속도가 빠르고 멀티스레드 환경에서도 안전하다는 조건을 만족해야한다. 내장 Message Broker 는 이를 ConcurrentHashMap 이라는 자료형으로 구현한다.

 

개발자는 이러한 구독정보를 활용하여 메세지를 읽었는지 확인 한다거나, 유저들이 입퇴장 시 서버측에서 메세지를 출력하는 등, 여러 기능을 구현하게 된다. 이 때, 스프링에 내장된 구독 정보을 활용하면 좋겠지만 안타깝게도 UserRegistry 의 자료는 캡슐화 되어있어 개발자가 접근할 수가 없다. 사실 접근할 수 있다고 해도 서버가 확장되어 채팅 서버가 다중화 된다면 구독 정보도 각각 나눠지게 되어 기능이 의도대로 동작 하지 못할 것이다. 이러한 이유로 구독 정보는 자체적인 관리가 필요하다.

 

---

2.  전략 패턴의 적용

1) 

사진에서 보이는 바와 같이 WebSocketController 와 WebSocketChannelInterceptor 에서 각각 subscriptionInfoMap 이란 이름의 ConcurrentHashMap 자료구조를 사용중이다. 사실 코드를 작성하면서도 양쪽으로 정보가 분산되다보니 불편하다는 생각을 하였는데, 이를 하나의 Repository 로 바꾸면 로직도 훨씬 간단해 질 것 같았다.

 

우선 WebSocketController 의 subscriptionInfoMap 에는

• key : Long channelId
• value : Set<Long> userId

가 저장 되어있고,

 

WebSocketChannelInterceptor 의 subscriptionInfoMap 에는

• key : Long sessionId
• value : SubscriptionInfo (Long userId, String subscriptionId, Long channelId)

이 저장되어 있다.

2.  Strategy Pattern 적용

1. SubscribeRepository interface 

public interface SubscribeRepository {

    /* Session 정보를 관리하는 메서드 */
    void saveUserIdBySessionId(Long userId, String sessionId);
    Optional<Long> findUserIdBySessionId(String sessionId);
    void deleteSessionId(String sessionId);

    /* 채널의 구독자를 관리하는 메서드 */
    Set<Long> findSubscribersByChannelId(Long channelId);
    void setSubscriberToChannel(Long userId, Long channelId);
    void deleteSubscriberFromChannel(Long userId, Long channelId);

    /* 세션의 구독 채널을 관리하는 메서드 */
    Set<Long> findSubscribeChannelIdBySessionId(String sessionId);

    /* 세션의 구독 id 를 관리하는 메서드 */
    void saveSubscriptionIdBySessionId(String subscriptionId, String sessionId);
    void deleteSubscriptionIdBySessionId(String subscriptionId, String sessionId);
    Set<String> findSubscriptionIdBySessionId(String sessionId);
    void deleteAllSubscriptionsBySessionId(String sessionId);

    /* 구독 id 의 채널 정보를 관리하는 메서드 */
    Optional<Long> findChannelIdBySubscriptionId(String subscriptionId);
    void setSubscriptionIdToChannel(String subscriptionId, Long channelId);
    void deleteSubscriptionId(String subscriptionId);

}

 

2. 구현체 InMemorySubscribeRepository

@Repository
@ConditionalOnProperty(name = "redis-config.websocket", havingValue = "false")
@RequiredArgsConstructor
public class InMemorySubscribeRepository implements SubscribeRepository {

    /* sessionId 에 UserId 저장합니다. */
    private final ConcurrentHashMap<String, Long> sessionIdUserIdMap = new ConcurrentHashMap<>();

    /* 채널을 구독하는 userId 를 저장합니다. */
    private final ConcurrentHashMap<Long, Set<Long>> channelSubscribersMap = new ConcurrentHashMap<>();

    /* subscriptionId 가 어떤 channel 을 가르키는지 저장합니다. */
    private final ConcurrentHashMap<String, Long> subscriptionIdChannelIdMap = new ConcurrentHashMap<>();

    /* session 에 SubscriptionId 를 저장합니다. */
    private final ConcurrentHashMap<String, Set<String>> sessionIdSubscriptionIdMap = new ConcurrentHashMap<>();

    @Override
    public void saveUserIdBySessionId(Long userId, String sessionId) {
        sessionIdUserIdMap.put(sessionId, userId);
    }

    @Override
    public Optional<Long> findUserIdBySessionId(String sessionId) {
        return Optional.ofNullable(sessionIdUserIdMap.get(sessionId));
    }

    @Override
    public void deleteSessionId(String sessionId) {
        sessionIdUserIdMap.remove(sessionId);
    }

    @Override
    public void saveSubscriptionIdBySessionId(String subscriptionId, String sessionId) {
        sessionIdSubscriptionIdMap.computeIfAbsent(sessionId, key -> new HashSet<>()).add(subscriptionId);
    }

    @Override
    public Set<String> findSubscriptionIdBySessionId(String sessionId) {
        return sessionIdSubscriptionIdMap.get(sessionId);
    }

    @Override
    public Set<Long> findSubscribeChannelIdBySessionId(String sessionId) {
        return sessionIdSubscriptionIdMap.get(sessionId)
                .stream().map(subscriptionIdChannelIdMap::get).collect(Collectors.toSet());
    }

    @Override
    public void deleteAllSubscriptionsBySessionId(String sessionId) {
        sessionIdSubscriptionIdMap.remove(sessionId);
    }


    @Override
    public Set<Long> findSubscribersByChannelId(Long channelId) {
        return channelSubscribersMap.getOrDefault(channelId, new HashSet<>());
    }

    @Override
    public Optional<Long> findChannelIdBySubscriptionId(String subscriptionId) {
        return Optional.ofNullable(subscriptionIdChannelIdMap.get(subscriptionId));
    }

    @Override
    public void setSubscriberToChannel(Long userId, Long channelId) {
        Set<Long> usersId = channelSubscribersMap.get(channelId);
        if (usersId == null) {
            channelSubscribersMap.put(channelId, new HashSet<>(Collections.singletonList(userId)));
        }
        else usersId.add(userId);
    }

    @Override
    public void setSubscriptionIdToChannel(String subscriptionId, Long channelId) {
        subscriptionIdChannelIdMap.put(subscriptionId,channelId);
    }

    @Override
    public void deleteSubscriberFromChannel(Long userId, Long channelId) {
        Set<Long> subscribersId = channelSubscribersMap.get(channelId);
        if (subscribersId == null) return;
        else subscribersId.remove(userId);
        if (subscribersId.isEmpty()) channelSubscribersMap.remove(channelId);
    }

    @Override
    public void deleteSubscriptionId(String subscriptionId) {
        subscriptionIdChannelIdMap.remove(subscriptionId);
    }

    @Override
    public void deleteSubscriptionIdBySessionId(String subscriptionId, String sessionId) {
        sessionIdSubscriptionIdMap.computeIfPresent(sessionId, (k, m) -> {
            m.remove(subscriptionId);
            return m;
        });
    }
}

 

위와 같이 전략 패턴에 따라서 SubscribeRepository 인터페이스를 하나 만들고 이를 구현한 InMemorySubscribeRepository 를 작성하였다. 분산 되어있던 데이터인 ConcurrentHashMap 은 이 구현체의 멤버변수로 선언 되어있다. 전략 패턴을 적용하면서 구독 관리를 위해 필요한 메서드들을 더 많이 식별되어서 자료구조가 더 많아 졌다. 사실 구현체가 많이 복잡해보이긴 한다.

 

하지만 자료구조체의 수가 더 늘어났으니 더 비효율적인 것같고 더 복잡해 보이지만, 구현체의 변수가 늘어났든 로직이 복잡해졌든 전혀 중요하지않다 중요한 것은 구현체가 아닌 interface 의 설계이다. 

1. interface 가 구현체에 의존하지 않고 (구현체랑은 상관이 없어야 한다)
2. 메서드들이 합리적으로 잘 짜여 있다면

이를 어떻게 구현했는지는 완전히 부차적인 문제가 되기 때문이다. 

 

---

이로써 추후 캐싱과 동시성 이슈에 최적화 된 DB 를 도입한다면 기존 코드를 변경하지 않고 이식할 수 있게 되었다. 이는 객체 지향 프로그래밍 의 다섯가지 원칙(SOLID) 중에서도 개방-폐쇄 원칙과 부합한다고 할 수 있다. 

 

지난 포스팅에 이어서 또 '진작 이렇게 했으면...' 하는 후회와 함께 새로운 디자인 패턴을 체득할 수 있는 기회가 되었다. 

 

Redis 도입하여 구독 정보 Repository 에 적용했다. 다음글에서 다루었다. 

다음글 - [ https://techforme.tistory.com/26 ]

 

---

이전에 어댑터 패턴이라고 인식했었는데, 잘못 읽은 것이었다 ^^;; 어댑터 패턴은 인터페이스에 딱 들어맞지 않는 클래스를 인터페이스에 맞추기 위해 콘센트 어댑터 처럼 중간 다리 역할을 해주는 어댑터 클래스를 구현한 것으로 다음 기회에 적용해보아야겠다. 

 

이 역시 토비의 스프링을 읽고 난 후 다시보니 추상화 라는 키워드가 머리에 떠오른다. 다시 봐도 참 만족스럽다. 이후 레디스를 이식할 때에도 큰 고민 없이 Interface 의 구현에만 초점을 맞춰 개발을 진행할 수 있었다.