Kafka超詳細學習筆記【概念理解，安裝配置】

目錄

官方文檔：//kafka.apache.org/23/documentation.html#introduction

中文文檔：//kafka.apachecn.org/

本篇要點

  1. 介紹kafka的特性、概念、API及專業術語。
  2. 介紹Windows環境下kafka的安裝配置，啟動測試。
  3. Java客戶端連接kafka的案例演示。

Kafka介紹

Apache Kafka 是一個分散式流處理平台：distributed streaming platform。

作為流處理平台的三種特性

  1. 可發布和訂閱消息（流），這與消息隊列或企業消息系統類似。
  2. 以容錯（故障轉移）的方式存儲消息（流）。
  3. 提供實時的流處理。

主要應用

kafka主要應用於兩大類應用：

  1. 構建實時的流數據通道，可靠地獲取系統和應用程式之間的數據。
  2. 構建實時流的應用程式，對數據流進行轉換或反應。

四個核心API

  1. Producer API：發布消息到一個或多個topic主題上。
  2. Consumer API：訂閱一個或多個topic，處理產生的消息。
  3. Streams API：串流處理器，從一個或多個topic消費輸入流，併產生一個輸出流到一個或多個輸出topic，有效地將輸入流轉換到輸出流。
  4. Connector API：可構建或運行可重用地生產者或消費者，將topic連接到現有地應用程式或數據系統。

基本術語

Topic：kafka將消息分類，每一類的消息都有一個主題topic。

Producer：生產者，發布消息的對象。

Consumer：消費者，訂閱消息的對象。

Broker：代理，已發布的消息保存在一組伺服器中，稱之為kafka集群，集群中每個伺服器都是一個代理(broker)。消費者可以訂閱一個或多個主題，並從broker上拉取數據，從而消費這些已發布的消息。

Partition：Topic物理上的分組，一個Topic可以分為多個partition，每個partition都是一個順序的、不可變的消息隊列，且可以持續添加。Partition中的每條消息都會被分配一個有序的序列號，稱為偏移量（offset），因此每個分區中偏移量都是唯一的。

Consumer Group：每個Consumer屬於一個特定的Consumer Group，這是kafka用來實現一個Topic消息的廣播【發送給所有的consumer的發布訂閱式消息模型】和單播【發送給任意一個consumer隊列消息模型】的手段。一個topic可以有多個consumer group。

  • 如果要實現廣播，只要每個consumer有獨立的consumer group就可以，此時就是發布訂閱模型。
  • 如果要實現單播，只要所有的consumer在同一個consumer group中就可以，此時就是隊列模型。

關於Consumer group的補充：一般來說，我們可以創建一些consumer group作為邏輯上的訂閱者，每個組中包含數目不等的consumer，一個組內的多個消費者可以用來擴展性能和容錯。

關於partition分區的補充：

1、【負載均衡】處理更多的消息，不受單台伺服器的限制。

2、【順序保證】kafka不能保證並行的時候消息的有序性，但是可以保證一個partition分區之中，消息只能由消費者組中的唯一一個消費者處理，以保證一個分區的消息先後順序。

如下圖：2個kafka集群託管4個分區(p0-p3)，2個消費者組，組A有2個消費者實例，組B有4個消費者實例。

關於偏移量的補充：kafka集群將會保持所有的消息，直到他們過期，無論他們是否被消費。當消費者消費消息時，偏移量offset將會線性增加，但是消費者其實可以控制實際的偏移量，可以重置偏移量為更早的位置，意為著重新讀取消息，且不會影響其他消費者對此log的處理。

快速開始

安裝配置Zookeeper

Kafka的安裝配置啟動需要依賴於Zookeeper，Zookeeper的安裝配置可以參考我的前一篇文章。

當然，其實你下載kafka之後，就自動已經集成了Zookeeper，你可以通過修改配置，啟動內置的zookeeper。

關於使用內置的Zookeeper還是自己安裝的Zookeeper的區別，可以看看這篇文章：//segmentfault.com/q/1010000021110446

下載kafka

下載地址：//kafka.apache.org/downloads

下載二進位版本【Binary downloads】，下載完成之後，解壓到合適的目錄下。

筆者目錄為：D:\dev\kafka_2.11-2.3.1。

配置文件

進入config目錄下，找到server.properties文件並修改如下：

log.dirs=D:\\dev\\kafka_2.11-2.3.1\\config\\kafka-logs
zookeeper.connect=localhost:2182 # 默認埠是2181，這裡修改為2182

找到zookeeper.properties文件，修改如下：

dataDir=D:\\softs\\zookeeper-3.4.13\\datas 
dataLogDir=D:\\softs\\zookeeper-3.4.13\\logs
clientPort=2182

Windows的命令

在bin目錄下存放著所有可以使用的命令行指令，Linux和Windows的存放目錄需要注意：

啟動Zookeeper

D:\dev\kafka_2.11-2.3.1> .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

啟動Kafka

D:\dev\kafka_2.11-2.3.1> .\bin\windows\kafka-server-start.bat .\config\server.properties

進行測試

創建topic

創建1個分區1個副本，topic為test-topic

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic test-topic
Created topic test-topic.

查看topic

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2182
test-topic

生產者

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test-topic

消費者

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test-topic --from-beginning

生產者與消費者消息傳遞

刪除topic

如果kafka啟動時載入的配置文件中 server.properties 中沒有配置delete.topic.enable=true，則此刪除非真正刪除，而是僅僅將topic標記為marked for deletion

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-topics.bat --delete --zookeeper localhost:2182 --topic test-topic

Topic test-topic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

登錄內置的zookeeper客戶端

D:\dev\kafka_2.11-2.3.1\bin\windows>zookeeper-shell.bat localhost:2182

Connecting to localhost:2182
Welcome to ZooKeeper!
JLine support is disabled

物理刪除topic

ls /brokers
[ids, topics, seqid]
ls /brokers/topics
[test, test-topic, __consumer_offsets]
rmr /brokers/topics/test-topic # 物理刪除 test-topic 
ls /brokers/topics
[test, __consumer_offsets]

Java客戶端使用

引入依賴

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.6.0</version>
        </dependency>

生產者

public class ProducerExample {

    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put("zk.connect", "localhost:2182");
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all"); 
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "test";
        for (int i = 1; i <= 100; i++) {
            // send方法是非同步的 ， 返回Future對象，如果調用get()，將阻塞，直到相關請求完成並返回消息的metadata或拋出異常
            producer.send(new ProducerRecord<>(topic, "key" + i, "msg" + i * 100));
        }
        // 生產者的傳沖空間池保留尚未發送到伺服器的消息，後台I/O執行緒負責將這些消息轉換程請求發送到集群
        // 如果使用後不關閉生產者，將會丟失這些消息。
        producer.close();
    }

}

  • zk.connect：設置zookeeper的地址。

  • bootstrap.servers：用於建立與 kafka 集群連接的 host/port 組。

  • acks：判斷是不是成功發送，指定all將會阻塞消息，這種設置性能最低，但是是最可靠的。

  • retries：如果請求失敗，生產者會自動重試，我們指定是0次，如果啟用重試，則會有重複消息的可能性。

  • batch.size：(生產者)快取每個分區未發送的消息。快取的大小是通過 batch.size 配置指定的。值較大的話將會產生更大的批。並需要更多的記憶體（因為每個「活躍」的分區都有1個緩衝區）。

  • linger.ms：默認緩衝可立即發送，即便緩衝空間還沒有滿，但是，如果你想減少請求的數量，可以設置linger.ms大於0。這將指示生產者發送請求之前等待一段時間，希望更多的消息填補到未滿的批中。這類似於TCP的演算法，例如上面的程式碼段，可能100條消息在一個請求發送，因為我們設置了linger(逗留)時間為1毫秒，然後，如果我們沒有填滿緩衝區，這個設置將增加1毫秒的延遲請求以等待更多的消息。需要注意的是，在高負載下，相近的時間一般也會組成批，即使是 linger.ms=0。在不處於高負載的情況下，如果設置比0大，以少量的延遲代價換取更少的，更有效的請求。

  • buffer.memory：控制生產者可用的快取總量，如果消息發送速度比其傳輸到伺服器的快，將會耗盡這個快取空間。當快取空間耗盡，其他發送調用將被阻塞，阻塞時間的閾值通過max.block.ms設定，之後它將拋出一個TimeoutException。

  • key.serializer：用於序列化。

  • value.serializer：用於序列化。

消費者

public class ConsumerSample {

    public static void main(String[] args) {
        String topic = "test";// topic name

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "testGroup1");
        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");
        Consumer<String, String> consumer = new KafkaConsumer(props);
        // 訂閱多個主題
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            // 訂閱一組topic之後，調用poll時，消費者將自動加入到組中。
            // 只要持續調用poll，消費者將一直保持可用，並繼續從分配的分區中接收消息。
            // 消費者向伺服器定時發送心跳，如果在session.timeout.ms配置的時間內無法發送心跳，被視為死亡，分區將重新分配
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("*****************partition = %d, offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
        }
    }
}
  • bootstrap.servers：用於建立與 kafka 集群連接的 host/port 組。
  • group.id：消費者的組名，組名相同的消費者被視為同一個消費組。
  • enable.auto.commit：設置Consumer 的 offset 是否自動提交。
  • auto.commit.interval.ms：上面屬性設置為true，由本屬性設置自動提交 offset 到 zookeeper 的時間間隔，時間是毫秒
  • key.deserializer：用於反序列化。
  • value.deserializer：用於反序列化。

Kafka通過進程池瓜分消息並處理消息，這些進程可以在同一台機器運行，也可以分布到多台機器上，以增加可擴展型和容錯性，相同的group.id的消費者將視為同一個消費者組。

組中的每個消費者都通過subscribe API動態的訂閱一個topic列表。kafka將已訂閱topic的消息發送到每個消費者組中。並通過平衡分區在消費者分組中所有成員之間來達到平均。因此每個分區恰好地分配1個消費者（一個消費者組中）。所有如果一個topic有4個分區，並且一個消費者分組有隻有2個消費者。那麼每個消費者將消費2個分區。

消費者組的成員是動態維護的：如果一個消費者故障。分配給它的分區將重新分配給同一個分組中其他的消費者。同樣的，如果一個新的消費者加入到分組，將從現有消費者中移一個給它。這被稱為重新平衡分組。

啟動Zookeeper和kafka

創建topic

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic test

啟動zookeeper

D:\dev\kafka_2.11-2.3.1>.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

啟動kafka

D:\dev\kafka_2.11-2.3.1>.\bin\windows\kafka-server-start.bat .\config\server.properties

測試

先啟動消費者ConsumerExample，再啟動生產者ProducerExample，觀察控制台。

總結

  • kafka作為一個消息系統，它設計了partition分區，提供了負載均衡能力，保證了消息分區內的順序。
  • kafka擁有消費者組的概念，很好地實現發布訂閱和隊列式的消息模型。
  • kafka作為一個存儲系統，高性能，低延遲。
  • kafka能夠提供實時的流處理，提供強大的StreamsAPI，而不是簡單的讀寫和存儲。

參考閱讀

Tags:

VirMach 便宜 VPS

QNews

QNews