kafka(06)——kafka配置檔案的說明
阿新 • • 發佈:2018-12-13
server.properties的配置說明
#broker的全域性唯一編號,不能重複 broker.id=0 #用來監聽連結的埠,producer或consumer將在此埠建立連線 port=9092 #處理網路請求的執行緒數量 num.network.threads=3 #用來處理磁碟IO的執行緒數量 num.io.threads=8 #傳送套接字的緩衝區大小 socket.send.buffer.bytes=102400 #接受套接字的緩衝區大小 socket.receive.buffer.bytes=102400 #請求套接字的緩衝區大小 socket.request.max.bytes=104857600 #kafka執行日誌存放的路徑,根據我們的磁碟路徑來定,多個磁碟用逗號隔開 log.dirs=/export/data/kafka/ #topic在當前broker上的分片個數,分割槽的數量值,一般都需要我們在建立topic的時候手動指定,覆蓋這個配置 num.partitions=2 #用來恢復和清理data下資料的執行緒數量 num.recovery.threads.per.data.dir=1 #segment檔案保留的最長時間,超時將被刪除 log.retention.hours=1 #滾動生成新的segment檔案的最大時間 log.roll.hours=1 #日誌檔案中每個segment的大小,預設為1G log.segment.bytes=1073741824 #週期性檢查檔案大小的時間 log.retention.check.interval.ms=300000 #日誌清理是否開啟 log.cleaner.enable=true #broker需要使用zookeeper儲存meta資料 zookeeper.connect=node01:2181,node02:2181,node03:2181 #zookeeper連結超時時間 zookeeper.connection.timeout.ms=6000 #partion buffer中,訊息的條數達到閾值,將觸發flush到磁碟 log.flush.interval.messages=10000 #訊息buffer的時間,達到閾值,將觸發flush到磁碟 log.flush.interval.ms=3000 #刪除topic需要server.properties中設定delete.topic.enable=true否則只是標記刪除 delete.topic.enable=true #此處的host.name為本機IP(重要),如果不改,則客戶端會丟擲:Producer connection to localhost:9092 unsuccessful 錯誤! host.name=node01 #如果我們需要通過公網連線我們的kafka的時候,才需要配置這個,一般我們的kafka不會放在公網裡面 #advertised.host.name=192.168.140.128
生產者配置檔案
#指定kafka節點列表,用於獲取metadata,不必全部指定 metadata.broker.list=node01:9092,node02:9092,node03:9092 # 指定分割槽處理類。預設kafka.producer.DefaultPartitioner,表通過key雜湊到對應分割槽 #partitioner.class=kafka.producer.DefaultPartitioner # 是否壓縮,預設0表示不壓縮,1表示用gzip壓縮,2表示用snappy壓縮。壓縮後訊息中會有頭來指明訊息壓縮型別,故在消費者端訊息解壓是透明的無需指定。 compression.codec=none # 指定序列化處理類 serializer.class=kafka.serializer.DefaultEncoder # 如果要壓縮訊息,這裡指定哪些topic要壓縮訊息,預設empty,表示不壓縮。 #compressed.topics= # 設定傳送資料是否需要服務端的反饋,有三個值0,1,-1 # 0: producer不會等待broker傳送ack # 1: 當leader接收到訊息之後傳送ack # -1: 當所有的follower都同步訊息成功後傳送ack. request.required.acks=1 # 在向producer傳送ack之前,broker允許等待的最大時間 ,如果超時,broker將會向producer傳送一個error ACK.意味著上一次訊息因為某種原因未能成功(比如follower未能同步成功) request.timeout.ms=10000 # 同步還是非同步傳送訊息,預設“sync”表同步,"async"表非同步。非同步可以提高發送吞吐量, 也意味著訊息將會在本地buffer中,並適時批量傳送,但是也可能導致丟失未傳送過去的訊息 producer.type=async # 在async模式下,當message被快取的時間超過此值後,將會批量傳送給broker,預設為5000ms # 此值和batch.num.messages協同工作. queue.buffering.max.ms = 5000 # 在async模式下,producer端允許buffer的最大訊息量 # 無論如何,producer都無法儘快的將訊息傳送給broker,從而導致訊息在producer端大量沉積 # 此時,如果訊息的條數達到閥值,將會導致producer端阻塞或者訊息被拋棄,預設為10000 queue.buffering.max.messages=20000 # 如果是非同步,指定每次批量傳送資料量,預設為200 batch.num.messages=500 # 當訊息在producer端沉積的條數達到"queue.buffering.max.meesages"後 # 阻塞一定時間後,佇列仍然沒有enqueue(producer仍然沒有傳送出任何訊息) # 此時producer可以繼續阻塞或者將訊息拋棄,此timeout值用於控制"阻塞"的時間 # -1: 無阻塞超時限制,訊息不會被拋棄 # 0:立即清空佇列,訊息被拋棄 queue.enqueue.timeout.ms=-1 # 當producer接收到error ACK,或者沒有接收到ACK時,允許訊息重發的次數 # 因為broker並沒有完整的機制來避免訊息重複,所以當網路異常時(比如ACK丟失) # 有可能導致broker接收到重複的訊息,預設值為3. message.send.max.retries=3 # producer重新整理topic metada的時間間隔,producer需要知道partition leader的位置,以及當前topic的情況 # 因此producer需要一個機制來獲取最新的metadata,當producer遇到特定錯誤時,將會立即重新整理 # (比如topic失效,partition丟失,leader失效等),此外也可以通過此引數來配置額外的重新整理機制,預設值600000 topic.metadata.refresh.interval.ms=60000
消費者配置檔案說明
# zookeeper連線伺服器地址 zookeeper.connect=node01:2181,node02:2181,node03:2181 # zookeeper的session過期時間,預設5000ms,用於檢測消費者是否掛掉 zookeeper.session.timeout.ms=5000 #當消費者掛掉,其他消費者要等該指定時間才能檢查到並且觸發重新負載均衡 zookeeper.connection.timeout.ms=10000 # 指定多久消費者更新offset到zookeeper中。注意offset更新時基於time而不是每次獲得的訊息。一旦在更新zookeeper發生異常並重啟,將可能拿到已拿到過的訊息 zookeeper.sync.time.ms=2000 #指定消費 group.id=fgm # 當consumer消費一定量的訊息之後,將會自動向zookeeper提交offset資訊 # 注意offset資訊並不是每消費一次訊息就向zk提交一次,而是現在本地儲存(記憶體),並定期提交,預設為true auto.commit.enable=true # 自動更新時間。預設60 * 1000 auto.commit.interval.ms=1000 # 當前consumer的標識,可以設定,也可以有系統生成,主要用來跟蹤訊息消費情況,便於觀察 conusmer.id=xxx # 消費者客戶端編號,用於區分不同客戶端,預設客戶端程式自動產生 client.id=xxxx # 最大取多少塊快取到消費者(預設10) queued.max.message.chunks=50 # 當有新的consumer加入到group時,將會reblance,此後將會有partitions的消費端遷移到新 的consumer上,如果一個consumer獲得了某個partition的消費許可權,那麼它將會向zk註冊 "Partition Owner registry"節點資訊,但是有可能此時舊的consumer尚沒有釋放此節點, 此值用於控制,註冊節點的重試次數. rebalance.max.retries=5 # 獲取訊息的最大尺寸,broker不會像consumer輸出大於此值的訊息chunk 每次feth將得到多條訊息,此值為總大小,提升此值,將會消耗更多的consumer端記憶體 fetch.min.bytes=6553600 # 當訊息的尺寸不足時,server阻塞的時間,如果超時,訊息將立即傳送給consumer fetch.wait.max.ms=5000 socket.receive.buffer.bytes=655360 # 如果zookeeper沒有offset值或offset值超出範圍。那麼就給個初始的offset。有smallest、largest、anything可選,分別表示給當前最小的offset、當前最大的offset、拋異常。預設largest auto.offset.reset=smallest # 指定序列化處理類 derializer.class=kafka.serializer.DefaultDecoder
配置檔案均在kafka的安裝包下的config目錄下:/export/servers/kafka_2.11-1.0.0/config 當然這麼多配置不用全都配的,根據自己的需求進行配置即可。