系統引數

#唯一標識在叢集中的ID，要求是正數。 broker.id=0 #服務埠，預設9092 port=9092 #監聽地址 host.name=debugo01

# 處理網路請求的最大執行緒數 num.network.threads=2 # 處理磁碟I/O的執行緒數 num.io.threads=8 # 一些後臺執行緒數 background.threads = 4 # 等待IO執行緒處理的請求佇列最大數 queued.max.requests = 500

#  socket的傳送緩衝區（SO_SNDBUF） socket.send.buffer.bytes=1048576 # socket的接收緩衝區 (SO_RCVBUF) socket.receive.buffer.bytes=1048576 # socket請求的最大位元組數。為了防止記憶體溢位，message.max.bytes必然要小於 socket.request.max.bytes = 104857600

Topic引數

# 每個topic的分割槽個數，更多的partition會產生更多的segment file num.partitions=2 # 是否允許自動建立topic ，若是false，就需要通過命令建立topic auto.create.topics.enable =true # 一個topic ，預設分割槽的replication個數 ，不能大於叢集中broker的個數。 default.replication.factor =1 # 訊息體的最大大小，單位是位元組 message.max.bytes = 1000000

ZooKeeper引數

# Zookeeper quorum設定。如果有多個使用逗號分割 zookeeper.connect=debugo01:2181,debugo02,debugo03 # 連線zk的超時時間 zookeeper.connection.timeout.ms=1000000 # ZooKeeper叢集中leader和follower之間的同步實際 zookeeper.sync.time.ms = 2000

日誌引數

#日誌存放目錄，多個目錄使用逗號分割 log.dirs=/var/log/kafka

# 當達到下面的訊息數量時，會將資料flush到日誌檔案中。預設10000 #log.flush.interval.messages=10000 # 當達到下面的時間(ms)時，執行一次強制的flush操作。interval.ms和interval.messages無論哪個達到，都會flush。預設3000ms #log.flush.interval.ms=1000 # 檢查是否需要將日誌flush的時間間隔 log.flush.scheduler.interval.ms = 3000

# 日誌清理策略（delete|compact） log.cleanup.policy = delete # 日誌儲存時間 (hours|minutes)，預設為7天（168小時）。超過這個時間會根據policy處理資料。bytes和minutes無論哪個先達到都會觸發。 log.retention.hours=168 # 日誌資料儲存的最大位元組數。超過這個時間會根據policy處理資料。 #log.retention.bytes=1073741824

# 控制日誌segment檔案的大小，超出該大小則追加到一個新的日誌segment檔案中（-1表示沒有限制） log.segment.bytes=536870912 # 當達到下面時間，會強制新建一個segment log.roll.hours = 24*7 # 日誌片段檔案的檢查週期，檢視它們是否達到了刪除策略的設定（log.retention.hours或log.retention.bytes） log.retention.check.interval.ms=60000

# 是否開啟壓縮 log.cleaner.enable=false # 對於壓縮的日誌保留的最長時間 log.cleaner.delete.retention.ms = 1 day

# 對於segment日誌的索引檔案大小限制 log.index.size.max.bytes = 10 * 1024 * 1024 #y索引計算的一個緩衝區，一般不需要設定。 log.index.interval.bytes = 4096

副本引數

# partition management controller 與replicas之間通訊的超時時間 controller.socket.timeout.ms = 30000 # controller-to-broker-channels訊息佇列的尺寸大小 controller.message.queue.size=10 # replicas響應leader的最長等待時間，若是超過這個時間，就將replicas排除在管理之外 replica.lag.time.max.ms = 10000 # 是否允許控制器關閉broker ,若是設定為true,會關閉所有在這個broker上的leader，並轉移到其他broker controlled.shutdown.enable = false # 控制器關閉的嘗試次數 controlled.shutdown.max.retries = 3 # 每次關閉嘗試的時間間隔 controlled.shutdown.retry.backoff.ms = 5000

# 如果relicas落後太多,將會認為此partition relicas已經失效。而一般情況下,因為網路延遲等原因,總會導致replicas中訊息同步滯後。如果訊息嚴重滯後,leader將認為此relicas網路延遲較大或者訊息吞吐能力有限。在broker數量較少,或者網路不足的環境中,建議提高此值. replica.lag.max.messages = 4000 #leader與relicas的socket超時時間 replica.socket.timeout.ms= 30 * 1000 # leader複製的socket快取大小 replica.socket.receive.buffer.bytes=64 * 1024 # replicas每次獲取資料的最大位元組數 replica.fetch.max.bytes = 1024 * 1024 # replicas同leader之間通訊的最大等待時間，失敗了會重試 replica.fetch.wait.max.ms = 500 # 每一個fetch操作的最小資料尺寸,如果leader中尚未同步的資料不足此值,將會等待直到資料達到這個大小 replica.fetch.min.bytes =1 # leader中進行復制的執行緒數，增大這個數值會增加relipca的IO num.replica.fetchers = 1 # 每個replica將最高水位進行flush的時間間隔 replica.high.watermark.checkpoint.interval.ms = 5000

# 是否自動平衡broker之間的分配策略 auto.leader.rebalance.enable = false # leader的不平衡比例，若是超過這個數值，會對分割槽進行重新的平衡 leader.imbalance.per.broker.percentage = 10 # 檢查leader是否不平衡的時間間隔 leader.imbalance.check.interval.seconds = 300 # 客戶端保留offset資訊的最大空間大小 offset.metadata.max.bytes = 1024

消費者引數

# Consumer端核心的配置是group.id、zookeeper.connect # 決定該Consumer歸屬的唯一組ID，By setting the same group id multiple processes indicate that they are all part of the same consumer group. group.id # 消費者的ID，若是沒有設定的話，會自增 consumer.id # 一個用於跟蹤調查的ID ，最好同group.id相同 client.id = <group_id>

# 對於zookeeper叢集的指定，必須和broker使用同樣的zk配置 zookeeper.connect=debugo01:2182,debugo02:2182,debugo03:2182 # zookeeper的心跳超時時間，查過這個時間就認為是無效的消費者 zookeeper.session.timeout.ms = 6000 # zookeeper的等待連線時間 zookeeper.connection.timeout.ms = 6000 # zookeeper的follower同leader的同步時間 zookeeper.sync.time.ms = 2000 # 當zookeeper中沒有初始的offset時，或者超出offset上限時的處理方式 。 # smallest ：重置為最小值 # largest:重置為最大值 # anything else：丟擲異常給consumer auto.offset.reset = largest

# socket的超時時間，實際的超時時間為max.fetch.wait + socket.timeout.ms. socket.timeout.ms= 30 * 1000 # socket的接收快取空間大小 socket.receive.buffer.bytes=64 * 1024 #從每個分割槽fetch的訊息大小限制 fetch.message.max.bytes = 1024 * 1024

# true時，Consumer會在消費訊息後將offset同步到zookeeper，這樣當Consumer失敗後，新的consumer就能從zookeeper獲取最新的offset auto.commit.enable = true # 自動提交的時間間隔 auto.commit.interval.ms = 60 * 1000

# 用於消費的最大數量的訊息塊緩衝大小，每個塊可以等同於fetch.message.max.bytes中數值 queued.max.message.chunks = 10

# 當有新的consumer加入到group時,將嘗試reblance,將partitions的消費端遷移到新的consumer中, 該設定是嘗試的次數 rebalance.max.retries = 4 # 每次reblance的時間間隔 rebalance.backoff.ms = 2000 # 每次重新選舉leader的時間 refresh.leader.backoff.ms

# server傳送到消費端的最小資料，若是不滿足這個數值則會等待直到滿足指定大小。預設為1表示立即接收。 fetch.min.bytes = 1 # 若是不滿足fetch.min.bytes時，等待消費端請求的最長等待時間 fetch.wait.max.ms = 100 # 如果指定時間內沒有新訊息可用於消費，就丟擲異常，預設-1表示不受限 consumer.timeout.ms = -1

生產者引數

# 核心的配置包括： # metadata.broker.list # request.required.acks # producer.type # serializer.class

# 消費者獲取訊息元資訊(topics, partitions and replicas)的地址,配置格式是：host1:port1,host2:port2，也可以在外面設定一個vip metadata.broker.list

#訊息的確認模式 # 0：不保證訊息的到達確認，只管傳送，低延遲但是會出現訊息的丟失，在某個server失敗的情況下，有點像TCP # 1：傳送訊息，並會等待leader 收到確認後，一定的可靠性 # -1：傳送訊息，等待leader收到確認，並進行復制操作後，才返回，最高的可靠性 request.required.acks = 0

# 訊息傳送的最長等待時間 request.timeout.ms = 10000 # socket的快取大小 send.buffer.bytes=100*1024 # key的序列化方式，若是沒有設定，同serializer.class key.serializer.class # 分割槽的策略，預設是取模 partitioner.class=kafka.producer.DefaultPartitioner # 訊息的壓縮模式，預設是none，可以有gzip和snappy compression.codec = none # 可以針對默寫特定的topic進行壓縮 compressed.topics=null # 訊息傳送失敗後的重試次數 message.send.max.retries = 3 # 每次失敗後的間隔時間 retry.backoff.ms = 100 # 生產者定時更新topic元資訊的時間間隔 ，若是設定為0，那麼會在每個訊息傳送後都去更新資料 topic.metadata.refresh.interval.ms = 600 * 1000 # 使用者隨意指定，但是不能重複，主要用於跟蹤記錄訊息 client.id=””

# 非同步模式下緩衝資料的最大時間。例如設定為100則會集合100ms內的訊息後傳送，這樣會提高吞吐量，但是會增加訊息傳送的延時 queue.buffering.max.ms = 5000 # 非同步模式下緩衝的最大訊息數，同上 queue.buffering.max.messages = 10000 # 非同步模式下，訊息進入佇列的等待時間。若是設定為0，則訊息不等待，如果進入不了佇列，則直接被拋棄 queue.enqueue.timeout.ms = -1 # 非同步模式下，每次傳送的訊息數，當queue.buffering.max.messages或queue.buffering.max.ms滿足條件之一時producer會觸發傳送。 batch.num.messages=200

注：本文中的引數不一定是最新版本的Kafka。

本文轉自：http://debugo.com/kafka-params/