帶你玩轉Kafka之初步使用

Posted on by

前言

官方文檔：http://kafka.apache.org/

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

Apache Kafka是分佈式發佈-訂閱消息系統。

Apache Kafka與傳統消息系統相比，有以下不同：

  • 它被設計為一個分佈式系統，易於向外擴展；
  • 它同時為發佈和訂閱提供高吞吐量；
  • 它支持多訂閱者，當失敗時能自動平衡消費者；
  • 它將消息持久化到磁盤，因此可用於批量消費，例如ETL，以及實時應用程序。

1 簡單介紹

首先是一些概念:

Kafka作為一個集群，運行在一臺或者多臺服務器上.Kafka 通過 topic 對存儲的流數據進行分類。每條記錄中包含一個key，一個value和一個timestamp（時間戳）。

Kafka有四個核心的API:

The Producer API 允許一個應用程序發佈一串流式的數據到一個或者多個Kafka topic。

The Consumer API 允許一個應用程序訂閱一個或多個 topic ，並且對發佈給他們的流式數據進行處理。

The Streams API 允許一個應用程序作為一個流處理器，消費一個或者多個topic產生的輸入流，然後生產一個輸出流到一個或多個topic中去，在輸入輸出流中進行有效的轉換。

The Connector API 允許構建並運行可重用的生產者或者消費者，將Kafka topics連接到已存在的應用程序或者數據系統。比如，連接到一個關系型數據庫，捕捉表（table）的所有變更內容。

支持的語言(除瞭Java之外的)：

常見概念：

1 Topics和日志

讓我們首先深入瞭解下Kafka的核心概念:提供一串流式的記錄— topic

Topic 就是數據主題，是數據記錄發佈的地方,可以用來區分業務系統。Kafka中的Topics總是多訂閱者模式，一個topic可以擁有一個或者多個消費者來訂閱它的數據。

對於每一個topic， Kafka集群都會維持一個分區日志，如下所示：

每個分區都是有序且順序不可變的記錄集，並且不斷地追加到結構化的commit log文件。分區中的每一個記錄都會分配一個id號來表示順序，我們稱之為offset，offset用來唯一的標識分區中每一條記錄。

Kafka 集群保留所有發佈的記錄—無論他們是否已被消費—並通過一個可配置的參數——保留期限來控制. 舉個例子， 如果保留策略設置為2天，一條記錄發佈後兩天內，可以隨時被消費，兩天過後這條記錄會被拋棄並釋放磁盤空間。Kafka的性能和數據大小無關，所以長時間存儲數據沒有什麼問題.

日志中的 partition（分區）有以下幾個用途。第一，當日志大小超過瞭單臺服務器的限制，允許日志進行擴展。每個單獨的分區都必須受限於主機的文件限制，不過一個主題可能有多個分區，因此可以處理無限量的數據。第二，可以作為並行的單元集—關於這一點，更多細節如下

2 分佈式

日志的分區partition （分佈）在Kafka集群的服務器上。每個服務器在處理數據和請求時，共享這些分區。每一個分區都會在已配置的服務器上進行備份，確保容錯性.

每個分區都有一臺 server 作為 “leader”，零臺或者多臺server作為 follwers 。leader server 處理一切對 partition （分區）的讀寫請求，而follwers隻需被動的同步leader上的數據。當leader宕機瞭，followers 中的一臺服務器會自動成為新的 leader。每臺 server 都會成為某些分區的 leader 和某些分區的 follower，因此集群的負載是平衡的。

3 生產者

生產者可以將數據發佈到所選擇的topic中。生產者負責將記錄分配到topic的哪一個 partition（分區）中。可以使用循環的方式來簡單地實現負載均衡，也可以根據某些語義分區函數(例如：記錄中的key)來完成。下面會介紹更多關於分區的使用。

4 消費者

消費者使用一個 消費組 名稱來進行標識，發佈到topic中的每條記錄被分配給訂閱消費組中的一個消費者實例.消費者實例可以分佈在多個進程中或者多個機器上。

如果所有的消費者實例在同一消費組中，消息記錄會負載平衡到每一個消費者實例.

如果所有的消費者實例在不同的消費組中，每條消息記錄會廣播到所有的消費者進程.

如圖，這個 Kafka 集群有兩臺 server 的，四個分區(p0-p3)和兩個消費者組。消費組A有兩個消費者，消費組B有四個消費者。

通常情況下，每個 topic 都會有一些消費組，一個消費組對應一個”邏輯訂閱者”。一個消費組由許多消費者實例組成，便於擴展和容錯。這就是發佈和訂閱的概念，隻不過訂閱者是一組消費者而不是單個的進程。

在Kafka中實現消費的方式是將日志中的分區劃分到每一個消費者實例上，以便在任何時間，每個實例都是分區唯一的消費者。維護消費組中的消費關系由Kafka協議動態處理。如果新的實例加入組，他們將從組中其他成員處接管一些 partition 分區;如果一個實例消失，擁有的分區將被分發到剩餘的實例。

Kafka 隻保證分區內的記錄是有序的，而不保證主題中不同分區的順序。每個 partition 分區按照key值排序足以滿足大多數應用程序的需求。但如果你需要總記錄在所有記錄的上面，可使用僅有一個分區的主題來實現，這意味著每個消費者組隻有一個消費者進程。

保證

high-level Kafka給予以下保證:

生產者發送到特定topic partition 的消息將按照發送的順序處理。 也就是說，如果記錄M1和記錄M2由相同的生產者發送，並先發送M1記錄，那麼M1的偏移比M2小，並在日志中較早出現一個消費者實例按照日志中的順序查看記錄.對於具有N個副本的主題，我們最多容忍N-1個服務器故障，從而保證不會丟失任何提交到日志中的記錄.

關於保證的更多細節可以看文檔的設計部分。

2 下載安裝

Kafka依賴於Zookeeper，而Zookeeper又依賴於Java，因此在使用Kafka之前要安裝jdk1.8的環境和啟動zookeeper服務器。

下載或安裝地址：

JDK1.8：//www.jb51.net/article/229780.htm：
https://www.jb51.net/article/229783.htm：
https://kafka.apachecn.org/downloads.html

好，下面我們開始進行安裝

[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# tar -zxf kafka_2.11-1.0.0.tgz
[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# mv kafka_2.11-1.0.0 kafka-2.11

3 基本使用

3.1 啟動Kafka

首先檢查下自己的jdk 是否安裝：

[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# java -version
java version "1.8.0_144"
Java(TM) SE Runtime Environment (build 1.8.0_144-b01)
Java HotSpot(TM) 64-Bit Server VM (build 25.144-b01, mixed mode)

啟動Zookeeper：

[root@iZ2ze4m2ri7irkf6h6n8zoZ zookeeper-3.5.9]# ls
bin  conf  docs  lib  LICENSE.txt  NOTICE.txt  README.md  README_packaging.txt
[root@iZ2ze4m2ri7irkf6h6n8zoZ zookeeper-3.5.9]# cd conf/
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# ls
configuration.xsl  log4j.properties  zoo_sample.cfg
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# cp zoo_sample.cfg zoo.cfg
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# cd ../bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ls
README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./zkServer.
zkServer.cmd  zkServer.sh   
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.5.9/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

啟動Kafka：

[root@iZ2ze4m2ri7irkf6h6n8zoZ kafka-2.11]# ls
bin  config  libs  LICENSE  NOTICE  site-docs
[root@iZ2ze4m2ri7irkf6h6n8zoZ kafka-2.11]# cd config/
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# ls
connect-console-sink.properties    connect-file-source.properties  log4j.properties        zookeeper.properties
connect-console-source.properties  connect-log4j.properties        producer.properties
connect-distributed.properties     connect-standalone.properties   server.properties
connect-file-sink.properties       consumer.properties             tools-log4j.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cd ../bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server.properties 
[2021-11-20 10:21:10,326] INFO KafkaConfig values: 
......
[2021-11-20 10:21:12,423] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2021-11-20 10:21:12,423] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser)
[2021-11-20 10:21:12,424] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

3.2 簡單測試使用

新建和查看topic

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic ymx
Created topic "ymx".
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --list --zookeeper localhost:2181
ymx

生產者發送消息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic ymx
>Hello Kafka!
>Hello Ymx!
>Hello Kafka and Ymx!
>

消費者消費消息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ymx --from-beginning
Hello Kafka!
Hello Ymx!
Hello Kafka and Ymx!

3.3 搭建多代理集群

3.3.1 開始搭建

首先要copy下配置文件

[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cp server.properties server-01.properties 
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cp server.properties server-02.properties 
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# vim server-01.properties 
####  內容開始  ####
broker.id=1                  #  21行左右，broker的唯一標識（同一個集群中）
listeners=PLAINTEXT://:9093  #  31行左右，放開，代表kafka的端口號
log.dirs=/tmp/kafka-logs-01  #  60行左右，用逗號分隔的目錄列表，在其中存儲日志文件  
####  內容結束  ####
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# vim server-02.properties 
####  內容開始  ####
broker.id=2                  #  21行左右，broker的唯一標識（同一個集群中）
listeners=PLAINTEXT://:9094  #  31行左右，放開，代表kafka的端口號
log.dirs=/tmp/kafka-logs-02  #  60行左右，用逗號分隔的目錄列表，在其中存儲日志文件  
####  內容結束  ####

根據配置文件啟動Kafka（同一主機下）

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties

報錯信息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties 
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c0000000, 1073741824, 0) failed; error='Cannot allocate memory' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 1073741824 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/kafka-2.11/bin/hs_err_pid4036.log

原因：物理機或虛擬機內存不足，不足以保證Kafka啟動或運行時需要的內容容量

解決方式：

增加物理機或虛擬機的內存

減少Kafka啟動所需內容的配置，將要修改的文件為kafka-server-start.sh

export KAFKA_HEAP_OPTS="-Xmx512M -Xms256M" #29行左右

3.3.2 使用

解決好之後我們開始啟動：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties 
[2021-11-20 10:58:33,138] INFO KafkaConfig values: 

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-02.properties 
[2021-11-20 10:59:04,187] INFO KafkaConfig values:

ps:看下我們的阿裡雲服務器的狀況

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic mr-yan
Created topic "mr-yan".
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic mr-yan
Topic:mr-yan    PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: mr-yan   Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
  • PartitionCount：主題分區數。
  • ReplicationFactor：用來設置主題的副本數。
  • leader：是負責給定分區所有讀寫操作的節點。每個節點都是隨機選擇的部分分區的領導者。
  • replicas：是復制分區日志的節點列表，不管這些節點是leader還是僅僅活著。
  • isr：是一組“同步”replicas，是replicas列表的子集，它活著並被指到leader。

進行集群環境下的使用：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic mr-yan
>Hello Kafkas!  
>Hello Mr.Yan   
>

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic mr-yan
Hello Kafkas!
Hello Mr.Yan

3.3.3 驗證容錯性

首先我們停掉一個Kafka的Broker：

[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# ps -ef|grep server-01.properties
root     19859 28247  1 10:58 pts/3    ../config/server-01.properties
root     23934 16569  0 11:12 pts/11   00:00:00 grep --color=auto server-01.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# kill -9 28247
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# ps -ef|grep server-01.properties
root     32604 16569  0 11:13 pts/11   00:00:00 grep --color=auto server-01.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# cd /usr/local/kafka-2.11/bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic mr-yan
Topic:mr-yan    PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: mr-yan   Partition: 0    Leader: 0       Replicas: 1,0,2 Isr: 0,2

查看生產者和消費者的變化，並再次使用，發現仍可以進行使用

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic mr-yan
>Hello Kafkas!  
>Hello Mr.Yan   
>[2021-11-20 11:12:28,881] WARN [Producer clientId=console-producer] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
>Hello Kafkas too!
>Hello Mr.Yan too!
>

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic mr-yan
Hello Kafkas!
Hello Mr.Yan
[2021-11-20 11:12:28,812] WARN [Consumer clientId=consumer-1, groupId=console-consumer-22158] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2021-11-20 11:12:29,165] WARN [Consumer clientId=consumer-1, groupId=console-consumer-22158] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
Hello Kafkas too!
Hello Mr.Yan too!

4 小總結

主題，分區，副本的概念

Kafka是根據主題（topic）進行消息的傳遞，但是又有分區和副本的概念，下面來分別解釋下：

分區：kafka對每一條消息的key做一個hashcode運算，然後將得到的數值對分區數量進行模運算就得到瞭這條消息所在分區的數字。副本：同一分區的幾個副本之間保存的是相同的數據，副本之間的關系是“一主多從”，其中的主（leader）則負責對外提供讀寫操作的服務，而從（follower）則負責與主節點同步數據，當主節點宕機，從節點之間能重新選舉leader進行對外服務。

kafka會保證同一個分區內的消息有序，但是不保證主題內的消息有序。

參考：https://kafka.apachecn.org/quickstart.html

總結

到此這篇關於帶你玩轉Kafka之初步使用的文章就介紹到這瞭,更多相關Kafka初步使用內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：