[Thread] Message Broker와 스레드 (feat. Kafka)

처리를 왜 분산시켜?

학습을 시작하기에 앞서서 이런 의문점이 생겼던 것 같습니다. 처리를 왜 분산시키는가? 한 컴퓨터 안에서 서버를 여러 대 돌린다면 그 만큼 소통하는 데 오버헤드가 훨씬 적지 않겠느냐? 하는 것이었습니다. 또 분산해서 처리하려는 환경을 구축해놓는 것도 비용이 아닌가? 라고 생각했었습니다.

 

사이즈가 큰 프로덕션을 기준으로 해서 분산 처리환경을 구축해놓으면 다음과 같은 이점들이 있습니다.

1. 수직 확장(Scale-up) vs 수평 확장(Scale-out)
   1. 슈퍼 컴퓨터 한 대를 두는 것이 수직 확장이고, 저렴한 서버 여러 대를 병렬로 연결하는 것이 수평 확장입니다.
2. 고가용성 (High Availability, HA)
   1. SPOF(Single Point of Failure): 서버가 한 대뿐이라면 그 서버가 바로 '단일 장애점'이 되므로, 그 서버만 죽으면 바로 전체 서비스가 죽습니다.
   2. 서버를 여러대로 찢어놓으면 다른 서버로 트래픽을 돌릴 수 있습니다. 사용자가 서버의 장애 상태를 모르게할 수 있습니다.
3. 저지연 (Low Latency)
   1. Edge Computing / Multi-region: 전 세계 거점에 서버를 분산 배치하면 그 주변의 요청을 주변에 있는 거점의 서버가 처리합니다. 이를 통해서 응답 시간을 획기적으로 줄일 수 있습니다. 한국에서 미국으로 요청을 보내면 그만큼 물리적인 시간이 걸릴 것이니까요.
4. 경제성 (Elasticity)
   1. 매일 트래픽이 확 몰리지는 않을 것입니다. 그렇지 않은 날이 더욱 많을 것입니다. 그 하루를 위해서 비싼 서버를 계속 켜놓을 수 는 없습니다.
   2. Auto-scaling: 분산 환경에서는 트래픽이 줄어들면 서버를 자동으로 끄고(Scale-in), 많아지면 자동으로 키면(Scale-out) 비용을 최적화 할 수 있습니다. 

 

처리 분산의 예시

  한적한 동내 카페와 방송 탄 인기 카페로 이야기를 이어나가 보겠습니다.

 

  동네 카페에서는 사장님이 혼자 주문도 받고, 커피도 만들고, 계산도 합니다. 손님이 아메리카노를 주문하면 주문을 받고, 커피 머신에서 커피를 만들고, 포장하고, 계산합니다. 이때 다른 손님이 오면 그 손님은 사장님이 계산을 마칠 때까지 기다려야 합니다. 열정 가득한 사장님이기 때문에 혼자서 가게를 운영할 수 있었고, 노하우도 생겨서 손님이 조금씩 많아져도 괜찮게 서비스를 했습니다.

 

  평소에는 한적한 동네 카페였다가, 연애인이 커피 맛집이라고 방송에서 소문을 냈습니다. 사람들이 급격하게 늘었고, 사장님이 혼자서 처리하기에는 너무 많은 인파였습니다. 사장님은 많은 고민을 했습니다. 알바를 뽑아도 해결이 잘 안되는 듯 했기 때문입니다. 상황을 분석해보았습니다.

계산대에서 주문 받음 -> 커피 머신 앞으로 감 -> 커피 만듦 -> 포장대로 감 -> 포장함 -> 계산대로 감 -> 계산함

 

 

이런 식으로 순차적으로 일을 진행하는 방식에서 비효율적이라고 생각하는 부분이 보이기 시작했습니다.

1. 작업을 위해서 이동하는 시간
2. 계산대가 비어있으면 아무런 응대없이 기다려야하는 손님

  이런 것들이 문제점이라고 판단한 카페 사장님이 업무 방식의 개선을 시도합니다. 주문 담당과 음료 담당을 나눈 것입니다.

• 주문 담당은 오로지 주문만 합니다. 커피 머신이 밀리든 말든 주문만 받습니다. 주문을 받으면 주문 전표를 순서대로 계산대와 커피 머신 사이의 작은 테이블에다가 음료 담당이 잘 볼수있도록 둡니다.
• 음료 담당도 마찬가지로 주문 처리를 신경쓰지 않고 오로지 전달받은 음료를 만드는 것에만 신경을 씁니다. 1000개가 밀려있든 말든 자기 속도대로 하나씩 처리합니다.

이렇게 업무 방식을 개선하니, 다음과 같은 문제들이 해결되었습니다.

1. 사장님과 알바생은 경력이 달라서 일 처리 속도도 많이 다릅니다. 기존에는 서로가 순차적으로 실행해서 어떤 사람은 1분만에 커피가 나오고, 어떤 사람은 10분만에 커피가 나옵니다. 네이버 리뷰에서 서비스 타임이 느리다고 한 리뷰의 이유를 몰랐던 사장님이 평균 서비스 타임을 낮춤으로써 해결할 수 있었습니다.
2. 작업을 위한 작업대 간 이동 시간이 없어졌습니다. 가운데의 작은 테이블에다가 전표로 암묵적으로 소통하면 되기 때문입니다.
3. 손님의 불만이 없어졌습니다. 평균 서비스 타임의 이유도 있었지만, 일단 가게에 들어오면 계산대에 사람이 있어서 주문을 먼저 하고 기다릴수 있었습니다. 손님에게는 심지어 체감되지 않는 서비스 타임의 변화일지라도 일단 응대를 받으니 훨씬 덜 답답하겠죠.
4. 커피 머신이 잠깐 먹통이돼서 안 돌아가도 일단 주문은 받을 수 있습니다. 이에 대해서 손님에게 주문을 받으면서 '조금 시간이 걸릴 수 있어요'라고 응대할 수 있습니다.

사장님은 이런 문제를 해결하고 최대한의 수익을 가져갈 수 있었습니다!

계산대에서 주문 받음 -> (테이블에 전표로 소통) -> 커피룰 만들고 포장까지함 -> (다 됐다고 알림) -> 전달 및 계산함.

 

Apache Kafka

Apache Kafka

 

  Apache Kafka는 2010년 링크드인에서 개발을 시작해서 2011년에 오픈소스로 공개되었습니다. 여기서 2012년에 Apache Software Foundation으로 이관되어 개발이 이어져오고 있습니다. 지금은 전 세계 대규모 서비스의 표준 스트리밍 플랫폼입니다. 카프카는 여러 시스템이 주고받는 이벤트(로그, 메시지, 데이터)를 실시간으로 저장하고 전달하는 도구입니다. 대규모 데이터를 실시간으로 전달, 처리하기 위한 분산 메시징/스트리밍 플랫폼으로, 시스템과 시스템 사이에서 데이터를 빠르고 안정적으로 흘려보내는 파이프라인입니다.

 

  위에서 들었던 예시의 단어들을 개발 용어와 1대1 매핑해보자면, 다음과 같습니다.

카페	개발	설명
작은 테이블	Kafka	주문(데이터)이 잠시 머무르는 버퍼(Buffer) 공간
주문 전표	Message (Event)	처리해야 할 데이터(주문 정보)
주문 담당	Producer	데이터를 생성해서 카프카(작은 테이블)에 넣는 역할
음료 담당	Consumer	카프카(작은 테이블)에서 데이터를 가져와서(Consume) 처리하는 역할
주문만 받기	Non-blocking	뒷단이 바쁘든 말든 멈추지 않고 요청을 Accept
자기 속도대로	Flow Control	Consumer가 과부하에 걸리지 않게 앞단에서 적절히 조절.
작업대 간 이동 시간 제거	Loose Coupling	주문 서버와 처리 서버가 직접 API 호출로 대화하지 않고 카프카(작은 테이블)만 바라봄. -> 서로의 존재를 몰라도 시스템이 돌아감.
커피 머신이 먹통이 되도 돌아간다	장애 격리 (Fault Tolerance)	주문 전표는 음료 담당(Consumer)과 관계없이 잘 쌓이므로 기계를 빨리 고치고 처리하면 됨.

 

Kafka's System Flow

Kafka의 전체 흐름

 

Kafka의 전체 흐름을 보면 다음과 같습니다.

 

1. Producer

• 웹 애플리케이션 환경에서 사용자의 HTTP 요청을 처리하고 있는 스레드 (Spring boot 환경이라면 Tomcat 스레드)가 요청을 받음
• 톰캣은 자기 스레드 풀에서 Idle 상태로 놀고있던 스레드를 하나 깨워서 이 요청을 맡김 (카프카 프로듀서 입장에서 User Main Thread가 이 톰캣 스레드 풀 내의 스레드가 되는 것.)
  • 이 요청을 맏길 때, 즉 톰캣 스레드의 역할은 RecordAccumulator (버퍼)에 producer.send("order", data; 로 버퍼에 데이터를 넣는 것으로 끝남.
  • HTTP 응답을 사용자에게 반환하고 (200 OK 등) 다시 스레드 풀에서 Idle 상태로 놀러 감.
• Sender thread가 버퍼에 일정량의 데이터가 쌓였으면 카프카 브로커 서버에게 패킷을 보냄.
  • 이때 브로커에게 데이터를 쏘고 나서, 브로커의 응답을 기다리지 않고 max.in.flight.requests.per.connection 값 만큼 연속으로 패킷을 보냄.
• 그러다가 브로커 서버로 부터 ACK가 오면 그때 사용자와 공유하는 공유 메모리에다가 완료를 표시하고 성공 내역을 기록함. 
  
  • 사용자는 Sender thread가 반환하는 방법을 선택적으로 운용할 수 있음.
    • Sync/Blocking, Fire and Forget, Async/Callback

2. Broker

• 브로커 서버는 Acceptor thread를 통해 Producer가 송신하는 패킷을 리스닝함.
  • accept()에서는 CPU 작업(패킷 파싱, 헤더 읽기)를 진행함. 
  • accept()(연결 수립) 작업은 오버헤드가 거의 없는 매우 빠른 작업이므로 1개의 스레드여도 충분함.
• Acceptor thread는 Network thread들에게 해당 요청들을 배정함.
• Network threads는 고정된 상수 개수 만큼의 스레드들이 존재하며, 이 요청들을 Request Queue에 넣음.
• Request Queue를 바라보고 있는 Idle 상태의 I/O thread들이 Polling 방식으로 이 요청들을 일이 끝나는 대로 가져감.
• I/O thread들이 이 요청을 수행하고 Request Queue에 들어있던 요청 객체안에 달린 Tag를 이용해서 대응하는 Network thread의 Response queue에 넣음.
• I/O thread는 Response queue에 데이터를 넣을때 selector.wakeup()을 호출해서 Network thread를 깨워서 응답이 도착했음을 알려줌. Network thread는 이를 받아서 Acceptor thread를 거치지 않고 바로 Sender thread에게 응답을 보냄.
• Sender는 이 응답을 사용자가 설정한 방법으로 처리해서 요청을 마무리함.

 

  이렇게 2가지로 분리되어 프로듀서 서버와 브로커 서버로 나눠서 생각할 수 있습니다. 아래는 각각 시퀸스 다이어그램입니다.

 

Producer Region

(Inside of producer) Producer Sequence Diagram

Sender thread가 Network thread에게 대답을 안 기다리고 막 보내도 괜찮은가?

Sender가 응답을 기다리지 않고 여러 개의 패킷을 쏴버리는 중에 그 중 한 개가 네트워크 오류 등으로 실패할수도 있습니다.

이때 전송이 실패한 패킷에 대해서 TCP가 retry를 합니다. 이때 패킷이 잘 도착하면 DB에는 2번 패킷, 1번 패킷으로 순서가 뒤집혀서 들어가게 됩니다. 순서가 중요한 정보이면 이전에는 max.in.flight를 무조건 1로 설정했었습니다.

 

Idempotence(멱등성)

0.11 버전 이후의 Kafka에서는 'enable.idempotence=true' 옵션이 기본 값입니다. (Exactly Once)

이는 패킷에 Sequence number(번호표)를 붙이고는 broker가 도착해야하는 sequence 값을 기억해서 다른 패킷이 먼저 도착하면 메모리에 잠깐 들고있다가 desired sequence number를 가지고 있는 패킷이 도착하면 순서를 다시 맞춰서 저장합니다.

 

Kafka Message Transfer Policy

종류	이름	Acks	특징	단점	한 줄 정리
Lv 1	At Most Once	0	빠름	유실 가능성 존재	알아서 받으세요하고 Sender가 막 던짐
Lv 2	At Least Once	all == 1	데이터 안전	중복 가능성 존재	Ack를 못받으면 Sender가 똑같은 걸 받을 때까지 다시 던짐
Lv 3	Exactly Once	all + idempotence	데이터 오차 없음	약간의 오버헤드	Ack를 못받으면 Sender가 똑같은 걸 받을 때까지 다시 던지는데, 브로커가 중복을 눈치채고 1개만 저장함.

Broker Region

(Inside of broker) Kafka Reactor Pattern Sequence Diagram

 

Kafka Streams

Topic

• 데이터의 분류 기준, 폴더이름.
• 개발자가 데이터를 구분하기 위해 붙인 논리적인 이름 (폴더)
• order(주문 정보만 모으는 곳), payment(결제 정보만 모으는 곳), click_log(클릭 로그만 모으는 곳)

Partition

• 토픽에 데이터가 너무 많이 들어오면 컴퓨터 한 대로는 감당이 안되기 때문에, 토픽을 여러 갈래로 찢어서 저장하고, 이 갈래 하나하나를 파티션이라고 합니다.
• order-0, order-1 이런 식으로 디스크에 저장되는 물리적인 파일입니다.
• 순서는 파티션 안에서만 보장됩니다.

Task

• Kafka streams에서 가장 작은 작업 단위. (파티션과 1:1 매핑)
• task는 해야 할 일의 덩어리이고, thread는 그 일을 수행하는 일꾼으로 전자는 고정이고 후자는 유동적입니다.

Kafka 토픽의 데이터를 실시간으로 변환, 집계, 조인하는 라이브러리입니다.

Kafka streams의 핵심은 "병렬성을 최대한 사용해서 throughput을 극대화 하는 것"입니다.

주문 토픽에서 들어오는 주문을 실시간으로 지역별 매출 합계로 바꿔서 다른 토픽에 넣을 때 등의 작업에서 사용하게 됩니다.

 

기본적으로 Kafka streams는 적은 개수의 stream thread가 여러 task를 RR(Round-Robin)처럼 번갈아서 처리하고,
Task마다 독립된 Kafka consumer + Producer + 상태 저장소(RocksDB)를 가지고 있지만 실제 실행은 적은 개수의 스레드가 순차적으로 돌립니다. 가령 Task 1 조금 수행 -> Task 2 조금 수행 -> ... -> Task N 조금 수행 -> Task 1 조금 수행 ... (반복) 인 셈입니다.

이렇게 하면 context switching 비용은 있지만 자원을 100% 사용할 수 있습니다. 

 

CPU Utilization을 최대한 높아야 하는 방향성으로 스레딩 방식이 설계되어 있습니다.

utilization이 낮으면 over-provisioning(과다 할당)으로 불필요한 비용이 발생하고, 높으면 throughput과 ROI가 증가합니다.

너무 많은 스레드를 생성하면 context switiching overhead가 많아지는데, streams에서 rocksDB에 상태를 저장, 읽기를 하는 등의 작업 자체가 대표적인 I/O bound 작업이기 때문에 굳이 스레드를 많이 쓸 필요 없습니다. 이러한 관점에서 Kafka는 streams를 적은 스레드의 개수로 처리하는 방법을 취했습니다.

 

Kafka Streams

 

❓ 내가 아는 RR(Round-Robin)은 Time slice를 기준으로 돌아가는 메모리 선점여부에 대한 스케줄링인데? Acceptor에서 RR을 쓴다고?

OS 스케줄링의 라운드 로빈

• Scheduler (일하는 시간과 순서를 통제)
• 실행 도중에 time slice 만큼 CPU를 사용하면 다음 순번에게 CPU를 뺏김.

부하 분산의 라운드 로빈 (Kafka의 acceptor가 하는 일)

• Dispatcher (Connection을 누구에게 Assign할지 만 결정)
• 단순히 순서대로 돌아가면서 배정한다는 의므로, 시간 제한이나 쫓겨난다는 개념이 없음. 그냥 배분 순서가 Round로 돌아간다는 뜻.

acceptor's pseudo code (in kafka)

int threadIndex = 0;

int totalThreads = 3;

while (newClientArrives()) {

    int assignedThread = threadIndex % totalThreads; // 0, 1, 2, 0, 1, 2 ...

    dispatchTo(assignedThread);

    threadIndex++; // 다음 차례 준비

}

❓ 프로듀서와 브로커의 TCP 연결은 언제 연결되고 언제 소멸돼?

new KafkaProducer()를 선언한다고 해서 연결을 맺는게 아닙니다.

Sender thread가 데이터를 보내려고 봤더니 연결이 없을 때 맺습니다!

 

언제 연결하는가?

사용자 스레드가 Accumulator에 메시지를 넣고, send를 할 때가 된다면 Sender thread가 자신의 연결 목록을 보고, 브로커 A와 연결되어 있나 확인합니다.

-> 없으면 그때 TCP 3-way handshake를 수행합니다. 즉, 첫 번째 메시지를 보낼 때 연결이 맺어집니다. (Lazy init)

 

어디에 정보가 저장되어 있는가?

프로듀서와 브로커는 Java와 Scalar로 작성되어 있습니다. 이때 TCP 소켓 정보는 Sender thread가 소유한 network client 객체 내부에 보관됩니다. 또한 이 정보는 Sender thread만이 독점 관리합니다.

• KafkaProducer instance -> Sender 객체 -> NetworkClient 객체 -> Java NIO Selector
• 이 Selector 안에 브로커별로 SocketChannel이 등록되어 있음.
  • 브로커 1번: Connected (SocketChannel A)
  • 브로커 2번: Disconnected
  • 브로커 3번: Connected (SocketChannel B)

언제 연결 정보가 소멸되는가?

Handshake 비용이 매우 비싸기 때문에 한 번 연결하면 최대한 끊지 않고 재사용합니다. 끊기는 순간이 있다면, 아래와 같습니다.

• Idle Timeout: 설정된 시간 동안 데이터를 하나도 안 보내면 끊음 (keep-alive default 9min)
• 프로듀서 종료: producer.close()
• 네트워크 오류: 브로커가 죽었을 때.

 

❓네트워크 스레드가 왜 굳이 적은 양의 상수 개의 고정 스레드로 운용돼? 톰캣처럼 다이나믹 스레드 풀로 선언하고, Request Queue 없이 바로바로 I/O 작업을 수행하면 안돼?

Blocking과 Non-Blocking 차이의 핵심입니다.

 

Kafka는 연결된 클라이언트가 수천, 수만 개입니다.

톰캣 방식이라면 1:1 매핑으로 스레드를 만 개 만들어야하는데, 이 수치는 CPU에게 Scheduling하다가 죽으라는 소리입니다. 즉, thrashing 현상을 피할 수가 없습니다.

 

Kafka는 네트워크 스레드를 CPU만 쓰는 작업을 하게 합니다. I/O bound 작업이 아니기 때문에 굳이 스레드가 많아봤자 Context switching만 쓸데없이 일어납니다. 따라서 적은 수의 개수로 고정할 수 있습니다. 그리고 사이에 완충 작용을 하는 큐를 두고, 이를  I/O thread와 공유하면 수만 개의 연결을 충분히 처리할 수 있는 구조가 됩니다.

Disk 작업이 밀려도 네트워크 접속은 멈추지 않게 할수 있습니다!

 

Default size

Sender thread: 1

Request queue: 500

Response queue: 제한 없음

Network thread: 3 (CPU 코어 수 비례, 보통은 비슷하게 설정)

I/O thread: 8 (Disk 개수에 비례, disk가 10개라면 20~30개로 설정하는게 바람직.)

❓ Kafka는 왜 굳이 영속성 기록을 해? 멀쩡한 DB가 있는데 굳이 디스크에다가 로그를 쓸 필요가 있나?

결론적으로 말하자면 DB는 현재 상태를 저장하는 곳이고, Kafka는 Event의 기록(Log)을 저장하는 곳입니다. 정보의 역할이 다릅니다.

 

1. Replayability

• 개발자가 실수해서 지난 3일 동안 게시판 조회수가 2배로 증가하는 버그가 있었음.
• DB만 있으면 값은 다 update 되어있기 때문에 원래 얼마였는지 알 수가 없음.
• Kafka에는 조회수 증가 요청 로그가 그대로 남아있음. -> Consumer offset을 3일 전으로 돌려서 DB를 덮어쓰면 복구 가능.
• 즉, Kafka의 event log는 블랙박스 같은 역할을 함.

2. Traffic Spike Tolerance

• 블랙 프라이데이 세일로 평소보다 트래픽이 100배 폭주함.
• DB만 있으면 메모리의 한계로 DB 서버가 다운됨.
• Kafka의 메모리만 있으면 이것도 OOM으로 RAM에 올린 메모리이므로 데이터가 날라감.
• Kafka의 디스크 사용이 있어야 일단 log로 작성하고 disk 작업을 순차적으로 처리해서 손실없이 트래픽을 받아낼 수 있음.
• 즉, 무손실 트래픽 수용을 위한 완충재 역할을 함.

3. One source, Multi use

• 회원가입 이벤트가 발생함.
• 메인 DB(MySQL)은 회원정보 저장, 검색 엔진(Elasticsearch)은 닉네임 검색용 저장, 빅데이터(Hadoop)는 통계 분석용 저장 등... 하나의 정보를 많은 곳에서 사용함.
• Producer는 Kafka 디스크에 딱 한번 write하고, 이를 참조하는 DB들은 자기의 속도에 맞게 디스크의 내용을 읽어감. (메인 DB는 바로, Hadoop은 하루에 한번 몰아서 등.)

즉 이벤트 로그를 기록함으로써 더 효율적이고 안정적인 서비스를 가능하게 하기위해 카프카는 영속성 기록을 진행합니다!

❓ TCP의 재전송 정책

TCP는 정해진 횟수만큼, 시간을 점점 늘려가면서 연결 요청을 시도하다가 그래도 안되면 손절칩니다.

 

1. Exponential Backoff

ACK가 오지 않으면 네트워크가 막혔다고 판단을 해서, 좀 더 천천히 보내야겠다고 판단합니다. 그래서 재전송 간격을 2배씩 늘립니다.

 

2. tcp_retries2

OS에는 각각 TCP 재전송을 몇 번까지 할 것인가에 대해서 설정값이 있습니다. 리눅스 서버의 경우에는 커널 파라미터로 보통 15회로 설정되어 있습니다. 

 

3. 애플리케이션 자체 설정

Kafka같은 애플리케이션에서 timeout을 설정해서, TCP가 포기하기 전에 애플리케이션이 먼저 타임아웃을 내고 소켓을 닫아버립니다. (socket.close())

따라서 카프카는 새로운 소켓을 열거나 다른 브로커에게 새로운 요청을 보내는 식으로 Kafka retry를 하게 됩니다.

E-commerce example based on Kafka

참고 문헌

https://seamless.tistory.com/100

아파치 카프카 소개