上篇blog安裝了可視化的監控工具后，就到了我們最常用的環節，也就是通過代碼來控制Kafka，使用API來呼叫，Kafka檔案地址為Kafka官方檔案，接下來我們會充分使用到官方檔案中的示例，本篇blog分為如下幾個部分：

環境準備：創建一個java project，用來進行kafka代碼的撰寫
生產者API：探討生產者的發送方式，使用不同的生產者介面發送【同步發送、異步發送】
消費者API：探討生產者的發送方式，使用不同的生產者介面發送【offset提交】

接下來按照如下流程來一起學習吧，奧利給！

環境準備

首先新建一個java project，打開idea新建一個maven專案：
在這里插入圖片描述
然后引入kafka的的maven依賴：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>kafka</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>7</source>
                    <target>7</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.6.0</version>
    </dependency>
     <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.5</version>
     </dependency>

    </dependencies>
</project>

生產者API

在官方檔案中，我們可以看到Kafka的消費者API串列生產者API，這些都是當前Kafka支持的生產者相關的API，有如下四種構造方法：
在這里插入圖片描述

也有如下13種方法【非抽象的實體方法】：接下來分成幾個模式分別介紹下
在這里插入圖片描述

發送方式

發送方式分為兩種，同步發送和異步發送，主體的發送流程二者是相同的，主體流程如下：

首先創建ProducerRecord物件，此物件除了包括需要發送的資料value之外還必須指定topic，另外也可以指定key和磁區，當發送ProducerRecord的時候，生產者做的第一件事就是把key和value序列化為ByteArrays，以便它們可以通過網路發送，
接下來，資料會被發送到磁區器，如果在ProducerRecord中指定了一個磁區，那么磁區器會直接回傳指定的磁區；否則，磁區器通常會基于ProducerRecord的key值計算出一個磁區，一旦磁區被確定，生產者就知道資料會被發送到哪個topic和磁區，然后資料會被添加到同一批發送到相同topic和磁區的資料里面，一個單獨的執行緒會負責把那些批資料發送到對應的brokers，
當broker接收到資料的時候，如果資料已被成功寫入到Kafka，會回傳一個包含topic、磁區和偏移量offset的RecordMetadata物件；如果broker寫入資料失敗，會回傳一個例外資訊給生產者，當生產者接收到例外資訊時會嘗試重新發送資料，如果嘗試失敗則拋出例外，

Kafka 的 Producer 發送訊息采用的是異步發送的方式，在訊息發送的程序中，涉及到了兩個執行緒 ——main 執行緒和 Sender 執行緒，以及一個執行緒共享變數 ——RecordAccumulator，main 執行緒將訊息發送給RecordAccumulator，Sender 執行緒不斷從 RecordAccumulator 中拉取訊息發送到 Kafka broker
在這里插入圖片描述
只有資料積累到 batch.size 之后，sender 才會發送資料，如果資料遲遲未達到 batch.size，sender 等待 linger.time 之后就會發送資料，也就是發往broker的資料是一批一批過去的，

異步發送

異步發送的含義是：訊息的發送者只是將訊息發送過去，并不關心訊息的發送狀態，如果leader在發送ack后宕機的話，重復發送的訊息將不能保證原來的順序，最好選用帶回呼函式的方法，

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class Producer {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //設定kafka集群的地址
        props.put("bootstrap.servers", "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        //ack模式，all是最慢但最安全的
        props.put("acks", "-1");
        //失敗重試次數
        props.put("retries", 1);
        //每個磁區未發送訊息總位元組大小（單位：位元組），超過設定的值就會提交資料到服務端
        props.put("batch.size", 10);
        //props.put("max.request.size",10);
        //訊息在緩沖區保留的時間，超過設定的值就會被提交到服務端
        props.put("linger.ms", 10000);
        //整個Producer用到總記憶體的大小，如果緩沖區滿了會提交資料到服務端
        //buffer.memory要大于batch.size，否則會報申請記憶體不足的錯誤
        props.put("buffer.memory", 10240);
        //序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        org.apache.kafka.clients.producer.Producer producer=new KafkaProducer(props);
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<String, String>("tml-second", Integer.toString(i), "tml-second訊息:"+i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("訊息發送狀態監測");
                }
            });
        producer.close();
    }
}

我們可以從機器上看到訊息記錄
在這里插入圖片描述
為了更準確一些，我們用命令消費一下：

同步發送

同步發送用的比較少，唯一的不同就是他要求發送時按照順序，如果當條資料發送失敗，那么就阻塞執行緒，這樣就保證了訊息的嚴格順序【即使在重試狀態下發送的訊息】

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class Producer {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        //設定kafka集群的地址
        props.put("bootstrap.servers", "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        //ack模式，all是最慢但最安全的
        props.put("acks", "-1");
        //失敗重試次數
        props.put("retries", 1);
        //每個磁區未發送訊息總位元組大小（單位：位元組），超過設定的值就會提交資料到服務端
        props.put("batch.size", 10);
        //props.put("max.request.size",10);
        //訊息在緩沖區保留的時間，超過設定的值就會被提交到服務端
        props.put("linger.ms", 10000);
        //整個Producer用到總記憶體的大小，如果緩沖區滿了會提交資料到服務端
        //buffer.memory要大于batch.size，否則會報申請記憶體不足的錯誤
        props.put("buffer.memory", 10240);
        //序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        org.apache.kafka.clients.producer.Producer producer=new KafkaProducer(props);
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<String, String>("tml-second", Integer.toString(i), "tml-second訊息:"+i)).get();
        producer.close();
    }
}

防止訊息重復提交

在生產者策略的時候我們提到過，需要防止訊息重復提交，也即精準一次提交，我們有兩種級別，一種是冪等模式【一個broker的會話周期精準一次】，另一種是事務模式【全域的精準一次】，

冪等模式

代碼寫法類似，只需要給配置里加一個配置項

 //冪等模式
 props.put("enable.idempotence", true);

一旦設定了該屬性，那么retries默認是Integer.MAX_VALUE ，acks默認是all【-1】，

事務模式

事務模式的寫法略有不同：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer {
    public static void main(String[] args)  {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        props.put("transactional.id", "my_transactional_id");
        org.apache.kafka.clients.producer.Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
        producer.initTransactions();
        try {
            //資料發送必須在beginTransaction()和commitTransaction()中間，否則會報狀態不對的例外
            producer.beginTransaction();
            for (int i = 0; i < 100; i++)
                producer.send(new ProducerRecord<>("tml-second", Integer.toString(i), Integer.toString(i)));
            producer.commitTransaction();
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            // 這些例外不能被恢復，因此必須要關閉并退出Producer
            producer.close();
        } catch (KafkaException e) {
            // 出現其它例外，終止事務
            producer.abortTransaction();
        }
        producer.close();
    }
}

消費者API

在官方檔案中，我們可以看到Kafka的消費者API串列消費者API，有構造方法，和實體方法，構造方法有如下四種：
加粗樣式
也有45種方法【非抽象的實體方法】以及4種棄用方法，消費者提交方式有以下幾種：

自動提交：kafka管理offset的提交
手動提交：手動同步提交和手動異步提交

按照這種結構我們看下提交方式，

自動提交offset

提交的代碼如下：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class Consumer {
    public static void main(String[] args)  {
        Properties props = new Properties();
        //設定kafka集群的地址
        props.put("bootstrap.servers", "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        //設定消費者組，組名字自定義，組名字相同的消費者在一個組
        props.put("group.id", "tml-group");
        //開啟offset自動提交
        props.put("enable.auto.commit", "true");
        //自動提交時間間隔
        props.put("auto.commit.interval.ms", "1000");
        //序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //實體化一個消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //消費者訂閱主題，可以訂閱多個主題
        consumer.subscribe(Arrays.asList("tml-second"));
        //死回圈不停的從broker中拿資料
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }

}

可以看到提交的效果
在這里插入圖片描述

手動同步提交offset

通常從Kafka拿到的訊息是要做業務處理，而且業務處理完成才算真正消費成功，所以需要客戶端控制offset提交時間

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class Consumer {
    public static void main(String[] args)  {
        Properties props = new Properties();
        //設定kafka集群的地址
        props.put("bootstrap.servers",  "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        //設定消費者組，組名字自定義，組名字相同的消費者在一個組
        props.put("group.id", "tml_group");
        //開啟offset自動提交
        props.put("enable.auto.commit", "false");
        //序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //實體化一個消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //消費者訂閱主題，可以訂閱多個主題
        consumer.subscribe(Arrays.asList("tml-second"));
        final int minBatchSize = 50;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                //insertIntoDb(buffer);
                for (ConsumerRecord bf : buffer) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", bf.offset(), bf.key(), bf.value());
                }
                consumer.commitSync();
                buffer.clear();
            }
        }
    }

}

手動異步提交offset

雖然同步提交 offset 更可靠一些，但是由于其會阻塞當前執行緒，直到提交成功，因此吞吐量會收到很大的影響，因此更多的情況下，會選用異步提交 offset 的方式

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class Consumer {
    public static void main(String[] args)  {
        Properties props = new Properties();
        //設定kafka集群的地址
        props.put("bootstrap.servers",  "192.168.5.101:9092,192.168.5.102:9092,192.168.5.103:9092");
        //設定消費者組，組名字自定義，組名字相同的消費者在一個組
        props.put("group.id", "tml_group");
        //開啟offset自動提交
        props.put("enable.auto.commit", "false");
        //序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //實體化一個消費者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //消費者訂閱主題，可以訂閱多個主題
        consumer.subscribe(Arrays.asList("tml-second"));
        final int minBatchSize = 50;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                //insertIntoDb(buffer);
                for (ConsumerRecord bf : buffer) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", bf.offset(), bf.key(), bf.value());
                }
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public  void  onComplete(Map<TopicPartition,
                                                OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit  failed  for"  +
                                    offsets);
                        }
                    }
                });
                buffer.clear();
            }
        }
    }

}

趟了無數的坑，終于把Kafka學習完了，接下來開始Redis之旅，開始由業務架構向基礎架構滲透，上可接客戶，中可玩兒平臺，下可探基礎，完成SaaS、PaaS以及IaaS的倍訓

部分內容來自 https://blog.csdn.net/wangzhanzheng/article/details/80801059

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

標籤：java

上一篇：java中public類問題

下一篇：爬梯：七大設計原則

【Kafka從入門到放棄系列 八】Kafka的API呼叫