Apache Kafka - Kafka Producer(카프카 프로듀서) - 2
Apache Kafka - Kafka(카프카)란 ? 분산 메시징 플랫폼 - 1
이전 포스팅에서 간단히 카프카란 무엇이며 카프카의 요소들에 대해 다루어보았다. 이번 포스팅에는 이전에 소개했던 요소중 카프카 프로듀서에 대해 다루어볼 것이다.
카프카 프로듀서란 카프카 클러스터에 대해 데이터를 pub, 기록하는 애플리케이션이다.
카프카 프로듀서의 내부 구조
우리는 카프카 프로듀서를 사용하기 위해 추상화된 API를 사용한다. 하지만 우리가 사용하는 API는 내부적으로 많은 절차에 따라 행동을 한다. 이러한 프로듀서가 책임지는 역할들은 아래와 같다.
- 카프카 브로커 URL 부트스트랩 : 카프카 프로듀서는 카프카 클러스터에 대한 메타데이터를 가져오기 위해 최소 하나 이상의 브로커에 연결한다. 프로듀서가 연결하길 원하는 첫 번째 브로커가 다운될 경우를 대비하여 보통 한개 이상의 브로커 URL 리스트를 설정한다.
- 데이터 직렬화 : 카프카는 TCP 기반의 데이터 송수신을 위해 이진 프로토콜을 사용한다. 이는 카프카에 데이터를 기록할 때, 프로듀서는 미리 설정한 직렬화 클래스를 이용해 직렬화 한 이후에 전송한다. 즉, 카프카 프로듀서는 네트워크 상의 브로커들에게 데이터를 보낵기 전에 모든 메시지 데이터 객체를 바이트 배열로 직렬화한다.
- 토픽 파티션의 결정 : 어떤 파티션으로 데이터가 전송돼야 하는지 결정하는 일은 카프카 프로듀서의 책임이다. 만약 프로듀서가 데이터를 보내기 위해 API에 파티션을 명시하지 않은 경우 파티셔너에 의해 키의 해시값을 이용해 파티션을 결정하고 데이터를 보낸다. 하지만 키값이 존재하지 않으면 라운드 로빈 방식으로 데이터를 전송한다.
- 처리 실패/재시도 : 처리 실패 응답이나 재시도 횟수는 프로듀서 애플리케이션에 의해 제어된다.
- 배치처리 : 호율적인 메시지 전송을 위해서 배치는 매우 유용하다. 미리 설정한 바이트 임계치를 넘지 않으면 버퍼에 데이터를 유지하고 임계치를 넘으면 데이터를 토픽으로 전송한다.
카프카 프로듀서의 내부 흐름을 설명하면, 프로듀서는 키/값 쌍으로 이루어진 데이터 객체를 전달 받는다. 그리고 해당 객체를 미리 정의된 시리얼라이저 혹은 사용자 정의 시리얼라이저를 이용해 직렬화 한다. 데이터 직렬화 후 데이터가 전송될 파티션을 결정한다. 파티션 정보가 API 호출에 포함되어 있다면 해당 파티션에 직접 데이터를 보내지만 파티션정보가 포함되지 않았다면 데이터 객체의 키값의 해시를 이용해 파티셔너가 데이터를 보낼 파티션을 결정해준다. 만약 키값이 null이라면 라운드 로빈 방식으로 파티션을 결정한다. 파티션 결정이 끝나면 리더 브로커를 이용하여 데이터 전송 요청을 보낸다.
카프카 프로듀서 API
카프카 프로듀서 API를 사용하기 위한 필수파라미터들이다.
- bootstrap.servers : 카프카 브로커 주소의 목록을 설정한다. 주소는 hostname:port 형식으로 지정되며 하나 이상의 브로커 리스트를 지정한다.
- key.serializer : 메시지는 키 값의 쌍으로 이뤄진 형태로 카프카 브로커에게 전송된다. 브로커는 이 키값이 바이트 배열로 돼있다고 가정한다. 그래서 프로듀서에게 어떠한 직렬화 클래스가 키값을 바이트 배열로 변환할때 사용됬는 지 알려주어야한다. 카프카는 내장된 직렬화 클래스를 제공한다.(ByteArraySerializer, StringSerializer, IntegerSerializer / org.apache.kafka.common.serialization)
- value.serializer : key.serializer 속성과 유사하지만 이 속성은 프로듀서가 값을 직렬화 하기 위해 사용하는 클래스를 알려준다.
public KafkaProducer<String, String> getProducer(){
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092,localhost:9093");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
return new KafkaProducer<>(producerProps);
}
자바 클래스로 위의 필수 설정들을 사용하여 카프카 프로듀서를 생성하는 코드이다. Properties 객체를 이용해 모든 설정 값을 설정해주고 카프카 프로듀서 객체 생성자의 매개변수로 해당 설정을 넣어준다.
카프카 프로듀서 객체와 ProducerRecord 객체
프로듀서는 메시지 전송을 위해 ProducerRecord 객체를 이용한다. 해당 ProducerRecord는 토픽이름, 파티션 번호, 타임스탬프, 키, 값 등을 포함하는 객체이다. 파티션 번호, 타임스탬프, 키 등은 선택 파라미터 이지만, 데이터를 보낼 토픽과 데이터 값을 반드시 포함해야한다. 그리고 보낼 파티션 선정에는 3가지 방법이 존재한다.
- 파티션 번호가 지정되면, 지정된 파티션에 데이터를 전송한다.
- 파티션이 지정되지 않고 키값이 존재하는 경우 키의 해시 값을 이용하여 파티션을 정한다.
- 키와 파티션 모두 지정되지 않은 경우 파티션은 라운드 로빈 방식으로 할당된다.
public void sendMessageSync(String topicName, String data) throws InterruptedException, ExecutionException {
try(KafkaProducer<String, String> producer = getProducer();){
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topicName, data);
Future<RecordMetadata> recordMetadata = producer.send(producerRecord);
RecordMetadata result = recordMetadata.get();
}
}
자바 코드를 이용해 데이터 객체를 생성한 후에 동기식으로 토픽에 데이터를 보내는 코드이다.
ProducerRecord(String topicName, Integer partition, K key, V value)
ProducerRecord(String topicName, Integer partition, Long timestamp, K key, V value)
ProducerRecord(String topicName, K key, V value)
ProducerRecord(String topicName, V value)
ProducerRecord의 생성자 리스트의 일부를 보여준다.
일단 send()를 사용해 데이터를 전송하면 브로커는 파티션 로그에 메시지를 기록하고, 메시지에 대한 서버 응답의 메타데이터를 포함한 RecordMetadata를 반환하는데, 이 객체는 오프셋,체크섬,타임스탬프,토픽,serializedKeySize 등을 포함한다.
앞에서 일반적인 메시지 게시 유형에 대해 설명했지만, 메시지 전송은 동기 또는 비동기로 수행될 수 있다.
- 동기 메시징 : 프로듀서는 메시지를 보내고 브로커의 회신을 기다린다. 카프카 브로커는 오류 또는 RecordMetadata를 보낸다. 일반적으로 카프카는 어떤 연결 오류가 발생하면 재전송을 시도한다. 그러나 직렬화, 메시지 등에 관련된 오류는 애플리케이션 단에서 처리돼야 하며, 이경우 카프카는 재전송을 시도하지 않고 예외를 즉시 발생시킨다.
- 비동기 메시징 : 일부 즉각적으로 응답 처리를 하지 않아도 되는 혹은 원하지 않는 경우 또는 한두 개의 메시지를 잃어버려도 문제되지 않거나 나중에 처리하기를 원할 수 있다. 카프카는 send() 성공 여부와 관계없이 메시지 응답 처리를 콜백 인터페이스 제공한다.
send(ProducerRecord<K,V> record, Callback callback)
위의 콜백 인터페이스는 onCompletion 메서드를 포함하는데, 오버라이드하여 구현하면 된다.
producer.send(producerRecord, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(exception != null) {
log.info("예외 발생");
/*
* 예외처리
*/
}else {
log.info("정상처리");
/*
* 정상처리 로직
*/
}
}
});
onCompletion 메소드 내부에 성공했거나 오류가 발생한 메시지를 처리할 수 있다.
사용자 정의 파티셔닝
카프카는 내부적으로 내장된 기본 파티셔너와 시리얼라이저를 사용한다. 하지만 가끔은 메시지가 해당 브로커에서 동일한 키는 동일한 파티션으로 가도록 사용자 정의 파티션 로직을 원할 수 있는데, 카프카는 이렇게 사용자 정의 파티셔너 구현을 할 수 있도록 인터페이스를 제공한다.
class CustomPartition implements Partitioner{
@Override
public void configure(Map<String, ?> configs) {
// TODO Auto-generated method stub
}
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes,
Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
//TODO : 파티션 선택 로직
return 0;
}
@Override
public void close() {
// TODO Auto-generated method stub
}
}
해당 커스텀 파티셔너를 사용하려면 Properties에 "partitioner.class"/"package.className" 키/값 형태소 설정해주면 된다.
public KafkaProducer<String, String> getProducer(){
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092,localhost:9093");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("partitioner.class","com.kafka.exam.CustomPartition");
return new KafkaProducer<>(producerProps);
}
추가 프로듀서 설정
- buffer.memory : 카프카 서버로 전송을 기다리는 메시지를 위해 프로듀서가 사용할 버퍼 메모리의 크기다. 간단히 전송되지 않은 메시지를 보관하는 자바 프로듀서가 사용할 전체 메모리다. 이 설정의 한계에 도달하면, 프로듀서는 예외를 발생시키기 전에 max.block.ms 시간 동안 메시지를 대기시킨다. 만약 배치 사이즈가 더 크다면 프로듀서 버퍼에 더 많은 메모리를 할당해야한다. 예를 들어 설명하자면 카프카 서버에 메시지를 전송하는 속도보다 프로듀서가 메시지를 전송하는 요청의 수가 더 빠를 경우(1개씩 메시지를 카프카 서버에 전송하는데 하나를 보내면 2개의 요청이 계속 쌓인다면) 아직 보내지 않은 메시지는 해당 버퍼에 쌓이게 된다. 이 내용에 더해서 큐에 있는 레코드가 계속 남아있는 것을 피하기 위해 request.timeout.ms를 사용해 지정된 시간동안 메시지가 보내지지 않고 큐에 쌓여있다면 메시지는 큐에서 제거되고 예외를 발생시킨다.
- acks : 이 설정은 각 파티션의 리더가 팔로워(복제본)에게 ACK을 받는 전략을 설정한다. ack=0이라면 리더에 데이터가 쓰이자 마자, 팔로워(복제본)가 데이터 로그를 쓰는 것을 신경쓰지 않고 커밋을 완료한다. ack=1 하나의 팔로워에게 ack를 받으면 커밋을 완료한다. ack=all 모든 팔로워에게 복제완료 ack를 받으면 커밋을 완료한다. 각 설정은 장단점이 있으므로 잘 조율해서 사용해야한다. 만약 ack=0이면 빠른 처리 성능을 보장하지만 매우 높은 가능성의 메시지 손실이 있을 수 있고, ack=1도 역시 처리성능은 빠르지만 여전히 메시지 손실가능성이 존재한다. 마지막으로 ack=all은 메시지 손실 가능성이 매우 낮지만 처리 성능이 느려진다.
- batch.size : 이 설정은 설정된 크기만큼 여러 데이터를 모아 배치로 데이터를 전송한다. 하지만 프로듀서는 단위 배치가 꽉 찰때까지 기다릴 필요는 없다. 배치는 특정 주기로 전성되며 배치에 포함된 메시지 수는 상관하지 않는다.
- linger.ms : 브로커로 현재의 배치를 전송하기 전에 프로듀서가 추가 메시지를 기다리는 시간을 나타낸다.
- compression.type : 프로듀서는 브로커에게 압축되지 않은 상태로 메시지를 전송하는 것이 기본이지만, 해당 옵션을 사용해 데이터를 압축해 보내 처리 성능을 높힐 수 있다. 배치처리가 많을 수록 유용한 옵션이다.
- retrites : 메시지 전송이 실패하면 프로듀서가 예외를 발생시키기 전에 메시지의 전송을 다시 시도하는 수.
기타 많은 설정들이 존재하지만 여기서 모두 다루지는 않는다. 공식 홈페이지를 이용해자.
자바 카프카 프로듀서 예제
코드 작성 전에 Maven에 카프카 dependency 설정을 해준다.
public KafkaProducer<String, String> getProducer(){
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092,localhost:9093");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("acks", "all");
producerProps.put("retries", 1);
producerProps.put("batch.size", 20000);
producerProps.put("linger.ms", 1);
producerProps.put("buffer.memory", 24568545);
return new KafkaProducer<>(producerProps);
}
/*
* 동기식
*
* ProducerRecord 생성자 매개변수(*표시는 필수 매개변수)
* *1)토픽이름,2)파티션 번호,3)타임스탬프,4)키,*5)값
*
* 파티션선정
* 1)파티션 번호가 ProducerRecord의 매개변수로 명시되어 있다면 그 파티션으로 보낸다.
* 2)파티션이 지정되지 않고 ProducerRecord 매개변수에 키가 지정된 경우 파티션은 키의 해시값을 사용해 정한다.
* 3)ProducerRecord 매개변수에 키와 파티션 번호 모두 지정되지 않은 경우에는 라운드 로빈 방식으로 정한다.
*/
public void sendMessageSync(String topicName, String data) throws InterruptedException, ExecutionException {
try(KafkaProducer<String, String> producer = getProducer();){
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topicName, data);
Future<RecordMetadata> recordMetadata = producer.send(producerRecord);
RecordMetadata result = recordMetadata.get();
}
}
/*
* 비동기식
*/
public void sendMessageAsync(String topicName, String data) {
try(KafkaProducer<String, String> producer = getProducer();){
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topicName, data);
producer.send(producerRecord, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(exception != null) {
log.info("예외 발생");
/*
* 예외처리
*/
}else {
log.info("정상처리");
/*
* 정상처리 로직
*/
}
}
});
}
}
카프카 프로듀서를 사용할 때 유의해야 하는 사항
- 데이터 유효성 검증 : 보낼 데이터의 유효성 검증은 간과하기 쉽다. 꼭 유효성 검증을 포함하는 로직을 작성하자.
- 예외처리 : 연결 실패,네트워크 시간 초과 등의 예외로는 프로듀서가 내부적으로 재전송을 한다. 하지만 모든 예외에 대해서 재전송을 하는 것은 아니다. 즉, 예외 발생에 대해 처리의 책임은 온전히 프로듀서 애플리케이션에 있는 것이다.
- 부트스트랩 URL 수 : 클러스터 노드수가 적다면 모든 URL를 목록으로 작성하자. 하지만 많약 노드가 아주 많다면 대표하는 브로커의 수만 목록으로 설정해주자.
- 잘못된 파티셔닝 방식의 사용방지 : 파티션은 카프카에서 병렬 처리의 단위다. 메시지가 모든 토픽 파티션으로 균일하게 분배되도록 올바른 파티셔닝 전략을 선택하자. 만약 잘못된 파티셔닝 방식을 도입하게 되면 특정 파티션에만 요청이 집중되는 최악의 상황이 발생할 수 있으니 메시지가 모든 가용한 파티션에 분배 되도록 올바른 파티셔닝 방식을 정의하자.
- 기존 토픽에 새로운 파티션 추가 방지 : 메시지를 분산시키기 위해서 키 값을 사용하여 파티션을 선택하는 토픽에는 파티션을 추가하는 것을 피해야한다. 새로운 파티션 추가는 각 키에 대해 산출된 해시 코드를 변경시키며, 이는 파티션의 수도 산출에 필요한 입력 중 하나로 사용하기 때문이다.
출처: https://coding-start.tistory.com/193?category=790331 [코딩스타트]
'Apache Kafka > Apache Kafka' 카테고리의 다른 글
| kafka 모니터링 도구 (0) | 2021.10.28 |
|---|---|
| Apache Kafka - Kafka(카프카)란 ? 분산 메시징 플랫폼 - 1 (0) | 2021.04.24 |
| Kafka - Kafka Stream API(카프카 스트림즈) - 2 (0) | 2021.04.24 |
| Kafka - Spring Cloud Stream Kafka Streams API(스프링 클라우드 스트림 카프카 스트림즈 API) (0) | 2021.04.24 |
| Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카) (0) | 2021.04.24 |
| Kafka - Kafka Streams API(카프카 스트림즈) - 1 (0) | 2021.04.24 |
| Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI (0) | 2021.04.23 |
| Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI (0) | 2021.04.20 |