默認 kafka server.properties 配置如下： ############################# Server Basics ############################# # 服務器基礎知識 # The id of the broker. This must be set to a unique integer for each broker. # 必須為每個代理設置一個唯一的整數 broker.id=0 ############################# Socket Server Settings ############################# # 套接字服務器設置 # The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 #listeners=PLAINTEXT://:9092 # Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured. Otherwise, it will use the value # returned from java.net.InetAddress.getCanonicalHostName(). #advertised.listeners=PLAINTEXT://your.host.name:9092 # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL # The number of threads that the server uses for receiving requests from the network and sending responses to the network num.network.threads=3 # 服務器用於從網絡接收請求並向網絡發送響應的線程數 默認是3 # The number of threads that the server uses for processing requests, which may include disk I/O num.io.threads=8 # 服務器用於處理請求的線程數，可能包括磁盤I / O. 默認是 8 # The send buffer (SO_SNDBUF) used by the socket server ＃套接字服務器使用的發送緩沖區（SO_SNDBUF） socket.send.buffer.bytes=102400 # The receive buffer (SO_RCVBUF) used by the socket server ＃套接字服務器使用的接收緩沖區（SO_RCVBUF） socket.receive.buffer.bytes=102400 # The maximum size of a request that the socket server will accept (protection against OOM) # 套接字服務器將接受的請求的最大大小（防止OOM） socket.request.max.bytes=104857600 ############################# Log Basics ############################# 日志基礎 # A comma separated list of directories under which to store log files ＃逗號分隔的目錄列表，用於存儲日志文件 log.dirs=/tmp/kafka-logs # The default number of log partitions per topic. More partitions allow greater ＃每個主題的默認日志分區數。更多分區允許更大 # parallelism for consumption, but this will also result in more files across #dileism for consumption，但這也會導致更多的文件 # the brokers. num.partitions=1 # 建議broker少的話，默認就幾個broker 就設置成幾個分區 ＃在啟動時用於日志恢復和在關閉時刷新的每個數據目錄的線程數。 ＃對於數據目錄位於RAID陣列中的安裝，建議增加此值。 # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. # This value is recommended to be increased for installations with data dirs located in RAID array. num.recovery.threads.per.data.dir=1 ############################# Internal Topic Settings ############################# 內部主題設置 ＃組元數據內部主題“__consumer_offsets”和“__transaction_state”的復制因子 ＃對於除開發測試之外的任何其他內容，建議使用大於1的值以確保可用性，例如3。 # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state" # For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3. offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 //關於這3個參數，可以在修改kafka程序中指定的 __consumer_offsets 的副本數 然后@上海-馬吉輝 說只要num.partitions=3，__consumer_offsets副本數就是3，我測試不是 還是1 所以還是以offsets.topic.replication.factor參數控制為准 如果不是第一次啟動kafka 那幾個配置只有在初次啟動生效的。 apache kafka 下載下來應該都默認是 1 吧，2.* 也是 1 啊。 可以這樣修改 先停止kafka集群，刪除每個broker data目錄下所有__consumer_offsets_* 然后刪除zookeeper下rmr /kafkatest/brokers/topics/__consumer_offsets 然后重啟kafka 消費一下，這個__consumer_offsets就會創建了 注意：是在第一次消費時，才創建這個topic的，不是broker集群啟動就創建，還有那個__trancation_state topic也是第一次使用事務的時候才會創建 小結：在生產上，沒人去刪zk里的內容，危險系數大，還是推薦動態擴副本，只要把json寫對就好 ############################# Log Flush Policy ############################# 日志刷新政策 ＃消息立即寫入文件系統，但默認情況下我們只有fsync（）才能同步 ＃懶惰的操作系統緩存。以下配置控制將數據刷新到磁盤。 ＃這里有一些重要的權衡： ＃1。持久性：如果您不使用復制，則可能會丟失未刷新的數據。 ＃2。延遲：當刷新確實發生時，非常大的刷新間隔可能會導致延遲峰值，因為會有大量數據需要刷新。 ＃3。吞吐量：沖洗通常是最昂貴的操作，並且小的沖洗間隔可能導致過多的搜索。 ＃以下設置允許配置刷新策略以在一段時間后刷新數據或 ＃每N條消息（或兩者）。這可以在全局范圍內完成，並在每個主題的基礎上進行覆蓋。 # Messages are immediately written to the filesystem but by default we only fsync() to sync # the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs here: # 1. Durability: Unflushed data may be lost if you are not using replication. # 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush. # 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or # every N messages (or both). This can be done globally and overridden on a per-topic basis. # The number of messages to accept before forcing a flush of data to disk ＃強制刷新數據到磁盤之前要接受的消息數 #log.flush.interval.messages=10000 # The maximum amount of time a message can sit in a log before we force a flush ＃強制刷新之前消息可以在日志中停留的最長時間 #log.flush.interval.ms=1000 ############################# Log Retention Policy ############################# 日志保留政策 ＃以下配置控制日志段的處理。政策可以 ＃設置為在一段時間后或在累積給定大小后刪除段。 ＃只要滿足這些條件* *，就會刪除一個段。刪除總是發生 ＃從日志的末尾開始。 # The following configurations control the disposal of log segments. The policy can # be set to delete segments after a period of time, or after a given size has accumulated. # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens # from the end of the log. # The minimum age of a log file to be eligible for deletion due to age ＃由於年齡原因有資格刪除的日志文件的最小年齡 log.retention.hours=168 ＃日志的基於大小的保留策略。除非剩下，否則將從日志中刪除段 ＃segments落在log.retention.bytes之下。功能獨立於log.retention.hours。 ＃log.retention.bytes = 1073741824 # A size-based retention policy for logs. Segments are pruned from the log unless the remaining # segments drop below log.retention.bytes. Functions independently of log.retention.hours. #log.retention.bytes=1073741824 ＃日志段文件的最大大小。達到此大小時，將創建新的日志段。 # The maximum size of a log segment file. When this size is reached a new log segment will be created. log.segment.bytes=1073741824 ＃檢查日志段以查看是否可以刪除日志段的時間間隔 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies log.retention.check.interval.ms=300000 ############################# Zookeeper ############################# # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=localhost:2181 # zookeeper集群的地址，可以是多個，多個之間用逗號分割 hostname1:port1,hostname2:port2,hostname3:port3 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000 # ZooKeeper的連接超時時間 ############################# Group Coordinator Settings ############################# 組協調員設置 ＃以下配置指定GroupCoordinator將延遲初始消費者重新平衡的時間（以毫秒為單位）。 ＃當新成員加入組時，重新平衡將進一步延遲group.initial.rebalance.delay.ms的值，最多為max.poll.interval.ms。 ＃默認值為3秒。 ＃我們將此覆蓋為0，因為它為開發和測試提供了更好的開箱即用體驗。 ＃但是，在生產環境中，默認值3秒更合適，因為這有助於避免在應用程序啟動期間不必要且可能很昂貴的重新平衡。 group.initial.rebalance.delay.ms = 0 # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance. # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms. # The default value for this is 3 seconds. # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing. # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup. group.initial.rebalance.delay.ms=0 kafka的擴展參數 抓重點說明 background.threads =4 # 一些后台任務處理的線程數，例如過期消息文件的刪除等，一般情況下不需要去做修改 queued.max.requests =500 # 等待IO線程處理的請求隊列最大數，若是等待IO的請求超過這個數值，那么會停止接受外部消息，應該是一種自我保護機制。 controller.socket.timeout.ms =30000 # partition leader與replicas之間通訊時,socket的超時時間 controller.message.queue.size=10 # partition leader與replicas數據同步時,消息的隊列尺寸 replica.lag.time.max.ms =10000 # replicas響應partition leader的最長等待時間，若是超過這個時間，就將replicas列入ISR(in-sync replicas)，並認為它是死的，不會再加入管理中 replica.lag.max.messages =4000 # 如果follower落后與leader太多,將會認為此follower[或者說partition relicas]已經失效 ##通常,在follower與leader通訊時,因為網絡延遲或者鏈接斷開,總會導致replicas中消息同步滯后 ##如果消息之后太多,leader將認為此follower網絡延遲較大或者消息吞吐能力有限,將會把此replicas遷移 ##到其他follower中. ##在broker數量較少,或者網絡不足的環境中,建議提高此值. // Leader會跟蹤與其保持同步的Replica列表，該列表稱為ISR（即in-sync Replica）。如果一個Follower宕機，或者落后太多，Leader將把它從ISR中移除。這里所描述的“落后太多”指Follower復制的消息落后於Leader后的條數超過預定值（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.max.messages配置，其默認值是4000）或者Follower超過一定時間（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.time.max.ms來配置，其默認值是10000）未向Leader發送fetch請求。 replica.socket.timeout.ms=30*1000 # follower與leader之間的socket超時時間 replica.socket.receive.buffer.bytes=64*1024 # leader復制時候的socket緩存大小 建議 1048576 B = 1M replica.fetch.max.bytes =1024*1024 # replicas每次獲取數據的最大大小 replica.fetch.wait.max.ms =500 # replicas同leader之間通信的最大等待時間，失敗了會重試 replica.fetch.min.bytes =1 # fetch的最小數據尺寸,如果leader中尚未同步的數據不足此值,將會阻塞,直到滿足條件 num.replica.fetchers=1 # leader進行復制的線程數，增大這個數值會增加follower的IO replica.high.watermark.checkpoint.interval.ms =5000 # 每個replica檢查是否將最高水位進行固化的頻率 leader.imbalance.per.broker.percentage =10 # leader的不平衡比例，若是超過這個數值，會對分區進行重新的平衡 leader.imbalance.check.interval.seconds =300 # 檢查leader是否不平衡的時間間隔 zookeeper.connect = localhost:2181 # zookeeper集群的地址，可以是多個，多個之間用逗號分割 hostname1:port1,hostname2:port2,hostname3:port3 zookeeper.session.timeout.ms=6000 # ZooKeeper的最大超時時間，就是心跳的間隔，若是沒有反映，那么認為已經死了，不易過大 zookeeper.connection.timeout.ms =6000 # ZooKeeper的連接超時時間 zookeeper.sync.time.ms =2000 # ZooKeeper集群中leader和follower之間的同步時間 ############################################### grep '^[a-Z]' server.properties broker.id=1 # //當前機器在集群中的唯一標識，和zookeeper的myid性質一樣 host.name=10.9.39.110 # 這個參數默認是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。 盡量寫ip num.network.threads=8 # 這個是borker進行網絡處理的線程數 一般num.network.threads主要處理網絡io，讀寫緩沖區數據，基本沒有io等待，配置線程數量為cpu核數加1 num.io.threads=16 # num.io.threads主要進行磁盤io操作，高峰期可能有些io等待，因此配置需要大些。配置線程數量為cpu核數2倍，最大不超過3倍 socket.send.buffer.bytes=102400 # 發送緩沖區buffer大小，數據不是一下子就發送的，先回存儲到緩沖區了到達一定的大小后在發送，能提高性能 100kb （發送緩沖區）推薦1M socket.receive.buffer.bytes=102400 # kafka接收緩沖區大小，當數據到達一定大小后在序列化到磁盤 100kb （接收緩沖區） 推薦1M socket.request.max.bytes=104857600 # 這個參數是向kafka請求消息或者向kafka發送消息的請請求的最大數，這個值不能超過java的堆棧大小 104857600B = 100M （防止oom） log.dirs=/data/kafka/kafka-logs # 消息存放的目錄，這個目錄可以配置為“，”逗號分割的表達式，上面的num.io.threads要大於這個目錄的個數， //這個目錄如果配置多個目錄，新創建的topic他把消息持久化的地方是，當前以逗號分割的目錄中，那個分區數最少就放那一個 num.partitions=1 # 默認的分區數，一個topic默認1個分區數 我建議根據brocker數設置 broker有3個 就設置成默認分區為3 num.recovery.threads.per.data.dir=1 # 每個數據目錄用來日志恢復的線程數目 對於數據目錄位於RAID陣列中的安裝，建議增加此值。 一般保持默認 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 //以上3個推薦＃組元數據內部主題“__consumer_offsets”和“__transaction_state”的復制因子對於除開發測試之外的任何其他內容，建議使用大於1的值以確保可用性，例如3。 #log.flush.interval.messages=10000 # 強制刷新數據到磁盤之前要接受的消息數 #log.flush.interval.ms=1000 ＃ 強制刷新之前消息可以在日志中停留的最長時間 log.retention.hours=24 # 默認消息的最大持久化時間，168小時，7天 ＃log.retention.bytes = 1073741824 ＃ 日志的基於大小的保留策略。除非剩下，否則將從日志中刪除段 segments落在log.retention.bytes之下。功能獨立於log.retention.hours。 log.segment.bytes=1073741824 # 這個參數是：因為kafka的消息是以追加的形式落地到文件，當超過這個值的時候，kafka會新起一個文件 log.retention.check.interval.ms=300000 # 每隔300000毫秒去檢查上面配置的log失效時間 zookeeper.connect=10.9.39.110:2181,10.9.139.65:2181,10.9.35.206:2181,10.9.88.40:2181,10.9.74.126:2181/kafkagroup # 設置zookeeper的連接端口 zookeeper.connection.timeout.ms=60000 # 設置zookeeper的連接超時時間 group.initial.rebalance.delay.ms=3 # 以下配置指定GroupCoordinator將延遲初始消費者重新平衡的時間（以毫秒為單位）。 官方推薦成 3