kafka深度研究之路(4)-kafka和zk配置文件詳細說明（來龍去脈）之kafka配置

2/kafka配置文件參數(shù)詳解  默認必須配置的參數(shù) 
默認 kafka server.properties  配置如下：
############################# Server Basics #############################          # 服務器基礎(chǔ)知識

# The id of the broker. This must be set to a unique integer for each broker.      # 必須為每個代理設(shè)置一個唯一的整數(shù)
broker.id=0

############################# Socket Server Settings #############################   # 套接字服務器設(shè)置

# 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                                       # 服務器用于從網(wǎng)絡接收請求并向網(wǎng)絡發(fā)送響應的線程數(shù)  默認是3 

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8                                               # 服務器用于處理請求的線程數(shù)，可能包括磁盤I / O.  默認是 8

# The send buffer (SO_SNDBUF) used by the socket server         ＃套接字服務器使用的發(fā)送緩沖區(qū)（SO_SNDBUF）
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server      ＃套接字服務器使用的接收緩沖區(qū)（SO_RCVBUF）
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)         # 套接字服務器將接受的請求的最大大?。ǚ乐筄OM）
socket.request.max.bytes=104857600

############################# Log Basics #############################   日志基礎(chǔ)

# 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    ＃每個主題的默認日志分區(qū)數(shù)。更多分區(qū)允許更大
# parallelism for consumption, but this will also result in more files across      #dileism for consumption，但這也會導致更多的文件
# the brokers.
num.partitions=1                                                                   # 建議broker少的話，默認就幾個broker 就設(shè)置成幾個分區(qū)

＃在啟動時用于日志恢復和在關(guān)閉時刷新的每個數(shù)據(jù)目錄的線程數(shù)。
＃對于數(shù)據(jù)目錄位于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  #############################   內(nèi)部主題設(shè)置
＃組元數(shù)據(jù)內(nèi)部主題“__consumer_offsets”和“__transaction_state”的復制因子
＃對于除開發(fā)測試之外的任何其他內(nèi)容，建議使用大于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
//關(guān)于這3個參數(shù)，可以在修改kafka程序中指定的 __consumer_offsets 的副本數(shù) 
然后@上海-馬吉輝 說只要num.partitions=3，__consumer_offsets副本數(shù)就是3，我測試不是 還是1
所以還是以offsets.topic.replication.factor參數(shù)控制為準
如果不是第一次啟動kafka  那幾個配置只有在初次啟動生效的。 apache kafka 下載下來應該都默認是 1 吧，2.* 也是 1 啊。
可以這樣修改
先停止kafka集群，刪除每個broker  data目錄下所有__consumer_offsets_*
然后刪除zookeeper下rmr /kafkatest/brokers/topics/__consumer_offsets    然后重啟kafka
消費一下，這個__consumer_offsets就會創(chuàng)建了
注意：是在第一次消費時，才創(chuàng)建這個topic的，不是broker集群啟動就創(chuàng)建，還有那個__trancation_state  topic也是第一次使用事務的時候才會創(chuàng)建

小結(jié)：在生產(chǎn)上，沒人去刪zk里的內(nèi)容，危險系數(shù)大，還是推薦動態(tài)擴副本，只要把json寫對就好

############################# Log Flush Policy #############################    日志刷新政策
＃消息立即寫入文件系統(tǒng)，但默認情況下我們只有fsync（）才能同步
＃懶惰的操作系統(tǒng)緩存。以下配置控制將數(shù)據(jù)刷新到磁盤。
＃這里有一些重要的權(quán)衡：
＃1。持久性：如果您不使用復制，則可能會丟失未刷新的數(shù)據(jù)。
＃2。延遲：當刷新確實發(fā)生時，非常大的刷新間隔可能會導致延遲峰值，因為會有大量數(shù)據(jù)需要刷新。
＃3。吞吐量：沖洗通常是最昂貴的操作，并且小的沖洗間隔可能導致過多的搜索。
＃以下設(shè)置允許配置刷新策略以在一段時間后刷新數(shù)據(jù)或
＃每N條消息（或兩者）。這可以在全局范圍內(nèi)完成，并在每個主題的基礎(chǔ)上進行覆蓋。

# 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   ＃強制刷新數(shù)據(jù)到磁盤之前要接受的消息數(shù)
#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 #############################   日志保留政策
＃以下配置控制日志段的處理。政策可以
＃設(shè)置為在一段時間后或在累積給定大小后刪除段。
＃只要滿足這些條件* *，就會刪除一個段。刪除總是發(fā)生
＃從日志的末尾開始。

# 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

＃日志段文件的最大大小。達到此大小時，將創(chuàng)建新的日志段。
# 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 #############################   組協(xié)調(diào)員設(shè)置
＃以下配置指定GroupCoordinator將延遲初始消費者重新平衡的時間（以毫秒為單位）。
＃當新成員加入組時，重新平衡將進一步延遲group.initial.rebalance.delay.ms的值，最多為max.poll.interval.ms。
＃默認值為3秒。
＃我們將此覆蓋為0，因為它為開發(fā)和測試提供了更好的開箱即用體驗。
＃但是，在生產(chǎn)環(huán)境中，默認值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的擴展參數(shù) 抓重點說明
background.threads =4                           # 一些后臺任務處理的線程數(shù)，例如過期消息文件的刪除等，一般情況下不需要去做修改
queued.max.requests =500                          # 等待IO線程處理的請求隊列最大數(shù)，若是等待IO的請求超過這個數(shù)值，那么會停止接受外部消息，應該是一種自我保護機制。
controller.socket.timeout.ms =30000             # partition leader與replicas之間通訊時,socket的超時時間 
controller.message.queue.size=10                # partition leader與replicas數(shù)據(jù)同步時,消息的隊列尺寸
replica.lag.time.max.ms =10000                  # replicas響應partition leader的最長等待時間，若是超過這個時間，就將replicas列入ISR(in-sync replicas)，并認為它是死的，不會再加入管理中 
replica.lag.max.messages =4000                  # 如果follower落后與leader太多,將會認為此follower[或者說partition?relicas]已經(jīng)失效
                                                ##通常,在follower與leader通訊時,因為網(wǎng)絡延遲或者鏈接斷開,總會導致replicas中消息同步滯后
                                                ##如果消息之后太多,leader將認為此follower網(wǎng)絡延遲較大或者消息吞吐能力有限,將會把此replicas遷移
                                                ##到其他follower中.
                                                ##在broker數(shù)量較少,或者網(wǎng)絡不足的環(huán)境中,建議提高此值.
                                                // Leader會跟蹤與其保持同步的Replica列表，該列表稱為ISR（即in-sync Replica）。如果一個Follower宕機，或者落后太多，Leader將把它從ISR中移除。這里所描述的“落后太多”指Follower復制的消息落后于Leader后的條數(shù)超過預定值（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.max.messages配置，其默認值是4000）或者Follower超過一定時間（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.time.max.ms來配置，其默認值是10000）未向Leader發(fā)送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每次獲取數(shù)據(jù)的最大大小 
replica.fetch.wait.max.ms =500                  # replicas同leader之間通信的最大等待時間，失敗了會重試
replica.fetch.min.bytes =1                      # fetch的最小數(shù)據(jù)尺寸,如果leader中尚未同步的數(shù)據(jù)不足此值,將會阻塞,直到滿足條件
num.replica.fetchers=1                          # leader進行復制的線程數(shù)，增大這個數(shù)值會增加follower的IO
replica.high.watermark.checkpoint.interval.ms =5000   # 每個replica檢查是否將最高水位進行固化的頻率
leader.imbalance.per.broker.percentage =10     # leader的不平衡比例，若是超過這個數(shù)值，會對分區(qū)進行重新的平衡
leader.imbalance.check.interval.seconds =300   # 檢查leader是否不平衡的時間間隔
zookeeper.connect = localhost:2181             # zookeeper集群的地址，可以是多個，多個之間用逗號分割 hostname1:port1,hostname2:port2,hostname3:port3
zookeeper.session.timeout.ms=6000              # ZooKeeper的最大超時時間，就是心跳的間隔，若是沒有反映，那么認為已經(jīng)死了，不易過大
zookeeper.connection.timeout.ms =6000          # ZooKeeper的連接超時時間
zookeeper.sync.time.ms =2000                   # ZooKeeper集群中l(wèi)eader和follower之間的同步時間

###############################################
grep '^[a-Z]' server.properties 
broker.id=1                             # //當前機器在集群中的唯一標識，和zookeeper的myid性質(zhì)一樣
host.name=10.9.39.110                   # 這個參數(shù)默認是關(guān)閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。 盡量寫ip
num.network.threads=8                   # 這個是borker進行網(wǎng)絡處理的線程數(shù) 一般num.network.threads主要處理網(wǎng)絡io，讀寫緩沖區(qū)數(shù)據(jù)，基本沒有io等待，配置線程數(shù)量為cpu核數(shù)加1
num.io.threads=16                       # num.io.threads主要進行磁盤io操作，高峰期可能有些io等待，因此配置需要大些。配置線程數(shù)量為cpu核數(shù)2倍，最大不超過3倍  
socket.send.buffer.bytes=102400         # 發(fā)送緩沖區(qū)buffer大小，數(shù)據(jù)不是一下子就發(fā)送的，先回存儲到緩沖區(qū)了到達一定的大小后在發(fā)送，能提高性能   100kb （發(fā)送緩沖區(qū)）推薦1M
socket.receive.buffer.bytes=102400      # kafka接收緩沖區(qū)大小，當數(shù)據(jù)到達一定大小后在序列化到磁盤   100kb   （接收緩沖區(qū)）  推薦1M
socket.request.max.bytes=104857600      # 這個參數(shù)是向kafka請求消息或者向kafka發(fā)送消息的請請求的最大數(shù)，這個值不能超過java的堆棧大小 104857600B =  100M （防止oom）
log.dirs=/data/kafka/kafka-logs         # 消息存放的目錄，這個目錄可以配置為“，”逗號分割的表達式，上面的num.io.threads要大于這個目錄的個數(shù)，
                                        //這個目錄如果配置多個目錄，新創(chuàng)建的topic他把消息持久化的地方是，當前以逗號分割的目錄中，那個分區(qū)數(shù)最少就放那一個
num.partitions=1                        # 默認的分區(qū)數(shù)，一個topic默認1個分區(qū)數(shù)  我建議根據(jù)brocker數(shù)設(shè)置 broker有3個 就設(shè)置成默認分區(qū)為3
num.recovery.threads.per.data.dir=1     # 每個數(shù)據(jù)目錄用來日志恢復的線程數(shù)目 對于數(shù)據(jù)目錄位于RAID陣列中的安裝，建議增加此值。 一般保持默認
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1 
//以上3個推薦＃組元數(shù)據(jù)內(nèi)部主題“__consumer_offsets”和“__transaction_state”的復制因子對于除開發(fā)測試之外的任何其他內(nèi)容，建議使用大于1的值以確?？捎眯?，例如3。
#log.flush.interval.messages=10000       # 強制刷新數(shù)據(jù)到磁盤之前要接受的消息數(shù) 
#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            # 這個參數(shù)是：因為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  # 設(shè)置zookeeper的連接端口
zookeeper.connection.timeout.ms=60000   # 設(shè)置zookeeper的連接超時時間 
group.initial.rebalance.delay.ms=3      # 以下配置指定GroupCoordinator將延遲初始消費者重新平衡的時間（以毫秒為單位）。 官方推薦成 3 

########################################

分享標題：kafka深度研究之路(4)-kafka和zk配置文件詳細說明（來龍去脈）之kafka配置
新聞來源：http://bm7419.com/article10/igsjdo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站建設(shè)、微信小程序、App開發(fā)、手機網(wǎng)站建設(shè)、響應式網(wǎng)站、商城網(wǎng)站

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容