9. Kafka - Producer 메커니즘 1

Kafka Producer의 내부 배치 전송 방식

 

1) KafkaProducer.send()

2) Serializer -> 직렬화

3) Partitioner -> 파티셔닝

4) Compression -> 압축 방식

5) Record Accumulator
6) Sender 

 

이렇게 send 메서드부터 Sender까지 총 6단계로 이루어져 있다.

 

1. KafkaProducer.send()

.send()는 호출 시마다 하나의 ProducerRecord를 입력하지만 내부 메모리에서 파티션에 나뉘는 것에 따라 Record Batch 형태로 전송된다.

buffer.memory 설정 크디만큼 메모리에 보관하였다가 send되며 여러 개의 batch들로 한꺼번에 전송될 수 있다.

이것은 카프카의 Main Thread에 의해 이루어진다.

 

2. Serializer

 

send() 메서드를 통해 메시지(이벤트)를 보내면, 직렬화기를 통해 토픽에 전송하기 위한 바이트 배열로 변환한다.

즉 데이터를 전송하기 위한 절차로, JSON이면 JsonSerializer, String이면 StringSerializer와 같이 여러 직렬화기를 제공하고 있다.

보낼 데이터들이 직렬화가 이루어진 후, 파티셔너의 역할로 넘어간다.

 

3. Partitioner

 

Partitioner는 여러 파티션으로 나누어져 있는 토픽으로 보내기 위해 파티셔닝 함수에 의해 특정 파티션으로 할당하는 역할을 한다.

이 과정에서, 메세지의 키를 기반으로 파티션을 선택하게 되고, 이는 파티션으로 나누어져 있는 경우 레코드의 순서를 보장하는 역할을 한다.

 

4. Compression (압축 방식)

 

압축을 사용하는 경우, 파티셔닝된 메시지는 필요에 따라 압축되며, 이는 네트워크 대역폭을 줄이고 디스크 공간을 절약하기 위해 사용된다.

압축 방식을 지정하여 사용하지 않는 경우 압축하지 않는다.

 

5. Record Accumulator

 

압축된 메시지는 레코드 어큐뮬레이터에 추가된다. 레코드 어큐뮬레이터는 여러 프로듀서 스레드에서 오는 메시지를 모아서

"배치 처리"를 수행하는 메모리 영역이다. 즉 각 파티션에 보내질 데이터들을 모아서 보내는 역할을 한다.

 

linger.ms = Sender Thread로 메시지를 보내기 전에, 배치로 메시지를 보내기 위해 최대 대기하는 시간

buffer.memory = Record accumulator의 전체 메모리 사이즈

batch.size = 단일 배치의 사이즈 (각 파티션마다 보내질 배치들의 각 사이즈)

 

6. Sender

 

Record Accumulator에서 배치 처리된 메시지가 Sender로 보내지며

앞선 것과는 달리 Sender Thread가 별개로 존재하여 데이터를 전송하는 역할을 한다.

 

+)

Deserializer (역직렬화)는 Consumer에서 발생한다. 즉, 브로커에서 메시지를 가져와서 처리하기 전에, 메시지는 역직렬화가 이루어진 후 소비된다.  

 

linger.ms를 0보다 크게 설정하는 경우, 일부러 기다리는 상황이 있을 수 있으므로 굳이 높일 필요 없으며

20ms 이하로 설정을 권장한다고 한다.