全網(wǎng)最詳細(xì)的大數(shù)據(jù)Kafka文章系列，強(qiáng)烈建議收藏加關(guān)注！?

新文章都已經(jīng)列出歷史文章目錄，幫助大家回顧前面的知識重點(diǎn)。

目錄

系列歷史文章

Kafka的java API編寫

一、生產(chǎn)者代碼

第一步: 需求

第二步: 準(zhǔn)備工作

第三步: 代碼開發(fā)

二、消費(fèi)者代碼

---

系列歷史文章

2021年大數(shù)據(jù)Kafka（十二）：??Kafka配額限速機(jī)制??

2021年大數(shù)據(jù)Kafka（十一）：??Kafka的消費(fèi)者負(fù)載均衡機(jī)制和數(shù)據(jù)積壓問題??

2021年大數(shù)據(jù)Kafka（十）：kafka生產(chǎn)者數(shù)據(jù)分發(fā)策略

2021年大數(shù)據(jù)Kafka（九）：kafka消息存儲及查詢機(jī)制原理

2021年大數(shù)據(jù)Kafka（八）：Kafka如何保證數(shù)據(jù)不丟失

2021年大數(shù)據(jù)Kafka（七）：Kafka的分片和副本機(jī)制

2021年大數(shù)據(jù)Kafka（六）：??安裝Kafka-Eagle??

2021年大數(shù)據(jù)Kafka（五）：??Kafka的java API編寫??

2021年大數(shù)據(jù)Kafka（四）：??kafka的shell命令使用??

2021年大數(shù)據(jù)Kafka（三）：??Kafka的集群搭建以及shell啟動命令腳本編寫??

2021年大數(shù)據(jù)Kafka（二）：??Kafka特點(diǎn)總結(jié)和架構(gòu)??

2021年大數(shù)據(jù)Kafka（一）：??消息隊(duì)列和Kafka的基本介紹??

Kafka的java API編寫

一、生產(chǎn)者代碼

第一步: 需求

接下來，編寫Java程序，將1-100的數(shù)字消息寫入到Kafka中

第二步: 準(zhǔn)備工作

1) 創(chuàng)建maven項(xiàng)目 導(dǎo)入相關(guān)的依賴

<repositories><!-- 代碼庫 --><repository> <id>central</id> <url>http://maven.aliyun.com/nexus/content/groups/public//</url><releases> <enabled>true</enabled></releases><snapshots> <enabled>true</enabled> <updatePolicy>always</updatePolicy> <checksumPolicy>fail</checksumPolicy></snapshots></repository>
</repositories>
<dependencies>
<!-- kafka客戶端工具 --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.1</version>
</dependency>
<!-- 工具類 --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-io</artifactId> <version>1.3.2</version>
</dependency>
<!-- SLF橋接LOG4J日志 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.6</version>
</dependency>
<!-- SLOG4J日志 --> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.16</version>
</dependency>
</dependencies>
<build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.7.0</version> <configuration> <source>1.8</source> <target>1.8</target></configuration></plugin></plugins>
</build>

2) 導(dǎo)入log4j.properties 將log4j.properties配置文件放入到resources文件夾中

log4j.rootLogger=INFO,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

3) 創(chuàng)建包和類 創(chuàng)建包c(diǎn)n.it.kafka，并創(chuàng)建KafkaProducerTest類

第三步: 代碼開發(fā)

生產(chǎn)者代碼1: 默認(rèn)異步發(fā)生數(shù)據(jù)方式, 不含回調(diào)函數(shù)

package com.it.producer;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;
// kafka的生產(chǎn)者的代碼:
public class KafkaProducerTest {public static void main(String[] args) {//1.1: 構(gòu)建生產(chǎn)者的配置信息:Properties props = new Properties();props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");props.put("acks", "all"); // 消息確認(rèn)機(jī)制:  all表示 必須等待kafka端所有的副本全部接受到數(shù)據(jù) 確保數(shù)據(jù)不丟失// 說明: 在數(shù)據(jù)發(fā)送的時(shí)候, 可以發(fā)送鍵值對的, 此處是用來定義k v的序列化的類型props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//1. 創(chuàng)建 kafka的生產(chǎn)者對象:  KafkaProducerProducer<String, String> producer = new KafkaProducer<String, String>(props);//2. 執(zhí)行數(shù)據(jù)的發(fā)送for (int i = 0; i < 100; i++) {// producerRecord對象:  生產(chǎn)者的數(shù)據(jù)承載對象ProducerRecord<String, String> producerRecord =new ProducerRecord<String, String>("product-topic",  Integer.toString(i));producer.send(producerRecord);}//3. 釋放資源producer.close();}
}

生產(chǎn)者的代碼2: 同步發(fā)送操作

package com.it.producer;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;
import java.util.concurrent.ExecutionException;// kafka的生產(chǎn)者的代碼:
public class KafkaProducerTest2 {@SuppressWarnings("all")public static void main(String[] args) {//1.1: 構(gòu)建生產(chǎn)者的配置信息:Properties props = new Properties();props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");props.put("acks", "all"); // 消息確認(rèn)機(jī)制:  all表示 必須等待kafka端所有的副本全部接受到數(shù)據(jù) 確保數(shù)據(jù)不丟失// 說明: 在數(shù)據(jù)發(fā)送的時(shí)候, 可以發(fā)送鍵值對的, 此處是用來定義k v的序列化的類型props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//1. 創(chuàng)建 kafka的生產(chǎn)者對象:  KafkaProducerProducer<String, String> producer = new KafkaProducer<String, String>(props);//2. 執(zhí)行數(shù)據(jù)的發(fā)送for (int i = 0; i < 100; i++) {// producerRecord對象:  生產(chǎn)者的數(shù)據(jù)承載對象ProducerRecord<String, String> producerRecord =new ProducerRecord<String, String>("product-topic",  Integer.toString(i));try {producer.send(producerRecord).get(); // get方法, 表示是同步發(fā)送數(shù)據(jù)的方式} catch (Exception e) {// 如果發(fā)生操作, 出現(xiàn)了異常, 認(rèn)為, 數(shù)據(jù)發(fā)生失敗了 ....e.printStackTrace();}}//3. 釋放資源producer.close();}
}

生產(chǎn)者代碼3: 異步發(fā)生數(shù)據(jù), 帶有回調(diào)函數(shù)操作

package com.it.producer;import org.apache.kafka.clients.producer.*;import java.util.Properties;
// kafka的生產(chǎn)者的代碼:
public class KafkaProducerTest {public static void main(String[] args) {//1.1: 構(gòu)建生產(chǎn)者的配置信息:Properties props = new Properties();props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");props.put("acks", "all"); // 消息確認(rèn)機(jī)制:  all表示 必須等待kafka端所有的副本全部接受到數(shù)據(jù) 確保數(shù)據(jù)不丟失// 說明: 在數(shù)據(jù)發(fā)送的時(shí)候, 可以發(fā)送鍵值對的, 此處是用來定義k v的序列化的類型props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//1. 創(chuàng)建 kafka的生產(chǎn)者對象:  KafkaProducerProducer<String, String> producer = new KafkaProducer<String, String>(props);//2. 執(zhí)行數(shù)據(jù)的發(fā)送for (int i = 0; i < 100; i++) {// producerRecord對象:  生產(chǎn)者的數(shù)據(jù)承載對象ProducerRecord<String, String> producerRecord =new ProducerRecord<String, String>("product-topic",  Integer.toString(i));producer.send(producerRecord, new Callback() {public void onCompletion(RecordMetadata metadata, Exception exception) {// 在參數(shù)2中, 表示發(fā)生的狀態(tài)異常, 如果 異常為null 表示數(shù)據(jù)以及發(fā)送成功, 如果不為null, 表示數(shù)據(jù)沒有發(fā)送成功if(exception != null){// 執(zhí)行數(shù)據(jù)發(fā)生失敗的后措施...}}}); // 異步發(fā)送方式}//3. 釋放資源producer.close();}
}

二、消費(fèi)者代碼

消費(fèi)者代碼1: 自動提交偏移量數(shù)據(jù)

package com.it.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;// kafka的消費(fèi)者的代碼
public class KafkaConsumerTest {public static void main(String[] args) {//1.1: 指定消費(fèi)者的配置信息Properties props = new Properties();props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");props.setProperty("group.id", "test"); // 消費(fèi)者組的名稱props.setProperty("enable.auto.commit", "true"); // 消費(fèi)者自定提交消費(fèi)偏移量信息給kafkaprops.setProperty("auto.commit.interval.ms", "1000"); // 每次自動提交偏移量時(shí)間間隔  1s一次props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//1. 創(chuàng)建kafka的消費(fèi)者核心類對象:  KafkaConsumerKafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);//2. 讓當(dāng)前這個(gè)消費(fèi), 去監(jiān)聽那個(gè)topic?consumer.subscribe(Arrays.asList("product-topic")); // 一個(gè)消費(fèi)者 可以同時(shí)監(jiān)聽多個(gè)topic的操作while (true) { // 一致監(jiān)聽//3. 從topic中 獲取數(shù)據(jù)操作:  參數(shù)表示意思, 如果隊(duì)列中沒有數(shù)據(jù), 最長等待多長時(shí)間// 如果超時(shí)后, topic中依然沒有數(shù)據(jù), 此時(shí)返回空的  records(空對象)ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));//4. 遍歷ConsumerRecords, 從中獲取消息數(shù)據(jù)for (ConsumerRecord<String, String> record : records) {String value = record.value();System.out.println("接收到消息為:"+value);}}}
}

消費(fèi)者代碼2: 手動提交偏移量數(shù)據(jù)?

package com.it.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;// kafka的消費(fèi)者的代碼
public class KafkaConsumerTest2 {public static void main(String[] args) {//1.1 定義消費(fèi)者的配置信息Properties props = new Properties();props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");props.setProperty("group.id", "test01"); // 消費(fèi)者組的名稱props.setProperty("enable.auto.commit", "false"); // 消費(fèi)者自定提交消費(fèi)偏移量信息給kafka//props.setProperty("auto.commit.interval.ms", "1000"); // 每次自動提交偏移量時(shí)間間隔  1s一次props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//1. 創(chuàng)建消費(fèi)者的核心類對象KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(props);//2. 指定要監(jiān)聽的topicconsumer.subscribe(Arrays.asList("product-topic"));//3. 獲取數(shù)據(jù)while(true) {ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {String value = consumerRecord.value();// 執(zhí)行消費(fèi)數(shù)據(jù)操作System.out.println("數(shù)據(jù)為:"+ value);// 當(dāng)執(zhí)行完成后, 認(rèn)為消息已經(jīng)消費(fèi)完成consumer.commitAsync(); // 手動提交偏移量信息}}}
}

---

• 📢博客主頁：https://lansonli.blog.csdn.net
• 📢歡迎點(diǎn)贊 👍 收藏 ?留言 📝 如有錯(cuò)誤敬請指正！
• 📢本文由 Lansonli 原創(chuàng)，首發(fā)于 CSDN博客🙉
• 📢大數(shù)據(jù)系列文章會每天更新，停下休息的時(shí)候不要忘了別人還在奔跑，希望大家抓緊時(shí)間學(xué)習(xí)，全力奔赴更美好的生活?

總結(jié)

以上是生活随笔為你收集整理的2021年大数据Kafka（五）：❤️Kafka的java API编写❤️的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。