从零开始学习Kafka
簡(jiǎn)介
kafka是一個(gè)分布式消息隊(duì)列。具有高性能、持久化、多副本備份、橫向擴(kuò)展能力。生產(chǎn)者往隊(duì)列里寫消息，消費(fèi)者從隊(duì)列里取消息進(jìn)行業(yè)務(wù)邏輯。一般在架構(gòu)設(shè)計(jì)中起到解耦、削峰、異步處理的作用。
Kafka核心組件-intsmaze
- Topic：消息根據(jù)Topic進(jìn)行歸類，可以理解為一個(gè)隊(duì)里。
- Producer：消息生產(chǎn)者，就是向kafka broker發(fā)消息的客戶端。
- Consumer：消息消費(fèi)者，向kafka broker取消息的客戶端。
- broker：每個(gè)kafka實(shí)例(server)，一臺(tái)kafka服務(wù)器就是一個(gè)broker，一個(gè)集群由多個(gè)broker組成，一個(gè)broker可以容納多個(gè)topic。
Zookeeper：依賴集群保存meta信息。
大家先看kafka的介紹或者教程啊，上來都顯示一堆長(zhǎng)篇大論，各自文字圖片，看著很懵逼，頭暈。搞程序的，要讓ta跑起來，再針對(duì)可運(yùn)行的成果，慢慢了解ta。所以本文會(huì)由淺入深，先實(shí)踐后理論，結(jié)合實(shí)踐講理論。Kafka安裝配置
下載
解壓
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啟動(dòng)zk【默認(rèn)端口2181】
bin/zookeeper-server-start.sh config/zookeeper.properties啟動(dòng)Kafka
使用?kafka-server-start.sh?啟動(dòng) kafka 服務(wù)：
bin/kafka-server-start.sh config/server.properties測(cè)試使用
創(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讀取消息【消費(fèi)者】
使用?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ū)的摘要，每個(gè)附加行給出了關(guān)于一個(gè)分區(qū)的信息。 由于我們只有一個(gè)分區(qū)，所以只有一行。
- “Leader”: 是負(fù)責(zé)給定分區(qū)的所有讀取和寫入的節(jié)點(diǎn)。 每個(gè)節(jié)點(diǎn)將成為分區(qū)隨機(jī)選擇部分的領(lǐng)導(dǎo)者。
- “Replicas”: 是復(fù)制此分區(qū)日志的節(jié)點(diǎn)列表，無論它們是否是領(lǐng)導(dǎo)者，或者即使他們當(dāng)前處于活動(dòng)狀態(tài)。
- “Isr”: 是一組“同步”副本。這是復(fù)制品列表的子集，當(dāng)前活著并被引導(dǎo)到領(lǐng)導(dǎo)者
擴(kuò)展-集群配置
Kafka 支持兩種模式的集群搭建：可以在單機(jī)上運(yùn)行多個(gè) broker 實(shí)例來實(shí)現(xiàn)集群，也可在多臺(tái)機(jī)器上搭建集群，下面介紹下如何實(shí)現(xiàn)單機(jī)多 broker 實(shí)例集群，其實(shí)很簡(jiǎn)單，只需要如下配置即可。
單機(jī)多broker 集群配置
利用單節(jié)點(diǎn)部署多個(gè) 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同樣，配置第三個(gè)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-logs2listeners 申明此kafka服務(wù)器需要監(jiān)聽的端口號(hào)，默認(rèn)會(huì)使用localhost的地址，如果是在遠(yuǎn)程服務(wù)器上運(yùn)行則必須配置，例如：
listeners=PLAINTEXT:// 192.168.180.128:9092
并確保服務(wù)器的9092端口能夠訪問
啟動(dòng)2/3 borker
bin/kafka-server-start.sh config/server-2.properties & bin/kafka-server-start.sh config/server-3.properties &至此，單機(jī)多broker實(shí)例的集群配置完畢。
擴(kuò)展-多機(jī)多borker集群
分別在多個(gè)節(jié)點(diǎn)按上述方式安裝 Kafka，配置啟動(dòng)多個(gè) Zookeeper 實(shí)例。
假設(shè)三臺(tái)機(jī)器 IP 地址是 ： 192.168.153.135， 192.168.153.136， 192.168.153.137
分別配置多個(gè)機(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ù)
從控制臺(tái)寫入數(shù)據(jù)并將其寫回控制臺(tái)是一個(gè)方便的起點(diǎn)，但您可能想要使用其他來源的數(shù)據(jù)或?qū)?shù)據(jù)從 Kafka 導(dǎo)出到其他系統(tǒng)。對(duì)于許多系統(tǒng)，您可以使用 Kafka Connect 來導(dǎo)入或?qū)С鰯?shù)據(jù)，而不必編寫自定義集成代碼。
Kafka Connect 是 Kafka 包含的一個(gè)工具，可以將數(shù)據(jù)導(dǎo)入和導(dǎo)出到 Kafka。它是一個(gè)可擴(kuò)展的工具，運(yùn)行 連接器，實(shí)現(xiàn)與外部系統(tǒng)交互的自定義邏輯。在這個(gè)快速入門中，我們將看到如何使用簡(jiǎn)單的連接器運(yùn)行 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
代碼測(cè)試
準(zhǔn)備測(cè)試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-999192.168.1.177為kafka所在機(jī)器的ip地址，9999端口號(hào)是對(duì)外提供的端口，下文會(huì)使用到
Springboot 發(fā)送消息、接受消息源碼
很簡(jiǎn)單的一個(gè)小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.StringSerializer192.168.1.177:9999即為kafka的配置文件中配置
pom.xml依賴:
依賴版本：
spring-boot.version：2.1.3.RELEASE
spring-kafka.version：2.2.0.RELEASE
【此處有坑】此處依賴版本可以不用這2個(gè)版本，但是一定要注意springboot和kafka的版本對(duì)應(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>啟動(dòng)kafka并run Application.java
bin/kafka-server-start.sh config/server-idea.properties &
已經(jīng)啟動(dòng)了zk，此處不用再啟動(dòng)，如果未啟動(dòng)，需要啟動(dòng)zk。
kafka啟動(dòng)成功之后，run Application，會(huì)看到日志如下：
已經(jīng)接收到消息了。
如果你是跟著本文從頭開始的，一定注意此處有個(gè)坑
如果你是從頭開始跟這個(gè)本文學(xué)習(xí)的，那么你直接啟動(dòng)的話，會(huì)發(fā)現(xiàn)消息發(fā)出去了，但是沒有接收到。
我也是查了好久，看了很多教程，別人都行我就不行。
如果你的zk有其他的topic節(jié)點(diǎn)的話，會(huì)收不到消息，直接上解決方案：刪除所有的zk節(jié)點(diǎn)。怎么刪除？
上碼：
/*** zookeeper znode遞歸刪除節(jié)點(diǎn)* @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("/");}/*** 遞歸刪除 因?yàn)閦ookeeper只允許刪除葉子節(jié)點(diǎn)，如果要?jiǎng)h除非葉子節(jié)點(diǎn)，只能使用遞歸* @param path* @throws IOException*/public static void rmr(String path) throws Exception {ZooKeeper zk = getZookeeper();//獲取路徑下的節(jié)點(diǎn)List<String> children = zk.getChildren(path, false);for (String pathCd : children) {//獲取父節(jié)點(diǎn)下面的子節(jié)點(diǎn)路徑String newPath = "";//遞歸調(diào)用,判斷是否是根節(jié)點(diǎn)if (path.equals("/")) {newPath = "/" + pathCd;} else {newPath = path + "/" + pathCd;}rmr(newPath);}//刪除節(jié)點(diǎn),并過濾zookeeper節(jié)點(diǎn)和 /節(jié)點(diǎn)if (path != null && !path.trim().startsWith("/zookeeper") && !path.trim().equals("/")) {zk.delete(path, -1);//打印刪除的節(jié)點(diǎn)路徑System.out.println("被刪除的節(jié)點(diǎn)為：" + path);}}/*** 獲取Zookeeper實(shí)例* @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的隊(duì)列信息以及消費(fèi)者信息以及kafka節(jié)點(diǎn)信息。下載地址：http://www.kafkatool.com/download.html
使用
先創(chuàng)建連接
下載安裝之后會(huì)彈出一個(gè)配置連接的窗口，我們可以看到這個(gè)窗口左上角為Add Cluster(添加集群)，但沒關(guān)系，對(duì)應(yīng)單節(jié)點(diǎn)的Kafka實(shí)例來說也是可以的，因?yàn)檫@個(gè)軟件監(jiān)控的是Zookeeper而不是Kafka，Kafka的集群搭建也是依賴Zookeeper來實(shí)現(xiàn)的，所以默認(rèn)情況下我們都是直接通過Zookeeper去完成大部分操作。
創(chuàng)建完成之后，連接
我們可以看到已經(jīng)創(chuàng)建好的Topic。這個(gè)軟件默認(rèn)顯示數(shù)據(jù)的類型為Byte，可以在設(shè)置里面找到對(duì)應(yīng)的修改選項(xiàng)
接下來就自己探索吧
理論學(xué)習(xí)
kafka單節(jié)點(diǎn)的結(jié)構(gòu)如下：
單節(jié)點(diǎn)broker包含多個(gè)topic主題，而每個(gè)topic則包含多個(gè)partition副本，每個(gè)partition會(huì)有序的存儲(chǔ)消息。
kafka的總體數(shù)據(jù)流
kafka對(duì)外使用topic的概念，生產(chǎn)者往topic里寫消息，消費(fèi)者從topic讀消息。為了做到水平擴(kuò)展，一個(gè)topic實(shí)際是由多個(gè)partition組成的，遇到瓶頸時(shí)，可以通過增加partition的數(shù)量來進(jìn)行橫向擴(kuò)容。單個(gè)parition內(nèi)是保證消息有序。每新寫一條消息，kafka就是在對(duì)應(yīng)的文件append寫，所以性能非常高。kafka的總體數(shù)據(jù)流是這樣的：
Producers往Brokers里面的指定Topic中寫消息，Consumers從Brokers里面拉去指定Topic的消息，然后進(jìn)行業(yè)務(wù)處理。
名詞解析
Producer
消費(fèi)者： Producer將消息發(fā)布到指定的Topic中,同時(shí)Producer也能決定將此消息歸屬于哪個(gè)partition;比如基于"round-robin"方式或者通過其他的一些算法等.
Consumer
每個(gè)consumer屬于一個(gè)consumer group;反過來說,每個(gè)group中可以有多個(gè)consumer.發(fā)送到Topic的消息,只會(huì)被訂閱此Topic的每個(gè)group中的一個(gè)consumer消費(fèi)（對(duì)于一條消息來說，同一組的消費(fèi)者只會(huì)有一個(gè)消費(fèi)者去消費(fèi)）.
?如果所有的consumer都具有相同的group,這種情況和queue模式很像;消息將會(huì)在consumers之間負(fù)載均衡.
?如果所有的consumer都具有不同的group,那這就是"發(fā)布-訂閱";消息將會(huì)廣播給所有的消費(fèi)者.
在kafka中,一個(gè)partition中的消息只會(huì)被group中的一個(gè)consumer消費(fèi);每個(gè)group中consumer消息消費(fèi)互相獨(dú)立;我們可以認(rèn)為一個(gè)group是一個(gè)"訂閱"者,一個(gè)Topic中的每個(gè)partions,只會(huì)被一個(gè)"訂閱者"中的一個(gè)consumer消費(fèi),不過一個(gè)consumer可以消費(fèi)多個(gè)partitions中的消息.kafka只能保證一個(gè)partition中的消息被某個(gè)consumer消費(fèi)時(shí),消息是順序的。事實(shí)上,從Topic角度來說，消息仍不是有序的。
Topics
一個(gè)Topic可以認(rèn)為是一類消息，每個(gè)topic將被分成多個(gè)partition(區(qū)),每個(gè)partition在存儲(chǔ)層面是append log文件。任何發(fā)布到此partition的消息都會(huì)被直接追加到log文件的尾部，每條消息在文件中的位置稱為offset（偏移量），offset為一個(gè)long型數(shù)字，它是唯一標(biāo)記一條消息。它唯一的標(biāo)記一條消息。kafka并沒有提供其他額外的索引機(jī)制來存儲(chǔ)offset，因?yàn)樵趉afka中幾乎不允許對(duì)消息進(jìn)行“隨機(jī)讀寫”。
Partition
topic物理上的分組，一個(gè)topic可以分為多個(gè)partition，每個(gè)partition是一個(gè)有序的隊(duì)列
以下是單個(gè)生產(chǎn)者和消費(fèi)者從兩個(gè)分區(qū)主題讀取和寫入的簡(jiǎn)單示例。
此圖顯示了一個(gè)producer向2個(gè)partition分區(qū)寫入日志，以及消費(fèi)者從相同日志中讀取的內(nèi)容。日志中的每條記錄都有一個(gè)相關(guān)的條目號(hào)，稱之為偏移量offset。消費(fèi)者使用此偏移來記錄其在partitiond讀取日志的位置。
當(dāng)然如果存在多個(gè)消費(fèi)者的話，根據(jù)groupId分組，同一組的消費(fèi)者不會(huì)重復(fù)讀取日志。
換句話說： 訂閱topic是以一個(gè)消費(fèi)組來訂閱的，一個(gè)消費(fèi)組里面可以有多個(gè)消費(fèi)者。同一個(gè)消費(fèi)組中的兩個(gè)消費(fèi)者，不會(huì)同時(shí)消費(fèi)一個(gè)partition。換句話來說，就是一個(gè)partition，只能被消費(fèi)組里的一個(gè)消費(fèi)者消費(fèi)，但是可以同時(shí)被多個(gè)消費(fèi)組消費(fèi)。因此，如果消費(fèi)組內(nèi)的消費(fèi)者如果比partition多的話，那么就會(huì)有個(gè)別消費(fèi)者一直空閑。其實(shí)consumer可以使用任意順序消費(fèi)日志消息,它只需要將offset重置為任意值.(offset將會(huì)保存在zookeeper中，kafka集群幾乎不需要維護(hù)任何consumer和producer狀態(tài)信息,這些信息有zookeeper保存）
?partition有多個(gè).最根本原因是kafka基于文件存儲(chǔ).通過分區(qū),可以將日志內(nèi)容分散到多個(gè)partition上,來避免文件大小達(dá)到單機(jī)磁盤的上限,每個(gè)partiton都會(huì)被當(dāng)前server(kafka實(shí)例)保存;可以將一個(gè)topic切分多任意多個(gè)partitions,來消息保存/消費(fèi)的效率.此外越多的partitions意味著可以容納更多的consumer,有效提升并發(fā)消費(fèi)的能力.
使用場(chǎng)景
消息系統(tǒng)、消息隊(duì)列
? 對(duì)于一些常規(guī)的消息系統(tǒng),kafka是個(gè)不錯(cuò)的選擇;partitons/replication和容錯(cuò),可以使kafka具有良好的擴(kuò)展性和性能優(yōu)勢(shì).不過到目前為止,我們應(yīng)該很清楚認(rèn)識(shí)到,kafka并沒有提供JMS中的"事務(wù)性""消息傳輸擔(dān)保(消息確認(rèn)機(jī)制)""消息分組"等企業(yè)級(jí)特性;kafka只能使用作為"常規(guī)"的消息系統(tǒng),在一定程度上,尚未確保消息的發(fā)送與接收絕對(duì)可靠(比如,消息重發(fā),消息發(fā)送丟失等)
日志聚合
?kafka的特性決定它非常適合作為"日志收集中心";application可以將操作日志"批量""異步"的發(fā)送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/壓縮消息等,這對(duì)producer端而言,幾乎感覺不到性能的開支.此時(shí)consumer端可以使hadoop等其他系統(tǒng)化的存儲(chǔ)和分析系統(tǒng).
網(wǎng)站活動(dòng)追蹤、調(diào)用鏈系統(tǒng)、事件采集
可以將網(wǎng)頁/用戶操作等信息發(fā)送到kafka中.并實(shí)時(shí)監(jiān)控,或者離線統(tǒng)計(jì)分析等
等等其他場(chǎng)景
server.properties配置文件解讀
############################# Server Basics ############################# # 節(jié)點(diǎn)的ID，必須與其它節(jié)點(diǎn)不同 broker.id=0 # 選擇啟用刪除主題功能，默認(rèn)false #delete.topic.enable=true ############################# Socket Server Settings ############################## 套接字服務(wù)器堅(jiān)挺的地址。如果沒有配置，就使用java.net.InetAddress.getCanonicalHostName()的返回值 # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092# 節(jié)點(diǎn)的主機(jī)名會(huì)通知給生產(chǎn)者和消費(fèi)者。如果沒有設(shè)置，如果配置了"listeners"就使用"listeners"的值。 # 否則就使用java.net.InetAddress.getCanonicalHostName()的返回值 #advertised.listeners=PLAINTEXT://your.host.name:9092# 將偵聽器的名稱映射到安全協(xié)議，默認(rèn)情況下它們是相同的。有關(guān)詳細(xì)信息，請(qǐng)參閱配置文檔 #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# 服務(wù)器用來接受請(qǐng)求或者發(fā)送響應(yīng)的線程數(shù) num.network.threads=3# 服務(wù)器用來處理請(qǐng)求的線程數(shù)，可能包括磁盤IO num.io.threads=8# 套接字服務(wù)器使用的發(fā)送緩沖區(qū)大小 socket.send.buffer.bytes=102400# 套接字服務(wù)器使用的接收緩沖區(qū)大小 socket.receive.buffer.bytes=102400# 單個(gè)請(qǐng)求最大能接收的數(shù)據(jù)量 socket.request.max.bytes=104857600############################# Log Basics ############################## 一個(gè)逗號(hào)分隔的目錄列表，用來存儲(chǔ)日志文件 log.dirs=/tmp/kafka-logs# 每個(gè)主題的日志分區(qū)的默認(rèn)數(shù)量。更多的分區(qū)允許更大的并行操作，但是它會(huì)導(dǎo)致節(jié)點(diǎn)產(chǎn)生更多的文件 num.partitions=1# 每個(gè)數(shù)據(jù)目錄中的線程數(shù)，用于在啟動(dòng)時(shí)日志恢復(fù)，并在關(guān)閉時(shí)刷新。 num.recovery.threads.per.data.dir=1############################# Internal Topic Settings ############################# # 內(nèi)部主題設(shè)置 # 對(duì)于除了開發(fā)測(cè)試之外的其他任何東西，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)制刷新之前，消息可以在日志中停留的最長(zhǎng)時(shí)間 #log.flush.interval.ms=1000############################# Log Retention Policy ############################## 以下的配置控制了日志段的處理。策略可以配置為每隔一段時(shí)間刪除片段或者到達(dá)一定大小之后。 # 當(dāng)滿足這些條件時(shí)，將會(huì)刪除一個(gè)片段。刪除總是發(fā)生在日志的末尾。# 一個(gè)日志的最小存活時(shí)間，可以被刪除 log.retention.hours=168# 一個(gè)基于大小的日志保留策略。段將被從日志中刪除只要剩下的部分段不低于log.retention.bytes。 #log.retention.bytes=1073741824# 每一個(gè)日志段大小的最大值。當(dāng)?shù)竭_(dá)這個(gè)大小時(shí)，會(huì)生成一個(gè)新的片段。 log.segment.bytes=1073741824# 檢查日志段的時(shí)間間隔，看是否可以根據(jù)保留策略刪除它們 log.retention.check.interval.ms=300000############################# Zookeeper #############################zookeeper.connect=localhost:2181# 連接到Zookeeper的超時(shí)時(shí)間 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)之處，敬請(qǐng)指正。
轉(zhuǎn)載于:https://www.cnblogs.com/Halburt/p/10842597.html
總結(jié)
以上是生活随笔為你收集整理的从零开始学习Kafka的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 网站搭建 (第09天) 博客统计排行
- 下一篇： Nginx upstream的5种权重分