[Redis] Redis 뜯어보기 3 - Redis Pub/Sub 동작 방식

https://ai-back-end.tistory.com/176

[Redis] Redis 뜯어보기 1 - 내부 구조 및 동작 방식(feat. I/O-Multiplexing)

Redis 프로세스가 어떻게 명령어를 받고 처리하는지에 대해서 알아봤었는데 이번에는 Redis하면 빠질 수 없는 Pub/Sub에 대해서 알아보려고한다.

https://www.youtube.com/watch?v=SF7eqlL0mjw&ab_channel=%ED%86%A0%EC%8A%A4

 

"토스 | SLASH 23 - 실시간 시세 데이터 안전하고 빠르게 처리하기"에서 Pub/Sub을 왜? 어디에? 사용햇는지 사례를 통해서 자세히 뜯어보고자 한다.

---

토스 증권에서 Redis Pub/Sub을 왜 사용 사례

토스 증권 시세 플랫폼 아키텍쳐

아시다시피, 주식시장은 아침부터 저녁까지 활발하게 움직인다.

특히 장이 시작되는 오전 9시에는 거래량이 크게 증가하기 때문에, 이러한 피크 시간에도 데이터 처리가 지연되거나 누락되지 않도록 주의해야 한다. 주식은 매 순간 주가가 오르내리기 때문이다.

투자자들의 혼란을 방지하기 위해 해당 시세데이터를 실시간으로 빠르고 안정적으로 처리하기 위해 위 시세 플랫폼에서 처리 중이라고 한다. 이때 중요한 점이 1. 낮은 지연시간 & 2. 빠른 장애복구.

위 이미지가 바로 시세 플랫폼에서 처리하고 있는 대략적 플로우라고 한다.

• 수신부는 거래소가 제공하는 시세 데이터를 UDP 멀티캐스트 그룹에 접속해서 읽어오는 일을 한다.
• 수신부는 처리부에게 데이터를 전송할 때 수신 시각을 Header에 포함하여 처리부에서 총 처리 시간을 측정하는데 사용한다.
• 처리부는 비즈니스 로직이 모인 곳으로 처리 결과를 Redis에 저장하거나 실시간 정보를 서비스들에게 바로 전달한다.
• 이때, 처리부의 비즈니스 로직 중에는 Blocking I/O가 있기 때문에 처리 시간에 가장 많은 영향을 준다.
• 조회부는 REST API를 서비스들에게 제공한다.

이떼, 처리부에서 비즈니스 로직으로 인하여 코드 변경이 가장 빈번하여 장애 발생 확률이 높으며 이 장애가 만약 장 중에 발생한다면 뒷 단의 시세 데이터를 사용하는 모든 서비스에 영향을 주게 되어 결국 사용자에게까지 잘못된 정보를 제공하게 된다.

장애가 복구된다 하더라도 장애 시간 동안 데이터 처리가 되지 않아 유실되거나 오염되었을 수도 있다.

이때, 한두 건 정도는 수작업으로 데이터를 보장할 수 있지만 초당 수천 거의 데이터를 다루기에 찰나의 장애에도 셀 수 없을 만큼의 많은 데이터 오류가 발생한다고 한다.

 

그래서 아래와 같이 처리부와 Redis를 두 개의 그룹으로 만드는 것이다. 평상시 A그룹을 통해 처리하다가 A에서 장애가 발생하면 모든 처리 요청을 B그룹으로 라우팅하는 것.

 

위와 같은 아키텍쳐를 가져갔을 때 무중단 배포가 가능해지며, 배포에 문제가 발생하면, 트래픽 라우팅 전환만으로 빠르게 롤백할 수 있기 때문에 배포 부담감을 줄일 수 있다고 한다.

또한 갑작스러운 시스템 장애를 대비해 각 그룹의 처리부를 두 개로 늘리고 ZooKeeper를 통해 리더를 선출하도록 했다. 그래서 각 그룹의 각 그룹의 리더만 데이터를 처리하도록 하여 중복 데이터 발생을 방지했다.

 

이렇게 처리부의 개수가 늘어나면서 고려 사항이 발생했다.

수신부는 처리부의 개수만큼 반복해서 데이터를 보내야 하는데 이는 곧 수신부의 성능 감소와 직결됐다.

특히, 아래 왼쪽 그림과 같이 처리부의 기능이 커져서 작은 서비스 단위로 분리하게 되면 처리부의 개수가 늘어날 수 있다.

이러한 문제를 해결하기 위해 메시지 브로커의 도입했다. 

 

메시지 브로커를 활용하여 수신부와 처리부를 Decoupling할 수 있고 수신부는 메시지 브로커로 데이터를 한 번만 전송하면 됐다.

그러나 시세 플랫폼에서 가장 중요한 것은 '낮은 지연시간'인데 아이러니하게도 메시지 브로커로 인해 지연시간이 늘어나게 됐다.

그래서 아무 메시지 브로커를 써서는 안됐다.

 

후보

1. UDP (User Datagram Protocol) 멀티캐스트
   → 라우팅 설정과 Kubernetes 배포 설정 필요로  빠른 개발이란 목적성에 반했음.
2. Kafka
   → 초기에는 Kafka사용. 자체 테스트 결과 지연시간이 15ms으로 3m/s인 Redis Pub/Sub에 비해 지연 시간이 길었음
3. Redis Pub/Sub
   → 낮은 지연 시간, 사용하기 쉽고 편리한 커맨드 지원,  Queue가 아닌 Pub/Sub 구조로 데이터를 보낸 뒤 유실 시키기 때문에 지연시간을 줄이는 면에서 유리.

따라서 Redis Pub/Sub을 활용했다고 한다.

 

 

Redis Pub/Sub 동작

여기서부터 블로그 내용과 추가적으로 서칭한 것을 기반으로 작성해보겠다.

https://github.com/redis

Redis

redis 코드도 일부 뜯어봤다.

 

일반 채널 구독 방식과 패턴 구독 방식이 다르지만 우선, 일반 채널 구독 방식을 기준으로 찾아봤다.

우선 redis/src/server.h를 보면, pubsub_channels라는 변수가 kvstore 타입으로 선언되어 있는 것을 확인할 수 있다.

redis/src/server.h

 

이후 redis/src/kvstore.h와 redis/src/kvstore.c를 살펴보면, kvstore 내부에는 아래와 같이 dict **dicts라는 필드가 존재하는 것을 알 수 있다.  즉, pubsub_channels는 내부적으로 하나 이상의 dict 자료구조를 가지며, 그 중 하나의 딕셔너리에서 채널별로 구독자를 관리한다고 한다.

redis/src/kvstore.c

 

실제로 pubsub_channels 내부 딕셔너리에서 key는 subscriber가 구독하고 있는 채널명(string)이고, value는 해당 채널을 구독하고 있는 client 구조체 포인터들의 연결 리스트(linked list)이다.

client 구조체

 

따라서,

1.  발행자가 특정 채널로 메시지를 보내면 Redis는 내부 dict에서 해당 채널에 해당하는 구독자 목록을 조회한다.
2. 조회된 연결 리스트를 순회하며, 리스트에 담긴 각 client 포인터를 통해 메시지를 전송한다.

즉, 전체 흐름을 정리하면, pubsub_channels 내부 dict → 채널명 key 조회 → clients 연결 리스트 순회 → 각 client에게 메시지 전송, 이런 순서로 동작.

 

---

다시, 시세 플랫폼 얘기로 돌아와서 

이렇게 Redis Pub/Sub을 통해 데이터를 아무리 빠르게 전달한다고 해도 만약 처리부에서 늦게 처리하게 되면 지연 시간은 다시 늘어날 수 밖에 없다. 

처리부에서 해야할일은 크게 두 가지이다.

하나는 TCP Socket으로부터 데이터를 읽는 일이고 나머지는 비즈니스 로직을 처리하는 것이다.

 

처리부는 Redis와 TCP연결을 맺은 후 Socket의 수신 버퍼로부터 데이터를 읽어가는데 이때 처리부가 데이터를 읽는 속도가 

지연시간에 큰 영향을 준다는 것이다.

바로, TCP 흐름 제어로 인해 발생하는 문제인데 이것은 송신 속도가 수신 속도보다 빠를 경우 데이터가 유실되는 것을 막기 위한 메커니즘으로 수신자는 Socket 수신 버퍼를 기준으로 한 번에 받을 수 있는 양 즉, Window. Size를 송신자에게 전달해주고 송신자는 다음 차례에 Window Size만큼의 데이터만 보냄으로써 데이터 유실을 방지하는 것이다.

 

따라서 수신부를 거쳐 Redis에서는 데이터를 계속 쏘다가 처리부의 Socket 수신 버퍼가 가득 차게되면 Redis는 데이터 전송을 줄이다가 멈추게 되어 그만큼 지연이 발생할 수 밖에 없다.

따라서, 비즈니스 로직 처리에 대해 별도의 Thread에 위임하며 특히, Blocking I/O 작업이 포함되어 있따면 반드시 그렇게 해야만 한다.

 

Spring Data Redis가 제공하는 ReactiveRedisTemplate을 사용하게되면 SpringDataRedis는 Lettuce라는 Redis Client 라이브러리를 기본적으로 사용한다.

Lettuce는 네트워크 라이브러리인 Netty를 사용하고
Netty의 Channel은 Socket을 추상화한 레이어로서 커넥션이 맺어진 이후 EventLoop에 등록된다.

여기서 EventLoop가 무한 루프를 돌면서 수신 버퍼의 데이터를 읽는 역할을 한다.

 

EventLoop는 운영체제에 따라 NIO, Epoll KQueue 등 여러방식을 사용하는데 NIO를 살펴보자.

NioEventLoop가 실행되면(run), 버퍼에 데이터가 있는지 확인하고(select), 데이터를 읽고(read), 변환하여(decode), 마지막으로 결과를 통보하는(notify) 식으로 동작한다.

즉, NioEventLoop는 비즈니스 로직을 처리하지 않는 것이 중요하다.

즉, 처리 성능을 높이기 위해 위와 같이 Blocking I/O가 포함된 비즈니스 로직에 대해서는 멀티스레딩을 통해 처리해야한다.

그러나 이때, 비동기로 처리하여 비즈니스 로직의 순서가 역전될 수 있는 위험성이 있다.

예를 들어 삼성전자 체결가 전문이 매우 짧은 시간 동안 여러 개가 들어왔는데 처리 순서가 바뀌게 되면 엉뚱한 가격이 보여질 수 있다.

 

이를 방지하기 위해 멀티스레딩 대신에 EventLoopGroup을 사용했다고 한다.

EventLoop는 내부 Queue를 이용하여 순서를 보장하고 하나의 Thread만 사용하기 때문에 동기화가 필요 없다는 장점이 있다.

EventLoop를 사용하더라도 어떤 EventLoop에서 처리해야 할지 알아야 순서를 보장할 수 있기 때문에 반드시 종목코드를 미리 알아야하고 이를 위해 JSON을 객체로 변환해야 했다.

 

 

 

EventLoop의 CPU 리소스 사용 증가 문제

그러나 이때, 트래픽이 증가할 수록 NioEventLoop의 CPU 자원을 사용하여 지연의 원인이 됐다.

해당 문제는 Redis Pub/Sub에서 제공하는 Channel을 사용했다고 한다.

쉬운 예로, 수신부가 데이터를 보낼 때 처리부의 EventLoop개수 만큼 Channel을 나누어 보내고 처리부는 이 Channel명을 보고 해당하는 EventLoop를 찾는 것이다. 이를 통해 수신부에서 발송한 데이터의 순서를 처리부에서 그대로 유지할 수 있고 무엇보다 NioEventLoop에서 더이상 객체 변환을 하지 않아도 되기 때문에 성능도 더 올라 갔다고 한다.