[개발] Redis를 사용한 데이터 캐싱으로 조회 API성능 향상시키기

캐싱

캐싱은 데이터를 적당한 장소에 저장하여 다음에 같은 데이터 조회 요청이 들어왔을 때 빠르게 데이터를 획득할 수 있도록 한다. API를 호출하면 보통 데이터베이스를 조회하고 어떠한 연산을 통해 결과값을 반환한다. 하지만 트래픽이 늘어나거나 해당 연산이 오래걸릴 경우 서비스가 느려질 수 있다. 그럴 경우 캐싱을 통해 미리 저장해두고 빠르게 반환하는 방법을 생각할 수 있다.

 

레디스

캐싱을 사용할때 레디스가 많이 언급된다. 레디스는 key-value를 저장하는 In-memory DB이며 다음과 같은 특징이 있다.

1. 싱글스레드만 접근 가능하다.

2. string외에 다양한 데이터 타입을 지원한다.

3. 인메모리 데이터베이스이면서 디스크에 저장까지 한다.

4. 속도가 굉장히 빠르다.

 

redis는 보통 같은 제품인 memcached와 비교가 많이 되는데 4번을 제외하면 memcached는 지원하지 않는 특징이므로 개인적으로 redis의 승이라고 생각한다.

 

데이터 준비

그럼 이제 본격적으로 레디스를 사용하여 데이터 캐싱을 진행하고 성능을 분석해보자. 

CREATE TABLE test (
    id int not null auto_increment, 
    name varchar(20) not null, 
    birth_month int not null, 
    created_at datetime default current_timestamp, 
    constraint primary key(id)
);

위와 같은 테이블을 준비한다. 그 후 데이터를 약 20만개 삽입했다. (알고리즘 문제 풀 때 효율성을 판단하는 기준과 비슷하다.)

import pymysql
import random

conn = pymysql.connect(host='localhost', user='', password='', db='', charset='utf8')

cur = conn.cursor()

for i in range(200000):
    ix = random.randint(1, 10)
    month = random.randint(1, 12)
    cur.execute(f"INSERT INTO test(name, birth_month) VALUES('user{ix}', {month})")
conn.commit()

 

API 준비

특정 월의 birth_month를 가진 사용자의 수를 불러오는 API를 만든다고 하자.

from fastapi import FastAPI
import pymysql

conn = pymysql.connect(host='localhost', user='', password='', db='', charset='utf8')

app = FastAPI()

@app.get("/")
async def get(month: int):
    cur = conn.cursor()
    cur.execute(f"SELECT COUNT(*) FROM test WHERE birth_month = {month}")
    res = cur.fetchone()[0]
    return res

 

테스트

해당 API가 얼마나 걸리는지 테스트해보자.

크롬으로 10월에 해당하는 API를 직접 호출했는데 약 30ms가 나왔다. 그렇다면 여기에서 레디스를 추가하면 어떨까?

 

레디스 추가

레디스 설치는 이 곳에서 진행한다.

from fastapi import FastAPI
import redis
import pymysql

conn = pymysql.connect(host='localhost', user='', password='', db='', charset='utf8')
rd = redis.Redis(host='localhost', port=6379, db=0)

app = FastAPI()

@app.get("/")
async def get(month: int):
    # 레디스에 캐싱된 데이터가 있는지 확인
    cache_data = rd.get(f"month_{month}")
    # 있다면 해당 데이터 반환
    if cache_data:
        return int(cache_data)

    cur = conn.cursor()
    cur.execute(f"SELECT COUNT(*) FROM test WHERE birth_month = {month}")
    res = cur.fetchone()[0]

    # 데이터 캐싱
    rd.set(f"month_{month}", res)
    return res

레디스로 캐싱하는 코드로 개선했는데 거의 dict 쓰는 것처럼 사용이 굉장히 간편하다. 데이터가 없으면 key_error가 아닌 None을 반환한다는 점도 눈에 띈다.

 

속도 변화 테스트

캐싱 전

캐싱 후

캐싱 전에 28.46ms였는데 캐싱 후에 8.83ms로 약 3~4배의 성능 향상이 있었다. 이는 모든 HTTP통신까지 포함한 시간이며 순수 서버의 연산 시간만 보면 캐싱 전에 24.32ms, 캐싱 후에 0.5ms로 50배에 가까운 성능 차이를 보여준다.

 

레디스를 사용하면 대규모 트래픽에도 지연이 없는 사용자 경험을 제공할 수 있을 것 같다. 다만 인메모리 데이터베이스이기 때문에 메모리가 초과하지 않도록 관리는 잘 해야 한다.