Kafka参数详解及调优--生产者
在實際的kafka開發中,我們會發現,無論是生產者還是消費者,都需要構建一個Properties對象,里面設置了很多參數。對于很多初學者來說,會看不懂這些參數分別代表什么含義。
在本篇文章我們就來詳細地了解一下這些參數的作用,并探討下如何使用合理的配置去優化提高生產/消費效率
kafka生產者參數
Properties props = new Properties(); props.put("bootstrap.servers", "192.168.137.200: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> procuder = new KafkaProducer<String,String>(props);在這段代碼中有很多常用的參數配置,在線上使用時,我們要根據實際的數據量和數據大小來決定這些配置的具體值。下面來挑出其中比較重要的幾個參數來詳細解析一下。
1 bootstrap.servers
host/port列表,用于初始化建立和Kafka集群的連接。列表格式為host1:port1,host2:port2,....,無需添加所有的集群地址,kafka會根據提供的地址發現其他的地址(你可以多提供幾個,以防提供的服務器關閉)
2 acks
生產者需要leader確認請求完成之前接收的應答數。此配置控制了發送消息的耐用性,支持以下配置:
- acks=0 如果設置為0,那么生產者將不等待任何消息確認。消息將立刻添加到socket緩沖區并考慮發送。在這種情況下不能保障消息被服務器接收到。并且重試機制不會生效(因為客戶端不知道故障了沒有)。每個消息返回的offset始終設置為-1。
- acks=1,這意味著leader寫入消息到本地日志就立即響應,而不等待所有follower應答。在這種情況下,如果響應消息之后但follower還未復制之前leader立即故障,那么消息將會丟失。
- acks=all 這意味著leader將等待所有副本同步后應答消息。此配置保障消息不會丟失(只要至少有一個同步的副本)。這是最強壯的可用性保障。等價于acks=-1。
3 buffer.memory
生產者用來緩存等待發送到服務器的消息的內存總字節數。如果消息發送比可傳遞到服務器的快,生產者將阻塞max.block.ms之后,拋出異常。
此設置應該大致的對應生產者將要使用的總內存,但不是硬約束,因為生產者所使用的所有內存都用于緩沖。一些額外的內存將用于壓縮(如果啟動壓縮),以及用于保持發送中的請求。
首先要明確一點,那就是在內存緩沖里大量的消息會緩沖在里面,形成一個一個的Batch,每個Batch里包含多條消息。然后KafkaProducer有一個Sender線程會把多個Batch打包成一個Request發送到Kafka服務器上去。 那么如果要是內存設置的太小,可能導致一個問題:消息快速的寫入內存緩沖里面,但是Sender線程來不及把Request發送到Kafka服務器。這樣是不是會造成內存緩沖很快就被寫滿?一旦被寫滿,就會阻塞用戶線程,不讓繼續往Kafka寫消息了。
所以對于“buffer.memory”這個參數應該結合自己的實際情況來進行壓測,你需要測算一下在生產環境,你的用戶線程會以每秒多少消息的頻率來寫入內存緩沖。比如說每秒300條消息,那么你就需要壓測一下,假設內存緩沖就32MB,每秒寫300條消息到內存緩沖,是否會經常把內存緩沖寫滿?經過這樣的壓測,你可以調試出來一個合理的內存大小。
4 batch.size
當多個消息要發送到相同分區的時,生產者嘗試將消息批量打包在一起,以減少請求交互。這樣有助于客戶端和服務端的性能提升。該配置的默認批次大小(以字節為單位):
不會打包大于此配置大小的消息。
發送到broker的請求將包含多個批次,每個分區一個,用于發送數據。
較小的批次大小有可能降低吞吐量(批次大小為0則完全禁用批處理)。一個非常大的批次大小可能更浪費內存。因為我們會預先分配這個資源。
比如說發送消息的頻率就是每秒300條,那么如果比如“batch.size”調節到了32KB,或者64KB,是否可以提升發送消息的整體吞吐量。
因為理論上來說,提升batch的大小,可以允許更多的數據緩沖在里面,那么一次Request發送出去的數據量就更多了,這樣吞吐量可能會有所提升。
但是這個東西也不能無限的大,過于大了之后,要是數據老是緩沖在Batch里遲遲不發送出去,那么豈不是你發送消息的延遲就會很高。
比如說,一條消息進入了Batch,但是要等待5秒鐘Batch才湊滿了64KB,才能發送出去。那這條消息的延遲就是5秒鐘。
所以需要在這里按照生產環境的發消息的速率,調節不同的Batch大小自己測試一下最終出去的吞吐量以及消息的 延遲,設置一個最合理的參數。
5 compression.type
數據壓縮的類型。默認為空(就是不壓縮)。有效的值有 none,gzip,snappy, 或 lz4。壓縮全部的數據批,因此批的效果也將影響壓縮的比率(更多的批次意味著更好的壓縮)。
6 retries
設置一個比零大的值,客戶端如果發送失敗則會重新發送。注意,這個重試功能和客戶端在接到錯誤之后重新發送沒什么不同。如果max.in.flight.requests.per.connection沒有設置為1,有可能改變消息發送的順序,因為如果2個批次發送到一個分區中,并第一個失敗了并重試,但是第二個成功了,那么第二個批次將超過第一個。
“retries”和“retries.backoff.ms”決定了重試機制,也就是如果一個請求失敗了可以重試幾次,每次重試的間隔是多少毫秒。這個大家適當設置幾次重試的機會,給一定的重試間隔即可,比如給100ms的重試間隔。
7 client.id
當發出請求時傳遞給服務器的id字符串。這樣做的目的是允許服務器請求記錄記錄這個【邏輯應用名】,這樣能夠追蹤請求的源,而不僅僅只是ip/prot
8 linger.ms
生產者組將發送的消息組合成單個批量請求。正常情況下,只有消息到達的速度比發送速度快的情況下才會出現。但是,在某些情況下,即使在適度的負載下,客戶端也可能希望減少請求數量。此設置通過添加少量人為延遲來實現。- 也就是說,不是立即發出一個消息,生產者將等待一個給定的延遲,以便和其他的消息可以組合成一個批次。這類似于Nagle在TCP中的算法。此設置給出批量延遲的上限:一旦我們達到分區的batch.size值的記錄,將立即發送,不管這個設置如何,但是,如果比這個小,我們將在指定的“linger”時間內等待更多的消息加入。此設置默認為0(即無延遲)。假設,設置 linger.ms=5,將達到減少發送的請求數量的效果,但對于在沒有負載情況,將增加5ms的延遲。
舉個例子:首先假設你的Batch是32KB,那么你得估算一下,正常情況下,一般多久會湊夠一個Batch,比如正常來說可能20ms就會湊夠一個Batch。
那么你的linger.ms就可以設置為25ms,也就是說,正常來說,大部分的Batch在20ms內都會湊滿,但是你的linger.ms可以保證,哪怕遇到低峰時期,20ms湊不滿一個Batch,還是會在25ms之后強制Batch發送出去。
如果要是你把linger.ms設置的太小了,比如說默認就是0ms,或者你設置個5ms,那可能導致你的Batch雖然設置了32KB,但是經常是還沒湊夠32KB的數據,5ms之后就直接強制Batch發送出去,這樣也不太好其實,會導致你的Batch形同虛設,一直湊不滿數據。
9 max.request.size
請求的最大大小(以字節為單位)。此設置將限制生產者的單個請求中發送的消息批次數,以避免發送過大的請求。這也是最大消息批量大小的上限。請注意,服務器擁有自己的批量大小,可能與此不同。
10 max.block.ms
該配置控制 KafkaProducer.send() 和 KafkaProducer.partitionsFor() 將阻塞多長時間。此外這些方法被阻止,也可能是因為緩沖區已滿或元數據不可用。在用戶提供的序列化程序或分區器中的鎖定不會計入此超時。默認為60000ms。
總結
以上是生活随笔為你收集整理的Kafka参数详解及调优--生产者的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 程序员,你想不想进大厂?
- 下一篇: 62岁程序员植入逻辑炸弹, 面临10年监