Kafka Javaチュートリアル -プロデューサー-

※Kafkaに関する情報は下記のURLから参照してください。
https://kafka.apache.org/intro

必要ライブラリ

• kafka-clients-3.2.0
• slf4j-api-1.7.36
• slf4j-simple-1.7.36

事前準備

• Kafkaの起動
  下記のコマンドで立ち上げを行います。
  <kafka-home>/bin/zookeeper-server-start.sh <kafka-home>/config/zookeeper.properties
• Kafka brokcerの起動
  下記のコマンドで立ち上げを行います。
  <kafka-home>/bin/kafka-server-start.sh <kafka-home>/config/server.properties

Coding

1. Kafkaのプロパティを定義します。

    final Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
    		BOOTSTRAP_SERVERS);
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    		"my-transactional-id");
   

   ProducerConfig.BOOTSTRAP_SERVERS_CONFIGには接続するKafkaサーバの[URL:PORT]を記載します。
   ProducerConfig.TRANSACTIONAL_ID_CONFIGにはトランザクションのIDを記載します。今回は任意の値で使います。

2. Producerを定義し、トランザクションを初期化します。

    Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
    producer.initTransactions();
   

3. トランザクション処理を開始します。

    producer.beginTransaction();
   

4. Kafkaへイベントを送信します。

    producer.send(new ProducerRecord<>(TOPIC, Integer.toString(i), "Java Producer Pub"));
   

今回のTOPIC、値については任意で設定し、行います。

5. 4の処理が問題なく、終わったらコミット、接続をクローズします。

	 producer.commitTransaction();
	 producer.close();		

参考

下記のレポジトリにソースコードを確認いただけます。
https://github.com/dlstjq7685/KafkaSimpleProducerTuttorial

ソースコードはkafkaのJavaDocの内容に沿って作成しております。
あくまで参考程度で書いたものですので、
その他の設定については別記事を参考にしてください。