簡介

kafka是一個分布式消息隊列。具有高性能、持久化、多副本備份、橫向擴(kuò)展能力。生產(chǎn)者往隊列里寫消息，消費者從隊列里取消息進(jìn)行業(yè)務(wù)邏輯。一般在架構(gòu)設(shè)計中起到解耦、削峰、異步處理的作用。

Kafka核心組件-intsmaze

• 　　Topic：消息根據(jù)Topic進(jìn)行歸類，可以理解為一個隊里。
• 　　Producer：消息生產(chǎn)者，就是向kafka broker發(fā)消息的客戶端。
• 　　Consumer：消息消費者，向kafka broker取消息的客戶端。
• 　　broker：每個kafka實例(server)，一臺kafka服務(wù)器就是一個broker，一個集群由多個broker組成，一個broker可以容納多個topic。

• 　　Zookeeper：依賴集群保存meta信息。
  　　
  大家先看kafka的介紹或者教程啊，上來都顯示一堆長篇大論，各自文字圖片，看著很懵逼，頭暈。搞程序的，要讓ta跑起來，再針對可運行的成果，慢慢了解ta。所以本文會由淺入深，先實踐后理論，結(jié)合實踐講理論。

  Kafka安裝配置

  下載

wget http://mirror.bit.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz

解壓

tar -zxvf kafka_2.11-2.2.0.tgz

修改 kafka-server 的配置文件

cd kafka_2.11-2.2.0vim config/server.properties

修改其中的：

# The id of the broker. This must be set to a unique integer for each broker. broker.id=1 # A comma separated list of directories under which to store log files log.dirs=/data/kafka-logs

啟動zk【默認(rèn)端口2181】

bin/zookeeper-server-start.sh config/zookeeper.properties

啟動Kafka

使用?kafka-server-start.sh?啟動 kafka 服務(wù)：

bin/kafka-server-start.sh config/server.properties

測試使用

創(chuàng)建 topic

使用 kafka-topics.sh 創(chuàng)建單分區(qū)單副本的 topic demo：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demo

查看 topic 列表：

bin/kafka-topics.sh --list --zookeeper localhost:2181

發(fā)送消息【生產(chǎn)者】

使用?kafka-console-producer.sh?發(fā)送消息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic demo

讀取消息【消費者】

使用?kafka-console-consumer.sh 接收消息并在終端打印：

?bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --from-beginning

注意不要使用
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning，高版本已經(jīng)不支持

查看描述 topics 信息

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic demo [root@localhost kafka_2.11-2.2.0]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic demo Topic:demo PartitionCount:1 ReplicationFactor:1 Configs:Topic: demo Partition: 0 Leader: 1 Replicas: 1 Isr: 1

第一行給出了所有分區(qū)的摘要，每個附加行給出了關(guān)于一個分區(qū)的信息。 由于我們只有一個分區(qū)，所以只有一行。

• “Leader”: 是負(fù)責(zé)給定分區(qū)的所有讀取和寫入的節(jié)點。 每個節(jié)點將成為分區(qū)隨機(jī)選擇部分的領(lǐng)導(dǎo)者。
• “Replicas”: 是復(fù)制此分區(qū)日志的節(jié)點列表，無論它們是否是領(lǐng)導(dǎo)者，或者即使他們當(dāng)前處于活動狀態(tài)。
• “Isr”: 是一組“同步”副本。這是復(fù)制品列表的子集，當(dāng)前活著并被引導(dǎo)到領(lǐng)導(dǎo)者

擴(kuò)展-集群配置

Kafka 支持兩種模式的集群搭建：可以在單機(jī)上運行多個 broker 實例來實現(xiàn)集群，也可在多臺機(jī)器上搭建集群，下面介紹下如何實現(xiàn)單機(jī)多 broker 實例集群，其實很簡單，只需要如下配置即可。

單機(jī)多broker 集群配置

利用單節(jié)點部署多個 broker。 不同的 broker 設(shè)置不同的 id，監(jiān)聽端口及日志目錄。 例如：

cp config/server.properties config/server-2.properties vi config/server-2.properties

修改內(nèi)容：

broker.id=2listeners = PLAINTEXT://127.0.0.1:9093log.dirs=/data/kafka-logs2

同樣，配置第三個broker：

cp config/server-2.properties config/server-3.properties vi config/server-3.properties

修改內(nèi)容：

broker.id=2listeners = PLAINTEXT://127.0.0.1:9093log.dirs=/data/kafka-logs2

listeners 申明此kafka服務(wù)器需要監(jiān)聽的端口號，默認(rèn)會使用localhost的地址，如果是在遠(yuǎn)程服務(wù)器上運行則必須配置，例如：　　　　　　　　　
listeners=PLAINTEXT:// 192.168.180.128:9092
并確保服務(wù)器的9092端口能夠訪問

啟動2/3 borker

bin/kafka-server-start.sh config/server-2.properties & bin/kafka-server-start.sh config/server-3.properties &

至此，單機(jī)多broker實例的集群配置完畢。

擴(kuò)展-多機(jī)多borker集群

分別在多個節(jié)點按上述方式安裝 Kafka，配置啟動多個 Zookeeper 實例。

假設(shè)三臺機(jī)器 IP 地址是 ： 192.168.153.135， 192.168.153.136， 192.168.153.137

分別配置多個機(jī)器上的 Kafka 服務(wù)，設(shè)置不同的 broker id，zookeeper.connect 設(shè)置如下:

config/server.properties里面的?zookeeper.connect

zookeeper.connect=192.168.153.135:2181,192.168.153.136:2181,192.168.153.137:2181

使用 Kafka Connect 來導(dǎo)入/導(dǎo)出數(shù)據(jù)

從控制臺寫入數(shù)據(jù)并將其寫回控制臺是一個方便的起點，但您可能想要使用其他來源的數(shù)據(jù)或?qū)?shù)據(jù)從 Kafka 導(dǎo)出到其他系統(tǒng)。對于許多系統(tǒng)，您可以使用 Kafka Connect 來導(dǎo)入或?qū)С鰯?shù)據(jù)，而不必編寫自定義集成代碼。

Kafka Connect 是 Kafka 包含的一個工具，可以將數(shù)據(jù)導(dǎo)入和導(dǎo)出到 Kafka。它是一個可擴(kuò)展的工具，運行 連接器，實現(xiàn)與外部系統(tǒng)交互的自定義邏輯。在這個快速入門中，我們將看到如何使用簡單的連接器運行 Kafka Connect，這些連接器將數(shù)據(jù)從文件導(dǎo)入到 Kafka topic，并將數(shù)據(jù)從 Kafka topic 導(dǎo)出到文件。

參考：

• http://www.54tianzhisheng.cn/2018/01/04/Kafka/
• http://kafka.apache.org/10/documentation/streams/quickstart
• http://kafka.apache.org/20/documentation.html#quickstart

代碼測試

準(zhǔn)備測試kafka

cp config/server.properties config/server-idea.properties vi config/server-idea.propertiesbroker.id=999listeners = PLAINTEXT://192.168.1.177:9999log.dirs=/data/kafka-logs-999

192.168.1.177為kafka所在機(jī)器的ip地址，9999端口號是對外提供的端口，下文會使用到

Springboot 發(fā)送消息、接受消息源碼

很簡單的一個小demo，可以直接拷貝使用。

KafkaApplication.java:

import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.KafkaProducer; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.core.KafkaTemplate;@SpringBootApplication public class KafkaApplication {public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(KafkaApplication.class, args);KafkaTemplate kafkaTemplate = context.getBean(KafkaTemplate.class);for (int i = 0; i < 10; i++) {//調(diào)用消息發(fā)送類中的消息發(fā)送方法kafkaTemplate.send("mytopic", System.currentTimeMillis() + "發(fā)送" + i);try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}}}@KafkaListener(topics = {"mytopic"},groupId = "halburt-demo2")public void consumer1(String message) {System.out.println("consumer1收到消息:" + message);}@KafkaListener(topics = {"mytopic"} ,groupId = "halburt-demo")public void consumer2(ConsumerRecord<?, ?> record) {System.out.println("consumer2收到消息");System.out.println(" topic" + record.topic());System.out.println(" key:" + record.key());System.out.println(" value:"+record.value());} }

application.yml:

server:port: 8090 spring:kafka:consumer:auto-commit-interval: 100bootstrap-servers: 192.168.1.177:9999enable-auto-commit: truegroup-id: halburt-demokey-deserializer: org.apache.kafka.common.serialization.StringDeserializermax-poll-records: 1value-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:concurrency: 5producer:bootstrap-servers: 192.168.1.177:9999key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

192.168.1.177:9999即為kafka的配置文件中配置

pom.xml依賴:

依賴版本：

spring-boot.version：2.1.3.RELEASE
spring-kafka.version：2.2.0.RELEASE

【此處有坑】此處依賴版本可以不用這2個版本，但是一定要注意springboot和kafka的版本對應(yīng)

<dependencies><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.2.0.RELEASE</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.1.3.RELEASE</version></dependency></dependencies>

啟動kafka并run Application.java

bin/kafka-server-start.sh config/server-idea.properties &

已經(jīng)啟動了zk，此處不用再啟動，如果未啟動，需要啟動zk。

cd /home/hd/kafka_2.11-2.2.0/ bin/zookeeper-server-start.sh config/zookeeper.properties&

kafka啟動成功之后，run Application，會看到日志如下：

已經(jīng)接收到消息了。

如果你是跟著本文從頭開始的，一定注意此處有個坑

如果你是從頭開始跟這個本文學(xué)習(xí)的，那么你直接啟動的話，會發(fā)現(xiàn)消息發(fā)出去了，但是沒有接收到。
我也是查了好久，看了很多教程，別人都行我就不行。
如果你的zk有其他的topic節(jié)點的話，會收不到消息，直接上解決方案：刪除所有的zk節(jié)點。怎么刪除？

上碼：

/*** zookeeper znode遞歸刪除節(jié)點* @author Halburt**/ public class DeleteZkNode {//zookeeper的地址 private static final String connectString = "192.168.1.177:2181";private static final int sessionTimeout = 2000;private static ZooKeeper zookeeper = null;/*** main函數(shù)* @param args* @throws Exception*/public static void main(String[] args) throws Exception {//調(diào)用rmr,刪除所有目錄rmr("/");}/*** 遞歸刪除 因為zookeeper只允許刪除葉子節(jié)點，如果要刪除非葉子節(jié)點，只能使用遞歸* @param path* @throws IOException*/public static void rmr(String path) throws Exception {ZooKeeper zk = getZookeeper();//獲取路徑下的節(jié)點List<String> children = zk.getChildren(path, false);for (String pathCd : children) {//獲取父節(jié)點下面的子節(jié)點路徑String newPath = "";//遞歸調(diào)用,判斷是否是根節(jié)點if (path.equals("/")) {newPath = "/" + pathCd;} else {newPath = path + "/" + pathCd;}rmr(newPath);}//刪除節(jié)點,并過濾zookeeper節(jié)點和 /節(jié)點if (path != null && !path.trim().startsWith("/zookeeper") && !path.trim().equals("/")) {zk.delete(path, -1);//打印刪除的節(jié)點路徑System.out.println("被刪除的節(jié)點為：" + path);}}/*** 獲取Zookeeper實例* @return* @throws IOException*/public static ZooKeeper getZookeeper() throws IOException {zookeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {}});return zookeeper;}}

終端命令查看消息

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.177:9999 --topic mytopic --from-beginning

安利一下可視化工具Kafka Tool 2

下載地址

Kafka Tool 2是一款Kafka的可視化客戶端工具，可以非常方便的查看Topic的隊列信息以及消費者信息以及kafka節(jié)點信息。下載地址：http://www.kafkatool.com/download.html

使用

先創(chuàng)建連接

下載安裝之后會彈出一個配置連接的窗口，我們可以看到這個窗口左上角為Add Cluster(添加集群)，但沒關(guān)系，對應(yīng)單節(jié)點的Kafka實例來說也是可以的，因為這個軟件監(jiān)控的是Zookeeper而不是Kafka，Kafka的集群搭建也是依賴Zookeeper來實現(xiàn)的，所以默認(rèn)情況下我們都是直接通過Zookeeper去完成大部分操作。

創(chuàng)建完成之后，連接

我們可以看到已經(jīng)創(chuàng)建好的Topic。這個軟件默認(rèn)顯示數(shù)據(jù)的類型為Byte，可以在設(shè)置里面找到對應(yīng)的修改選項

接下來就自己探索吧

理論學(xué)習(xí)

kafka單節(jié)點的結(jié)構(gòu)如下：

單節(jié)點broker包含多個topic主題，而每個topic則包含多個partition副本，每個partition會有序的存儲消息。

kafka的總體數(shù)據(jù)流

kafka對外使用topic的概念，生產(chǎn)者往topic里寫消息，消費者從topic讀消息。為了做到水平擴(kuò)展，一個topic實際是由多個partition組成的，遇到瓶頸時，可以通過增加partition的數(shù)量來進(jìn)行橫向擴(kuò)容。單個parition內(nèi)是保證消息有序。每新寫一條消息，kafka就是在對應(yīng)的文件append寫，所以性能非常高。kafka的總體數(shù)據(jù)流是這樣的：

Producers往Brokers里面的指定Topic中寫消息，Consumers從Brokers里面拉去指定Topic的消息，然后進(jìn)行業(yè)務(wù)處理。

名詞解析

Producer

消費者： Producer將消息發(fā)布到指定的Topic中,同時Producer也能決定將此消息歸屬于哪個partition;比如基于"round-robin"方式或者通過其他的一些算法等.

Consumer

每個consumer屬于一個consumer group;反過來說,每個group中可以有多個consumer.發(fā)送到Topic的消息,只會被訂閱此Topic的每個group中的一個consumer消費（對于一條消息來說，同一組的消費者只會有一個消費者去消費）.

?如果所有的consumer都具有相同的group,這種情況和queue模式很像;消息將會在consumers之間負(fù)載均衡.
?如果所有的consumer都具有不同的group,那這就是"發(fā)布-訂閱";消息將會廣播給所有的消費者.

在kafka中,一個partition中的消息只會被group中的一個consumer消費;每個group中consumer消息消費互相獨立;我們可以認(rèn)為一個group是一個"訂閱"者,一個Topic中的每個partions,只會被一個"訂閱者"中的一個consumer消費,不過一個consumer可以消費多個partitions中的消息.kafka只能保證一個partition中的消息被某個consumer消費時,消息是順序的。事實上,從Topic角度來說，消息仍不是有序的。

Topics

一個Topic可以認(rèn)為是一類消息，每個topic將被分成多個partition(區(qū)),每個partition在存儲層面是append log文件。任何發(fā)布到此partition的消息都會被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量），offset為一個long型數(shù)字，它是唯一標(biāo)記一條消息。它唯一的標(biāo)記一條消息。kafka并沒有提供其他額外的索引機(jī)制來存儲offset，因為在kafka中幾乎不允許對消息進(jìn)行“隨機(jī)讀寫”。

Partition

topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列

以下是單個生產(chǎn)者和消費者從兩個分區(qū)主題讀取和寫入的簡單示例。

此圖顯示了一個producer向2個partition分區(qū)寫入日志，以及消費者從相同日志中讀取的內(nèi)容。日志中的每條記錄都有一個相關(guān)的條目號，稱之為偏移量offset。消費者使用此偏移來記錄其在partitiond讀取日志的位置。

當(dāng)然如果存在多個消費者的話，根據(jù)groupId分組，同一組的消費者不會重復(fù)讀取日志。

換句話說： 訂閱topic是以一個消費組來訂閱的，一個消費組里面可以有多個消費者。同一個消費組中的兩個消費者，不會同時消費一個partition。換句話來說，就是一個partition，只能被消費組里的一個消費者消費，但是可以同時被多個消費組消費。因此，如果消費組內(nèi)的消費者如果比partition多的話，那么就會有個別消費者一直空閑。

其實consumer可以使用任意順序消費日志消息,它只需要將offset重置為任意值.(offset將會保存在zookeeper中，kafka集群幾乎不需要維護(hù)任何consumer和producer狀態(tài)信息,這些信息有zookeeper保存）

?partition有多個.最根本原因是kafka基于文件存儲.通過分區(qū),可以將日志內(nèi)容分散到多個partition上,來避免文件大小達(dá)到單機(jī)磁盤的上限,每個partiton都會被當(dāng)前server(kafka實例)保存;可以將一個topic切分多任意多個partitions,來消息保存/消費的效率.此外越多的partitions意味著可以容納更多的consumer,有效提升并發(fā)消費的能力.

使用場景

消息系統(tǒng)、消息隊列

? 對于一些常規(guī)的消息系統(tǒng),kafka是個不錯的選擇;partitons/replication和容錯,可以使kafka具有良好的擴(kuò)展性和性能優(yōu)勢.不過到目前為止,我們應(yīng)該很清楚認(rèn)識到,kafka并沒有提供JMS中的"事務(wù)性""消息傳輸擔(dān)保(消息確認(rèn)機(jī)制)""消息分組"等企業(yè)級特性;kafka只能使用作為"常規(guī)"的消息系統(tǒng),在一定程度上,尚未確保消息的發(fā)送與接收絕對可靠(比如,消息重發(fā),消息發(fā)送丟失等)

日志聚合

?kafka的特性決定它非常適合作為"日志收集中心";application可以將操作日志"批量""異步"的發(fā)送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/壓縮消息等,這對producer端而言,幾乎感覺不到性能的開支.此時consumer端可以使hadoop等其他系統(tǒng)化的存儲和分析系統(tǒng).

網(wǎng)站活動追蹤、調(diào)用鏈系統(tǒng)、事件采集

可以將網(wǎng)頁/用戶操作等信息發(fā)送到kafka中.并實時監(jiān)控,或者離線統(tǒng)計分析等

等等其他場景

server.properties配置文件解讀

############################# Server Basics ############################# # 節(jié)點的ID，必須與其它節(jié)點不同 broker.id=0 # 選擇啟用刪除主題功能，默認(rèn)false #delete.topic.enable=true ############################# Socket Server Settings ############################## 套接字服務(wù)器堅挺的地址。如果沒有配置，就使用java.net.InetAddress.getCanonicalHostName()的返回值 # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092# 節(jié)點的主機(jī)名會通知給生產(chǎn)者和消費者。如果沒有設(shè)置，如果配置了"listeners"就使用"listeners"的值。 # 否則就使用java.net.InetAddress.getCanonicalHostName()的返回值 #advertised.listeners=PLAINTEXT://your.host.name:9092# 將偵聽器的名稱映射到安全協(xié)議，默認(rèn)情況下它們是相同的。有關(guān)詳細(xì)信息，請參閱配置文檔 #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# 服務(wù)器用來接受請求或者發(fā)送響應(yīng)的線程數(shù) num.network.threads=3# 服務(wù)器用來處理請求的線程數(shù)，可能包括磁盤IO num.io.threads=8# 套接字服務(wù)器使用的發(fā)送緩沖區(qū)大小 socket.send.buffer.bytes=102400# 套接字服務(wù)器使用的接收緩沖區(qū)大小 socket.receive.buffer.bytes=102400# 單個請求最大能接收的數(shù)據(jù)量 socket.request.max.bytes=104857600############################# Log Basics ############################## 一個逗號分隔的目錄列表，用來存儲日志文件 log.dirs=/tmp/kafka-logs# 每個主題的日志分區(qū)的默認(rèn)數(shù)量。更多的分區(qū)允許更大的并行操作，但是它會導(dǎo)致節(jié)點產(chǎn)生更多的文件 num.partitions=1# 每個數(shù)據(jù)目錄中的線程數(shù)，用于在啟動時日志恢復(fù)，并在關(guān)閉時刷新。 num.recovery.threads.per.data.dir=1############################# Internal Topic Settings ############################# # 內(nèi)部主題設(shè)置 # 對于除了開發(fā)測試之外的其他任何東西，group元數(shù)據(jù)內(nèi)部主題的復(fù)制因子“__consumer_offsets”和“__transaction_state”，建議值大于1，以確保可用性(如3)。 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1############################# Log Flush Policy ############################## 在強(qiáng)制刷新數(shù)據(jù)到磁盤之前允許接收消息的數(shù)量 #log.flush.interval.messages=10000# 在強(qiáng)制刷新之前，消息可以在日志中停留的最長時間 #log.flush.interval.ms=1000############################# Log Retention Policy ############################## 以下的配置控制了日志段的處理。策略可以配置為每隔一段時間刪除片段或者到達(dá)一定大小之后。 # 當(dāng)滿足這些條件時，將會刪除一個片段。刪除總是發(fā)生在日志的末尾。# 一個日志的最小存活時間，可以被刪除 log.retention.hours=168# 一個基于大小的日志保留策略。段將被從日志中刪除只要剩下的部分段不低于log.retention.bytes。 #log.retention.bytes=1073741824# 每一個日志段大小的最大值。當(dāng)?shù)竭_(dá)這個大小時，會生成一個新的片段。 log.segment.bytes=1073741824# 檢查日志段的時間間隔，看是否可以根據(jù)保留策略刪除它們 log.retention.check.interval.ms=300000############################# Zookeeper #############################zookeeper.connect=localhost:2181# 連接到Zookeeper的超時時間 zookeeper.connection.timeout.ms=6000############################# Group Coordinator Settings #############################group.initial.rebalance.delay.ms=0

參考文章

https://www.cnblogs.com/likehua/p/3999538.html

https://www.jianshu.com/p/d3e963ff8b70

如有表述不當(dāng)之處，敬請指正。

轉(zhuǎn)載于:https://www.cnblogs.com/Halburt/p/10842597.html

總結(jié)

以上是生活随笔為你收集整理的从零开始学习Kafka的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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