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

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 점유% 입니다.)

캐시 적용 전

캐시 적용 후

한계점

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

2. 또한 구현을 하고나니

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

마치며

첫 채팅 기능 구현이라 미흡한 점이 많을 것이라, 더 효율적인 로직을 연구해 디벨롭 해 볼 예정입니다.

(이번 구현 과정에는 문제 발견 → 가설 검증을 해봤다는 점에 큰 의의를 두고 있습니다.)

채팅 기능, 캐시 구현 혹은 성능 측정에 대한 피드백이 있다거나 건강한 토론을 하고 싶은 분이 계시면 댓글 부탁드리겠습니다!