Apache Kafka是開(kāi)源分布式高并發(fā)消息中間件，支持每秒百萬(wàn)級(jí)消息并發(fā)，在互聯(lián)網(wǎng)高并發(fā)架構(gòu)：雙11、電商秒殺搶購(gòu)、網(wǎng)絡(luò)直播、IOT大數(shù)據(jù)采集、聊天App、導(dǎo)航等高并發(fā)架構(gòu)中大量使用。

生產(chǎn)環(huán)境一般要求搭建Kafka集群。Java開(kāi)發(fā)Kafka集群需要注意參數(shù)的詳細(xì)配置，Kafka參數(shù)的含義在配置集群的時(shí)候非常重要，尤其是關(guān)系性能和集群的參數(shù)。下面我們一起來(lái)看看Kafka的詳細(xì)參數(shù)。

Kafka服務(wù)器基礎(chǔ)配置

broker的身份ID。必須為每個(gè)代理設(shè)置一個(gè)唯一的整數(shù)。集群配置時(shí)候非常重要。不能重復(fù)。
broker.id = 1

Kafka套接字服務(wù)器配置

Kafka服務(wù)器套接字服務(wù)器偵聽(tīng)的地址。 如果未配置，就會(huì)使用java.net.InetAddress.getCanonicalHostName（）地址。格式：
＃listeners = listener_name：// host_name：port
例如：

listeners = PLAINTEXT：//localhost：9092

listeners = PLAINTEXT：//localhost：9091

Broker向生產(chǎn)者和消費(fèi)者通告的主機(jī)名和端口。如果沒(méi)有設(shè)定，它將使用“偵聽(tīng)器”的值。否則，它將使用該值從java.net.InetAddress.getCanonicalHostName（）返回的地址。
＃advertised.listeners = PLAINTEXT：//host.name：9092

Kafka偵聽(tīng)器名稱映射到安全協(xié)議，默認(rèn)為它們是相同名稱
＃listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL：SSL，SASL_PLAINTEXT：SASL_PLAINTEXT，SASL_SSL：SASL_SSL

服務(wù)器用于從網(wǎng)絡(luò)接收請(qǐng)求并向網(wǎng)絡(luò)發(fā)送響應(yīng)消息的線程數(shù)
num.network.threads = 3

服務(wù)器用于處理請(qǐng)求的線程數(shù)，可能包括磁盤I / O需要的線程.
num.io.threads = 8

套接字服務(wù)器使用的發(fā)送緩沖區(qū)（SO_SNDBUF）大小，字節(jié)。
socket.send.buffer.bytes = 102400

套接字服務(wù)器使用的接收緩沖區(qū)（SO_RCVBUF）大小，字節(jié)。
socket.receive.buffer.bytes = 102400

套接字服務(wù)器將接受的請(qǐng)求的最大大?。ǚ乐筄OM內(nèi)存溢出）大小，字節(jié)。
socket.request.max.bytes = 104857600

Kafka日志配置

逗號(hào)分隔的目錄列表，用于存儲(chǔ)日志文件
log.dirs =。/日志

每個(gè)主題的默認(rèn)日志分區(qū)數(shù)。更多分區(qū)允許更大并行處理消息，但這也會(huì)導(dǎo)致更多的文件
num.partitions = 1

在啟動(dòng)時(shí)用于日志恢復(fù)和在關(guān)閉時(shí)刷新的每個(gè)數(shù)據(jù)目錄文件夾需要的線程數(shù)。
＃對(duì)于數(shù)據(jù)目錄文件夾位于RAID陣列中的情況，建議增加此線程數(shù)值。
num.recovery.threads.per.data.dir = 1

Kafka內(nèi)部主題配置

Topic主題的消費(fèi)者的group元數(shù)據(jù)復(fù)制因子，"__consumer_offsets" 和"__transaction_state" 。
offsets.topic.replication.factor = 1
transaction.state.log.replication.factor = 1
transaction.state.log.min.isr = 1

Kafka日志刷新政策配置

Kafka消息立即寫(xiě)入文件系統(tǒng)，但默認(rèn)情況下我們只有fsync（）來(lái)緩慢延遲地同步操作系統(tǒng)緩存消息到磁盤上。 以下配置參數(shù)控制將消息數(shù)據(jù)刷新到磁盤過(guò)程。這里有一些重要的權(quán)衡：
1、持久性：如果Kafka不使用復(fù)制，則可能會(huì)丟失未刷新的數(shù)據(jù)。
2、延遲：當(dāng)Kafka刷新確實(shí)發(fā)生時(shí)，非常大的刷新間隔可能會(huì)導(dǎo)致延遲峰值，因?yàn)闀?huì)有大量數(shù)據(jù)需要刷新到磁盤，間隔太久緩沖消息越多。
3、吞吐量：Flush通常是最昂貴的操作，并且小的Flush間隔可能導(dǎo)致過(guò)多的磁盤IO操作搜索。
下面以下設(shè)置允許配置刷新策略以在一段時(shí)間后或每N條消息（或兩者）刷新數(shù)據(jù)。這可以在全局范圍內(nèi)配置完成，也可以針對(duì)每個(gè)主題的單獨(dú)配置。

＃強(qiáng)制刷新數(shù)據(jù)到磁盤之前要接受的消息數(shù)，10000消息時(shí)批量刷盤
＃log.flush.interval.messages = 10000

＃強(qiáng)制刷新之前消息可以在日志中停留的最長(zhǎng)時(shí)間 1000毫秒
＃log.flush.interval.ms = 1000

Kafka日志保存策略

以下配置參數(shù)控制Kafka日志段的處理策略，我們可以設(shè)置為在一段時(shí)間后或在累積給定大小后刪除日志數(shù)據(jù)段。只要滿足一下任意條件，就會(huì)刪除一個(gè)日志段。刪除總是從日志的末尾開(kāi)始。

log log的最小時(shí)間長(zhǎng)度，超過(guò)刪除日志。
＃由于時(shí)間可以刪除的日志文件的最小年齡 168小時(shí)
log.retention.hours = 168

基于大小的日志保留策略。 除非剩余的段大小在log.retention.bytes之下，否則將從日志中刪除段。 功能獨(dú)立于log.retention.hours限制。
＃log.retention.bytes = 1073741824

＃日志段文件的最大大小。達(dá)到此大小時(shí)，將創(chuàng)建新的日志段。
log.segment.bytes = 1073741824

＃檢查日志段以查看是否可以刪除日志段的時(shí)間間隔
＃保留策略
log.retention.check.interval.ms = 300000

Zookeeper參數(shù)設(shè)置

Zookeeper連接字符串。默認(rèn)單臺(tái)設(shè)置，集群需要多臺(tái)，逗號(hào)分隔的主機(jī)：端口，每個(gè)對(duì)應(yīng)一個(gè)Zookeeper。例如“127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002”。
＃所有kafka znodes的根目錄。集群設(shè)置模式
zookeeper.connect =localhost：2181，localhost：2182，localhost：2183

連接到zookeeper的超時(shí)時(shí)間（以毫秒為單位）
zookeeper.connection.timeout.ms = 6000

kafka Group協(xié)調(diào)配置

以下配置指定GroupCoordinator將延遲初始消費(fèi)者重新平衡的時(shí)間（以毫秒為單位）。
＃當(dāng)新成員加入組時(shí)，重新平衡延遲是group.initial.rebalance.delay.ms的值，最多max.poll.interval.ms。
默認(rèn)值為3秒。我們將此參數(shù)設(shè)置為0，方便開(kāi)發(fā)和測(cè)試。但是生產(chǎn)環(huán)境中默認(rèn)值推薦3秒更合適，因?yàn)檫@有助于避免在應(yīng)用程序啟動(dòng)期間不必要且可能很昂貴的重新平衡過(guò)程，減少系統(tǒng)資源的消耗。
group.initial.rebalance.delay.ms = 0

在集群和優(yōu)化情況下需要了解每個(gè)參數(shù)的確切含義，對(duì)于Kafka集群的設(shè)置，需要配置多個(gè)Zookeeper地址。默認(rèn)的日志清理、垃圾回收、連接池、線程模型都是非常重要的因素。
使用最新的Java Spring Boot 2.x版本連接Kafka需要在配置文件中修改地址參數(shù)：

spring.kafka.consumer.group-id=myGroup
spring.kafka.bootstrap-servers=localhost:9091,localhost:9092,localhost:9093