Redis Pub/Sub을 활용해보자!

레디스에 다양한 자료구조가 등장하는데, 이때 Redis Pub/Sub에 대한 것이 존재한다.

 

Publish/Subscibe 란?

특정한 주제에 대해서 해당 topic을 구독한 모두에게 메세지를 발행하는 통신 방법으로 채널을 구독한 통신자 모두에게 메세지를 전송하는 것을 의미한다. 

 즉, 특정 topic 이라는 Queue에 메세지를 쏘고 있다면, 해당 메세지를 구독하고 있는 객체들에게 메세지가 전달 되는 것을 의미한다.

 

출처 : https://dashbird.io/knowledge-base/well-architected/pub-sub-messaging/

 

 

Redis의 pub/sub

Redis에서 제공하는 pub/sub의 중요한 포인트는 "속도"이다. In-memory 기반의 DB이기 때문에 빠를 수 밖에 없다. 그러나, 안정성 면에서는 별로 좋지는 않다.

 

네트워크 환경의 오류가 판치는 세상에서 전송 보장 및 수신 보장이 되지 않는다면 안정성 있는 환경을 구축하기 어려울 수 있다. 하지만 프로젝트상 꼭 사용해야 하기 때문에 일단 쓴다...

 

 

 

FastAPI와 expire pub/sub

우리의 프로젝트는 RedisSet를 설정하고, 해당 Set이 만료될때 해당 Set의 주인에게 FCM 요청을 보내줘야 하는 로직으로 진행이 된다.

 

이를 구현하기 위해서 RedisSet의 만료를 확인하는 subscribe가 필요하고, 만료된 것들을 모아주는 topic도 필요하다.

 

 

- redis.config 설정 (아래를 추가해줘야 한다)

notify-keyspace-events Ex

※ 참고

뒤에 어떤 알파벳을 사용하느냐에 따라서 효과가 달라지게 된다.

Ex 에서 E는 keyevent events 로, 이벤트가 일어나는 것에 대해서 감지하며, 그 감지하는 요소로는 x (만료 이벤트)로 사용하겠다는 것이다.

 

$ - 스트링 명령

l - 리스트 명령

s - 셋 명령

h - 해시 명령

z - 소트 셋 명령

x - 만료 이벤트

e - 퇴출 이벤트

A - 모든 이벤트

 

- redis.py 의 코드

class MessageSet:
    def __init__(self, name):
        self.name = name
        self.redis = redis.Redis(host=os.getenv("REDIS_ENDPOINT"), port=6379, db=0)
        
    def expire(self):
        return self.redis.expire(self.name, 300)

해당 Set이 형성되게 되는데, expire을 하게 함으로서 300초 뒤에 만료되도록 코드를 작성하였다.

 

- main.py 의 코드

def expireCallBack(message):
	userId = message["data"].decode("utf-8")[8:]
	sendingCancelMessageToUser(userId)
	

async def listenExpireEvents():
	redis = Redis.EventListenerRedis()
	redisSubscriber = redis.redis.pubsub()
	redisSubscriber.psubscribe(**{"__keyevent@0__:*": expireCallBack})
	
	for message in redisSubscriber.listen():
		pass

def start_listening():
    asyncio.run(listenExpireEvents())

즉, 시작할때 비동기로 listenExpireEvents 를 진행한다. 하나의 쓰레드에서 EventListenerRedis를 구현하고, 

위에서 E 에 해당하는 것으로 keyevent가 일어나는 토픽을 확인하고 싶다는 것이고, 만일 토픽에서 어떤 일이 일어날 때마다

expireCallBack이라는 함수를 실행시키는 것이다. 해당 함수에는 보내진 message가 오게 되고 , key가 전송되게 된다.

 

나는 Set 이름을 "Matching + userId" 형태로 저장했기 때문에 8번 인덱스부터 슬라이싱 한 것을 보내도록 했다.

 

 

- 결과

저장해놓은 것이 사라져서 구글링해서 가져옴...ㅠ

ttl 은 redis의 만료 시간을 알려주는 것인데, 이렇게 만료된 것은 -1이 뜨게 된다.

 

 

또한 python에서도 실행된 FCM 토큰이 잘 날아가는 것도 확인할 수 있었다! (프론트 쪽에서 잘 들어간 것을 확인할 수 있었음)