Producer의 메시지 배치 전송 내부 메커니즘

Producer의 메시지 배치 전송 내부 메커니즘

• Serialize > Partitioning > Compression > Record Accumulator 저장 > Sender에서 별도의 Thread로 전송
• Record Accumulator에서 Batch 단위로 전송하게 된다.

Producer Record와 Record Batch

Producer Record와 Record Batch

• KafkaProducer 객체의 Send() 메서드는 호출 시마다 하나의 ProducerRecord를 입력하지만 바로 전송되지 않고 내부 메모리에서 단일 메시지를 Topic 파티션에 따라 Record Batch 단위로 묶은 뒤 전송한다.
• 메시지들은 Producer Client의 내부 메모리에 여러 개의 Batch들로 buffer.memory 설정 사이즈만큼 보관할 수 있으며 여러 개의 Batch들로 한꺼번에 전송될 수 있다.

상세 분석

• Record Accumulator는 Partitioner에 의해서 메시지 배치가 전송이 될 토픽과 Partition에 따라 저장되는 kafkaProducer 메모리 영역
• Sender Thread는 Record Accumulator에 누적된 메시지 배치를 꺼내서 브로커로 전송한다.
• KafkaProducer의 MainThread는 send() 메서드를 호출하고 Record Accumulator에 데이터를 저장하고 Sender Thread는 별개로 데이터를 브로커로 전송한다.

linger.ms와 batch.size

linger.ms와 batch.size 1

linger.ms와 batch.size 2

• Sender Thread는 기본적으로 전송할 준비가 되어 있으면 Record Accumulator에서 1개의 Batch를 가져갈수도, 여러개의 Batch를 가져갈 수도 있다. 
  • Batch에 메시지가 다 차지 않아도 가져갈 수 있음
• linger.ms를 0보다 크게 설정하여 Sender Thread가 하나의 Record Batch를 가져갈 때 일정 시간 대기하여 Recored Batch에 메시지를 보다 많이 채울 수 있도록 적용
• max.inflight.requests.per.connection 값에 따라 몇개의 Batch를 보낼 수 있는지 설정

Producer의 linger.ms에 대한 고찰

• linger.ms를 반드시 0보다 크게 설정할 필요가 없다.
• Producer와 Broker간의 전송이 매우 빠르고 Producer에서 메시지를 적절한 Record Accumulator에 누적된다면 linger.ms가 0이 되어도 무방
• 전반적인 Producer와 Broker간의 전송이 느리다면, linger.ms를 높여서 메시지가 배치로 적용될 수 있는 확률을 높이는 시도를 해볼 만하다.
• linger.ms는 보통 20ms 미만으로 설정 권장

실습하기

//Bootstrap.servers, key.serializer.class, value.serializer.class
//props.setProperty("bootstrap.servers", "192.168.56.101:9092");
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.101:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.ACKS_CONFIG, "0");

props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "32000");
props.setProperty(LINGER_MS_CONFIG, "20");

 

 

Producer의 Sync와 Callback Async에서의 Batch

• 기본적으로 KafkaProducer 객체의 send() 메서드는 비동기이며, Batch 기반으로 메시지 전송
• Callback  기반의 Async는 비 동기적으로 메시지를 보내면서 RecordMetaData를 Client가 받을 수 있는 방식을 제공
• Callback 기반의 Async는 여러 개의 메시지가 Batch로 만들어짐
• RecordMetaData recordMetadata = KafkaProducer.send().get()와 같은 방식으로 개별 메시지 별로 응답을 받을 때까지 block 되는 방식으로는 메시지 배치 처리가 불가능
  • 전송은 배치레벨이지만 배치에 메시지는 단 1개
  • 동기와 비동기는 단순 통신 속도의 차이뿐만 아니라 배치 시간만큼에 차이도 있는 것이다.