系列歷史文章

2021年大資料Kafka（五）：??Kafka的java API撰寫??

2021年大資料Kafka（四）：??kafka的shell命令使用??

2021年大資料Kafka（三）：??Kafka的集群搭建以及shell啟動命令腳本撰寫??

2021年大資料Kafka（二）：??Kafka特點總結和架構??

2021年大資料Kafka（一）：??訊息佇列和Kafka的基本介紹??

Kafka的java API撰寫

一、生產者代碼

第一步: 需求

接下來，撰寫Java程式，將1-100的數字訊息寫入到Kafka中

第二步: 準備作業

1) 創建maven專案匯入相關的依賴

<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) 匯入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) 創建包和類

創建包cn.it.kafka，并創建KafkaProducerTest類

第三步: 代碼開發

生產者代碼1: 默認異步發生資料方式, 不含回呼函式

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的生產者的代碼:
public class KafkaProducerTest {

    public static void main(String[] args) {

        //1.1: 構建生產者的配置資訊:
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.put("acks", "all"); // 訊息確認機制:  all表示 必須等待kafka端所有的副本全部接受到資料 確保資料不丟失
        // 說明: 在資料發送的時候, 可以發送鍵值對的, 此處是用來定義k v的序列化的型別
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        //1. 創建 kafka的生產者物件:  KafkaProducer
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        //2. 執行資料的發送
        for (int i = 0; i < 100; i++) {
            // producerRecord物件:  生產者的資料承載物件

            ProducerRecord<String, String> producerRecord =
                    new ProducerRecord<String, String>("product-topic",  Integer.toString(i));

            producer.send(producerRecord);

        }

        //3. 釋放資源
        producer.close();


    }
}

生產者的代碼2: 同步發送操作

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的生產者的代碼:
public class KafkaProducerTest2 {
    @SuppressWarnings("all")
    public static void main(String[] args) {

        //1.1: 構建生產者的配置資訊:
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.put("acks", "all"); // 訊息確認機制:  all表示 必須等待kafka端所有的副本全部接受到資料 確保資料不丟失
        // 說明: 在資料發送的時候, 可以發送鍵值對的, 此處是用來定義k v的序列化的型別
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        //1. 創建 kafka的生產者物件:  KafkaProducer
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        //2. 執行資料的發送
        for (int i = 0; i < 100; i++) {
            // producerRecord物件:  生產者的資料承載物件

            ProducerRecord<String, String> producerRecord =
                    new ProducerRecord<String, String>("product-topic",  Integer.toString(i));

            try {
                producer.send(producerRecord).get(); // get方法, 表示是同步發送資料的方式
            } catch (Exception e) {
                // 如果發生操作, 出現了例外, 認為, 資料發生失敗了 ....

                e.printStackTrace();
            }

        }

        //3. 釋放資源
        producer.close();


    }
}

生產者代碼3: 異步發生資料, 帶有回呼函式操作

package com.it.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
// kafka的生產者的代碼:
public class KafkaProducerTest {

    public static void main(String[] args) {

        //1.1: 構建生產者的配置資訊:
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.put("acks", "all"); // 訊息確認機制:  all表示 必須等待kafka端所有的副本全部接受到資料 確保資料不丟失
        // 說明: 在資料發送的時候, 可以發送鍵值對的, 此處是用來定義k v的序列化的型別
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        //1. 創建 kafka的生產者物件:  KafkaProducer
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        //2. 執行資料的發送
        for (int i = 0; i < 100; i++) {
            // producerRecord物件:  生產者的資料承載物件

            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) {
                    
                    // 在引數2中, 表示發生的狀態例外, 如果 例外為null 表示資料以及發送成功, 如果不為null, 表示資料沒有發送成功
                    if(exception != null){

                        // 執行資料發生失敗的后措施...
                        
                    }
                    
                }
            }); // 異步發送方式

        }

        //3. 釋放資源
        producer.close();


    }
}

二、消費者代碼

消費者代碼1: 自動提交偏移量資料

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的消費者的代碼
public class KafkaConsumerTest {

    public static void main(String[] args) {
        //1.1: 指定消費者的配置資訊
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.setProperty("group.id", "test"); // 消費者組的名稱
        props.setProperty("enable.auto.commit", "true"); // 消費者自定提交消費偏移量資訊給kafka
        props.setProperty("auto.commit.interval.ms", "1000"); // 每次自動提交偏移量時間間隔  1s一次
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //1. 創建kafka的消費者核心類物件:  KafkaConsumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //2. 讓當前這個消費, 去監聽那個topic?
        consumer.subscribe(Arrays.asList("product-topic")); // 一個消費者 可以同時監聽多個topic的操作
        while (true) { // 一致監聽
            //3. 從topic中 獲取資料操作:  引數表示意思, 如果佇列中沒有資料, 最長等待多長時間
            // 如果超時后, topic中依然沒有資料, 此時回傳空的  records(空物件)
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            //4. 遍歷ConsumerRecords, 從中獲取訊息資料
            for (ConsumerRecord<String, String> record : records) {

                String value = record.value();

                System.out.println("接收到訊息為:"+value);

            }

        }
    }
}

消費者代碼2: 手動提交偏移量資料

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的消費者的代碼
public class KafkaConsumerTest2 {

    public static void main(String[] args) {

        //1.1 定義消費者的配置資訊
        Properties props = new Properties();

        props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.setProperty("group.id", "test01"); // 消費者組的名稱
        props.setProperty("enable.auto.commit", "false"); // 消費者自定提交消費偏移量資訊給kafka
        //props.setProperty("auto.commit.interval.ms", "1000"); // 每次自動提交偏移量時間間隔  1s一次
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");


        //1. 創建消費者的核心類物件
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(props);

        //2. 指定要監聽的topic
        consumer.subscribe(Arrays.asList("product-topic"));

        //3. 獲取資料
        while(true) {

            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {

                String value = consumerRecord.value();
                // 執行消費資料操作
                System.out.println("資料為:"+ value);

                // 當執行完成后, 認為訊息已經消費完成

                consumer.commitAsync(); // 手動提交偏移量資訊
            }

        }

    }
}

📢博客主頁：https://lansonli.blog.csdn.net
📢歡迎點贊 👍 收藏 ?留言 📝 如有錯誤敬請指正！
📢本文由 Lansonli 原創，首發于 CSDN博客🙉
📢大資料系列文章會每天更新，停下休息的時候不要忘了別人還在奔跑，希望大家抓緊時間學習，全力奔赴更美好的生活?

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/293318.html

標籤：其他

上一篇：AWS強烈反擊Elastic，欲打造自己的Elasticsearch開源產品OpenSearch

下一篇：解決Word中出現的多級串列編號不顯示的問題