구현한 채팅에 캐시 적용해서 성능 개선하기

승톨·2021년 4월 10일

Cache socket

Update

글 마지막에 적어놓은 아이디어를 구현해 현재 채팅 기능의 구조는 아래의 사항이 적용되었습니다.

방에 처음 진입 시, 최근 N개의 메세지를 DB 혹은 캐시 조회를 통해 보여준다.(이전과 동일)
클라이언트가 메세지 생성 후 서버로 전송하면, 서버는 받은 메세지를 channel layer를 통해 해당 그룹에 broadcast만 한다.
- (해당 시점에 DB/캐시에 메세지 오브젝트 만드는 방식은 유지)

Intro

현재 실시간 Q&A를 할 수 있는 서비스를 만들면서 라이브 방에서 채팅 하는 기능 구현을 진행했습니다.
django의 Channels라는 라이브러리를 이용해서 만들었고 소켓 서버는 daphne로 세팅했습니다.

기본적인 기능을 구현하는 법은 튜토리얼에 자세하게 나와있기 때문에 궁금하신 분들은 튜토리얼을 보시면 좋을 것 같네요.
지난번 글에서 발생한 삽질을 해결했지만, 프로덕션 서버에서 유저 테스트를 하다가 또 다른 문제를 맞닥뜨리게 되었습니다.

문제 발생

유저 몇 십 명이 하나의 라이브 방에 들어와서 채팅을 하고 있는 상황이었습니다.
몇 명의 유저들이 장난의 목적으로 몇 십자의 텍스트를 연속적으로 계속 타이핑했고,
그러자 (정확히 잴 순 없었지만 몇 분 뒤) 점점 채팅을 읽어오는 속도가 느려지기 시작하면서, 채팅 창에 글자를 입력하면 몇 초 뒤에 내 글이 올라오는 현상이 발생했습니다.
즉, 메세지 싱크 오류 현상이 발생했습니다.

현재 아키텍쳐

문제가 발생했기 때문에, 현재 채팅 기능 아키텍쳐를 점검해보았습니다.

소켓 서버는 3개, Nginx로 로드밸런싱 중.
사용자가 방 진입 후 소켓을 통해 connect 요청을 하면, 서버에서는 라이브 방 조회 후 방에 해당하는 channel layer group에 해당 사용자의 channel을 추가합니다.
- 메세지를 전달 받을 때는 해당 channel layer group에 들어있는 channel들에게 브로드 캐스팅 됩니다.
추가 후 서버는 사용자에게 보여줄 최근 N개의 메세지를 조회해서 브라우저에 전달 & 요청에 대한 accept 처리를 합니다.

사용자가 메세지를 입력하면 send 요청이 서버에 오게 되고, 서버는 DB를 호출해 메세지 레코드를 생성하고, 생성한 메세지를 포함해서 최근 N개의 메세지를 serialize해서 다시 보내줍니다.
보낸 메세지는 해당 사용자(channel)가 속한 그룹에 브로드 캐스팅 됩니다.

즉, 메세지 생성은 DB 레코드 삽입
메세지 조회는 DB 레코드들 조회

로 해결하는 구조를 가져갔습니다.

이런 구조이다보니 짧은 시간내에 다량의 메세지를 날리면 디스크 I/O가 느리기 때문에 충분히 싱크 오류가 날 수 있는 상황이라는 생각이 들었습니다.

아이디어

어떻게 하면 DB I/O 타임을 줄여볼 수 있을까?라는 과제가 생겼고,

"메모리는 디스크보다 데이터를 읽는 속도가 빠를 것이기 때문에 메모리를 캐시 메모리 형태로 사용할 수 있지않을까?"라는 아이디어를 떠올렸습니다.

데이터를 쓰는 것도 메모리에 해두면, 데이터 조회 시에 최근 메세지 동기화도 될 것이라는 생각에

현재 채널명을 담고 있는 Redis를 메세지 캐싱하는 용도로 사용해보기로 했고,

현재 구조를 유지한 채로 성능 개선이 될 수 있지않을까?라는 기대를 품었습니다.

실험

그 결과 , 아래의 가설을 만들었고 가설을 검증해보기 위해 캐시 구현에 들어갔습니다.

가설

Redis에 메세지, 라이브 방 id 등을 캐싱해두면 디스크 I/O Time이 줄어들어 성능 개선 효과가 있을 것이다.

구현

Redis에 1번 데이터베이스를 추가해 메세지 캐싱 용도로 사용.
Sorted Set 자료구조를 사용해서 메세지 오브젝트 리스트를 구현
- Key = message_{특정 room name}
- score = 메세지를 보낸 시간
- Value = 클라이언트에서 json 형태로 전달해주는 메세지 데이터

로직

클라이언트가 메세지 조회 요청을 할 때, 이미 캐시에 Key가 담겨있으면, 바로 읽어서 클라이언트에 전달할 수 있음.
캐시에 Key가 없는 경우 비동기로 메세지 조회해오기

def get_serialized_messages(self, room, con):
        # 캐시에 있는 메세지를 가지고오기
        group_key = "message:%s" % room
        res = con.exists(group_key)
        if res == 1:
            messages = con.zrange(group_key, 0,-1) # 현재 group에 있는 메세지 전부 가져오기
            message_array = []
            for m in messages: # bytes list로 리턴된 걸 디코드하기
                m = m.decode('utf-8')
                m = literal_eval(m) # string dict를 dict로 만들기

                message_array.append(m)         
            # dict가 담긴 array를 json array 형태로 저장.
            return json.dumps(message_array)
        else:
            # db에서 메세지를 가지고오기
            messages = reversed(Message.objects.filter(livehole=room).order_by('-id')[:20])
            serializer = MessageSerializer(messages, many=True)
            return json.dumps(serializer.data)

메세지 생성 요청을 할 때에는 DB와 캐시에 모두 데이터를 생성.

캐시에도 특정 N개의 메세지를 유지하기 위해 캐시 capacity가 꽉 차면 메세지 교체를 진행
DB에 데이터 쓰는 건 비동기로 진행.
DB 레코드 생성 시 자주 읽어와서 써야하는 데이터(라이브 방 id)는 캐싱을 통해 조회 속도 줄이기

def create_room_message(self, text, username, room,author_group, con):
        res = con.hexists(author_group, username) # 캐시에 author들이 있는지 확인 : 있으면 캐시히트
        if res == 1:
            author_id= con.hget(author_group, username).decode("utf-8") # 바로 불러옴. 대신 bytes로 리턴이라 디코딩필요
        else:
            author_id = User.objects.get(username=username).id # 메세지 record를 만들기 위해 id로 가지고오기
            con.hset(author_group,username,author_id) # 처음 이후부터는 바로 캐시에서 불러오기 위해서 캐시에 추가

        livehole_group = "group:livehole"
        res_livehole = con.hexists(livehole_group, room) # 캐시 livehole group에 해당 livehole num이 있는지 확인
        if res_livehole == 1:
            livehole = con.hget(livehole_group,room).decode("utf-8")
            # 메세지 오브젝트를 캐시에 넣기 위해 만들기
        else:
            livehole = LiveHole.objects.get(id=room).id
            # 라이브 홀의 경우 id를 바로 알고 있으니까 캐시에 추가할 필요는 없음.
        
        group_key = "message:%s" % livehole
        now_time = datetime.now()

        # 캐시에 저장할 메세지 오브젝트 만들기
        message = {
            "sender" : username,
            "text" : text,
            "livehole" : livehole,
            "sent_timestamp" : now_time
        }
 
        # DB에 넣기 위해 sender를 id로 잠시 변경
        message['sender'] = int(author_id)
        message_serializer = MessageSerializer(data = message)
        if message_serializer.is_valid():
            message_serializer.save()
        # 새로운 메세지를 sorted set에 넣기
        message['id'] = message_serializer.data['id']
        message['sent_timestamp'] = str(now_time)
        message['sender'] = username
        dict = {}
        dict[str(message)] = time.time()
        con.zadd(group_key, dict)
        cache.expire(group_key, timeout=500)

        # 메세지 capacity 확인 후 eviction 및 캐시에 넣기
        group_messages_count = con.zcard(group_key)
        if group_messages_count >= 20:
            ress = con.zpopmin(group_key)
        return message

아키텍쳐

캐시를 구현 한 이후의 아키텍쳐는 다음과 같습니다.

테스트

구현 후 처음에 진행했던 테스트와 비슷한 환경의 유저 테스트를 다시 진행해보았습니다.
(정확한 수치는 아니지만) latency가 감소한 것 같다는 피드백을 얻었습니다.

그러나, 위의 피드백은 어디까지나 직감이었기 때문에 간단한 load test라도 진행해서 데이터를 확인해보고 싶었습니다.

따라서 캐시 적용 전 vs 적용 후를 비교하고자 Locust를 이용해 웹 소켓 요청 → 메세지 발신, 수신 을 하는 테스트를 진행했습니다.

환경

유저 20명이 동시접속했다고 가정
평균 메세지 전송 수 : 1600건
메세지 수신 수 : 32000건

결과

응답시간은 거의 동일했습니다.
CPU 점유율과 Disk IO/sec의 경우 캐시를 적용한 버전이 뚜렷하게 감소한 것을 볼 수 있었습니다.
(핑크색이 DISK IO, 파란색이 CPU 점유% 입니다.)

캐시 적용 전

캐시 적용 후

한계점

혼자 리서치를 통해 진행한 구현이라 정확한 기준에 근거한 캐시 구현, 성능 측정이 아닐 수 있다고 생각합니다.
또한 구현을 하고나니

메세지를 보낼때마다 DB에 레코드를 생성하고 DB를 조회해서 메세지를 보여주는 방식보다,
방에 처음 진입할 때, 최근 N개의 메세지를 데이터 조회를 통해 보여주고, 메세지 생성 시에는 클라이언트에서 보관하고 있다가, 방이 완전히 종료되면 전체 메세지를 DB에 저장하는 방식이 낫지않을까?
라는 아이디어가 떠올랐습니다.
이 경우, N개의 메세지를 보여줄 때 캐시를 활용할 수 있을 것 같습니다.