Kafka: Producer

xellos·2022년 7월 9일

kafka

Kafka

목록 보기

2/3

Kafka Producer의 역할

Topic에 해당하는 메시지를 생성
특정 Topic으로 데이터를 publish
처리 실패 / 재시도

1) 코드에서 살펴보기

의존성 추가: Gradle

주의사항: Broker 버전과 Client 버전의 하위 호환이 완벽하지 않으므로 둘의 버전을 맞추는 것이 좋다.

implementation group: 'org.apache.kafka', name: 'kafka-clients', version: '3.1.0'

코드 작성

Kafka 에서는 되도록 2개 이상의 IP 와 PORT 를 설정하도록 권장하고 있다.
key는 메시지를 보낼때 토픽의 파티션이 지정될 때 쓰인다.
Kafka는 Key를 해시화해서 각 파티션과 1:1 로 매칭시킨다.

//키가 없는 경우
public class Producer {
	public static void main(String[] args) throws IOException {
    	Properties configs = new Properties();
        configs.put("bootstrap.server", "localhost:9092");
        configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
       	configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        KafkaProducer <String, String> producer = new KakfkaProducer<>(configs);
        ProducerRecord record = new ProducerRecord<String, String>("click_log", "login");
        
        producer.send(record);
        producer.close();
    }
}

//키가 있는 경우
public class Producer {
	public static void main(String[] args) throw IOException {
    	Properties configs = new Properties();
        configs.put("bootstrap.server", "localhost:9092");
        configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
       	configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        KafkaProducer <String, String> producer = new KakfkaProducer<>(configs);
        ProducerRecord record1 = new ProducerRecord<String, String>("click_log", "1", "buy");
        ProducerRecord record2 = new ProducerRecord<String, String>("click_log", "2", "login");
        
        producer.send(record1);
        producer.send(record2);
        producer.close();
    }
}