Redis와 클라이언트 캐시 간 데이터 동기화 기술

Redis와 클라이언트 캐시 간 데이터 동기화 기술 - Redis Client Caching 살펴보기

mihosang 25.03.20

2,534 9 5

facebook twitter kakao link

9 5

facebook twitter kakao link

DEVOTEE 요약: Redis는 데이터베이스 캐시로 자주 사용되며, 클라이언트 캐시를 통해 효율성을 높일 수 있습니다. 그러나 클라이언트 캐시는 항상 최신 데이터를 보장하지 않으므로, Redis와 클라이언트 캐시 간의 데이터 동기화를 위해 Redis Client Caching 기능을 사용할 수 있습니다. 이 기능을 통해 클라이언트는 관심 있는 key가 변경되면 Invalidation 메시지를 받아 캐시를 갱신할 수 있습니다.

DEVOTEE 추천 블로그

Overview

Redis는 In memory 기반의 데이터베이스로 빠른 읽기/쓰기 성능과 다양한 자료구조로 데이터베이스 캐시로 많이 사용되고 있습니다.

만약 Redis 에 저장한 데이터가 빈번하게 변경되지 않지만 payload 사이즈가 커서 매번 Redis에서 값을 가져오는 것이 비효율적인 경우,

클라이언트 캐시를 사용하기도 합니다. 하지만 클라이언트 캐시가 항상 최신 데이터를 가지고 있지 않을 수 있습니다.

이런 경우 클라이언트 캐시와 Redis 의 데이터를 동기화 하는 것이 필요합니다.

쉬운 방법으로는 Local cache의 Expiration을 설정하여 일정 시간이 지나면 Redis의 값을 읽어서 동기화하는 방법도 있습니다.

이 방법은 캐시의 데이터가 즉시 동기화되지 않아도 무방한 경우 단순하고 효과적인 방법이기도 합니다.

하지만 Redis의 데이터가 변경되는 것을 즉시 Notification을 받아서 Local Cache를 Redis와 동기화할 수 있으면 어떨까요?

이 글에서는 Redis Client Caching 을 사용하여 클라이언트 캐시와 Redis 의 데이터를 동기화 하는 방법을 설명합니다.

필요 환경

Redis Server 버전 6.0+
지원하는 Client library 및 Version
Client
Version
redis-py
v5.1.0
Jedis
v5.2.0

Client	Version
redis-py	v5.1.0
Jedis	v5.2.0

Redis Client Caching 동작 방식

Redis Client Caching 은 Redis 의 데이터를 클라이언트 캐시에 저장하고 동기화 하는 기법입니다.

먼저 Local Cache를 사용하는 기본적인 방법부터 알아보고 이후 Redis Client Caching 을 사용하는 방법을 알아보겠습니다.

Redis에서 매번 값을 읽으면 불필요한 Network I/O가 발생하고 Redis의 자원을 비효율적으로 사용하게 되거나 서비스의 Latency를 증가시키게 됩니다.

이런 문제를 해결하기 위해서는 클라이언트 캐시를 사용하는 것이 필요합니다.

Redis Client Caching

위그림은 Local Cache의 기본적인 동작을 설명하고 있습니다.

여기서는 하나의 클라이언트가 Redis에서 값을 가져와서 Local cache를 참조하도록 설계되었는데,

만약 다른 클라이언트에서 foo 값을 변경하면 Local cache의 값과 Redis의 값이 다르게 됩니다.

이런 경우 클라이언트는 다시 Redis에서 값을 가져와서 동기화해야 합니다.

Redis Client

클라이언트A가 값을 변경하면 Redis는 해당 값의 Invalidation 여부를 클라이언트B에게 알립니다.

클라이언트B는 이 메시지를 받으면 Redis에서 값을 읽어 Local cache를 갱신합니다. 이런 방식으로 클라이언트 캐시와 Redis의 데이터를 동기화 할 수 있습니다.

이렇게 Invalidation 메시지를 받아서 동기화하는 방식이 Redis Client Caching 입니다.

Redis Client Caching 모드

Redis Client Caching 은 크게 두 가지 모드로 동작합니다.

1. Default 모드

이 모드에서 Redis 서버는 클라이언트A가 관심있는 key가 어떤 것인지 기억합니다.

클라이언트A가 관심있는 key가 변경되면 Redis 서버는 클라이언트A에게 Invalidation 메시지를 보냅니다.

이 모드는 클라이언트가 관심있는 key가 적은 경우 유용하지만, 서버의 메모리를 사용한다는 점을 유념해야 합니다.

2. Broadcast 모드

이 모드에서는 클라이언트가 어떤 key를 관심있는지 명시적으로 알려야 합니다.

이 모드는 서버의 메모리를 사용하지 않지만, 모든 key의 변화를 받을 수는 없기 때문에 Prefix 를 사용하여 관심있는 key를 지정하는게 효과적입니다.

Redis Client Caching 예제

Redis Client Caching 을 사용하는 예제를 살펴보겠습니다. 이 예제는 redis-py 를 사용하여 구현되었습니다.

먼저 Redis를 기반으로 하는 Async Cache를 생성하고 이를 사용하는 예제를 살펴보겠습니다.

import asyncio

from redis.asyncio import Redis


class ClientSideCache:
    def __init__(self):
        self.client = None
        self.localcache = {}

    async def __aenter__(self):
        await self.connect()
        return self
    
    async def connect(self):
        self.client = Redis(host="localhost", port=6379)

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.aclose()

    async def set(self, key, value):
        await self.client.set(key, value)
        self.localcache[key] = value

    async def get(self, key):
        if key in self.localcache:
            value = self.localcache[key]
            print(f"Local Cache Hit: {key} = {value}")
            return value
        else:
            value = await self.client.get(key)
            print(f"Get value from Redis: {key} = {value}")
            self.localcache[key] = value.decode("utf-8")
            return value.decode("utf-8")


async def main():
    async with ClientSideCache() as client:
        for i in range(10):
            await asyncio.sleep(1)
            print(await client.get("user:1"))


if __name__ == "__main__":
    asyncio.run(main())

이 간단한 코드에서는 Local cache를 사용하여 Redis의 값을 cache 하였지만 Expire를 따로 구현해두지 않았기 때문에, Redis의 값이 바뀌더라도 계속 캐시된 값을 반환합니다.

다음은 Redis Client Caching 을 사용하여 캐시를 동기화하는 코드입니다.

import asyncio

from redis.asyncio import Redis


class ClientSideCache:
    def __init__(self):
        self.client = None
        self.localcache = {}

    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.aclose()

    async def connect(self):
        self.client = await Redis(host="localhost", port=6379)

    async def set(self, key, value):
        await self.client.set(key, value)
        self.localcache[key] = value

    async def get(self, key):
        if key in self.localcache:
            value = self.localcache[key]
            print(f"Local Cache Hit: {key} = {value}")
            return value
        else:
            value = await self.client.get(key)
            print(f"Get value from Redis: {key} = {value}")
            self.localcache[key] = value.decode("utf-8")
            return value.decode("utf-8")

    async def _listen_invalidate(self):
        async with self.client.pubsub() as pubsub:
            try:
                #1. Get client id
                await pubsub.execute_command(b"CLIENT", b"ID")
                client_id = await pubsub.connection.read_response()
                print(f"#1 Read client_id: {client_id}")

                #2. Subscribe to client tracking
                await pubsub.execute_command(f"CLIENT TRACKING on REDIRECT {client_id} BCAST PREFIX user:")
                await pubsub.connection.read_response()

                #3. Subscribe to invalidate channel
                await pubsub.subscribe("__redis__:invalidate")
                
                #4. Listen for invalidate messages
                while True:
                    message = await pubsub.get_message(ignore_subscribe_messages=True)
                    if message:
                        key = message["data"][0].decode("utf-8")
                        del self.localcache[key]
            except Exception as e:
                print(f"Error in _listen_invalidate: {e}")
                raise

    async def manage_invalidation_task(self):
        while True:
            try:
                await self._listen_invalidate()
            except Exception as e:
                print(f"Restarting _listen_invalidate due to error: {e}")
                await asyncio.sleep(1)


async def main():
    async with ClientSideCache() as client:
        task = asyncio.create_task(client.manage_invalidation_task())
        for i in range(5):
            await asyncio.sleep(1)
            await client.get("user:1")
        task.cancel()


if __name__ == "__main__":
    asyncio.run(main())

위 코드에서는 Broadcast 모드를 다루며 다음과 같은 step을 따라 Client Side Caching을 적용합니다.

Redis Client 연결
Client ID 조회 : CLIENT ID 명령어를 사용하여 클라이언트 ID를 조회합니다.
Client Tracking 설정 : CLIENT TRACKING on REDIRECT {client_id} BCAST PREFIX user: 명령어를 사용하여 클라이언트 ID를 Tracking 하도록 알립니다.
Invalidate 메시지 수신 : __redis__:invalidate 채널을 구독하여 Invalidate 메시지를 수신합니다.
Local Cache 동기화 : Invalidate 메시지를 수신하면 Local Cache를 동기화합니다.

위 코드에서는 클라이언트가 관심있는 key가 변경되면 Redis 서버는 클라이언트에게 Invalidation 메시지를 보냅니다.

이 메시지를 받으면 클라이언트는 Local Cache를 갱신합니다. 이런 방식으로 클라이언트 캐시와 Redis의 데이터를 동기화 할 수 있습니다.