文章目錄

• 1.1概述
• 1.2消息系統介紹
• • 1.2.1點對點消息傳遞
  • 1.2.2發布-訂閱消息傳遞
• 1.3 Kafka的優點
• 1.4 Kafka架構以及術語解釋
• • 1.4.1 Broker
  • 1.4.2 Topic
  • 1.4.3 Partition
  • 1.4.4 Producer
  • 1.4.5 Consumer
  • 1.4.6 Consumer Group
  • 1.4.7 Leader
  • 1.4.8 Follower
  • 1.4.9 Offset
  • 1.4.10 Segment
• 1.5 Kafka主要配置參數
• • 1.5.1 Broker Config
  • 1.5.2 Producer Config
  • 1.5.3 Consumer Config

1.1概述

Kafka最初是由Linkedin公司開發的，是一個分布式的、分區的、多副本的、多訂閱者、基于zookeeper協調的分布式日志系統，當然也可以當作MQ系統，常見的可以用作日志收集，消息服務等。Linkedin于2010年貢獻給了Apache基金并成為頂級開源項目。

主要應用場景是：日志收集系統和消息系統。
Kafka主要設計目標：

以時間復雜度為O(1)的方式提供持久化能力，即時對TB級別以上數據也能保證常數時間的訪問性能

高吞吐率。即時在非常廉價的商用機器上也能做到單機支持每秒100K條消息的傳輸。

支持Kafka Server間的消息分區，及分布式消費，同時保證每個partition內的消息順序傳輸。

同時支持離線數據處理和實時數據處理

支持在線水平擴展

1.2消息系統介紹

一個消息系統負責將數據從一個應用系統傳遞到另外一個應用，應用只需關注于數據，無需關注數據在兩個或多個應用間是如何傳遞的。分布式消息傳輸基于可靠的消息隊列，在客戶端應用和消息系統之間異步傳遞消息。
主要由兩種消息傳遞模式：

點對點傳遞模式

發布-訂閱模式

大部分的消息系統選用發布-訂閱模式。Kafka就是一種發布訂閱模式

1.2.1點對點消息傳遞

點對點消息系統中，消息持久化到一個隊列中。此時，將有一個或多個消費者消費隊列中的數據。但是一條消息只能被消費一次。當一個消費者消費了隊列中的某條數據之后，該數據則從消息隊列中刪除。該模式即使有多個消費者同時消費數據，也能保證數據的處理順序，示意圖如下：

生產這發送一條消息到Queue，只有一個消費者能收到。

1.2.2發布-訂閱消息傳遞

在發布-訂閱消息系統中，消息被持久化到一個topic中。于點對點消息系統不同的四，消費者可以訂閱一個或多個topic，消費者可以消費該topic中的所有的數據，同一條數據可以被多個消費者消費，數據被消費后不會立馬刪除。在發布-訂閱消息系統中。消息的生產者被稱為發布者，消費者被稱為訂閱者。模式示意圖如下；

發布者發送到topic的消息，只有訂閱了topic的訂閱者才會收到消息。

1.3 Kafka的優點

• 高吞吐量、低延遲：kafka每秒可以處理幾十萬條雄安錫，它的延遲最低只有幾毫秒
• 可擴展性：kafka集群支持熱擴展
• 持久性、可靠性：消息被持久化到本地磁盤，并且支持數據備份防止數據丟失
• 容錯性：允許集群中節點失敗（若副本數量為n，則允許n-個節點失敗）
• 高并發：支持數千個客戶端同時讀寫
  .

1.4 Kafka架構以及術語解釋

上圖中

1.4.1 Broker

Kafka 集群包含一個或多個服務器，服務器節點稱為broker。

broker存儲topic的數據。如果某topic有N個partition，集群有N個broker，那么每個broker存儲該topic的一個partition。

如果某topic有N個partition，集群有(N+M)個broker，那么其中有N個broker存儲該topic的一個partition，剩下的M個broker不存儲該topic的partition數據。

如果某topic有N個partition，集群中broker數目少于N個，那么一個broker存儲該topic的一個或多個partition。在實際生產環境中，盡量避免這種情況的發生，這種情況容易導致Kafka集群數據不均衡。

1.4.2 Topic

每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數據而不必關心數據存于何處）類似于數據庫的表名

1.4.3 Partition

topic中的數據分割為一個或多個partition。每個topic至少有一個partition。每個partition中的數據使用多個segment文件存儲。partition中的數據是有序的，不同partition間的數據丟失了數據的順序。如果topic有多個partition，消費數據時就不能保證數據的順序。在需要嚴格保證消息的消費順序的場景下，需要將partition數目設為1。

1.4.4 Producer

生產者即數據的發布者，該角色將消息發布到Kafka的topic中。broker接收到生產者發送的消息后，broker將該消息追加到當前用于追加數據的segment文件中。生產者發送的消息，存儲到一個partition中，生產者也可以指定數據存儲的partition。

1.4.5 Consumer

消費者可以從broker中讀取數據。消費者可以消費多個topic中的數據。

1.4.6 Consumer Group

每個Consumer屬于一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬于默認的group）。這是kafka用來實現一個topic消息的廣播（發給所有的consumer）和單播（發給任意一個consumer）的手段。一個topic可以有多個CG。topic的消息會復制-給consumer。如果需要實現廣播，只要每個consumer有一個獨立的CG就可以了。要實現單播只要所有的consumer在同一個CG。用CG還可以將consumer進行自由的分組而不需要多次發送消息到不同的topic。

1.4.7 Leader

每個partition有多個副本，其中有且僅有一個作為Leader，Leader是當前負責數據的讀寫的partition

1.4.8 Follower

Follower跟隨Leader，所有寫請求都通過Leader路由，數據變更會廣播給所有Follower，Follower與Leader保持數據同步。如果Leader失效，則從Follower中選舉出一個新的Leader。當Follower與Leader掛掉、卡住或者同步太慢，leader會把這個follower從“in sync replicas”（ISR）列表中刪除，重新創建一個Follower

1.4.9 Offset

kafka的存儲文件都是按照offset.kafka來命名，用offset做名字的好處是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。當然the first offset就是00000000000.kafka

1.4.10 Segment

每個partion(目錄)相當于一個巨型文件被平均分配到多個大小相等segment(段)數據文件中。但每個段segment file消息數量不一定相等，這種特性方便old segment file快速被刪除

1.5 Kafka主要配置參數

1.5.1 Broker Config

屬性默認值描述

broker.id	0	必填參數，broker的唯一標識
log.dirs	/tmp/kafka-logs	Kafka數據存放的目錄。可以指定多個目錄，中間用逗號分隔，當新partition被創建的時會被存放到當前存放partition最少的目錄。
port	9092	BrokerServer接受客戶端連接的端口號
zookeeper.connect	null	Zookeeper的連接串，格式為：hostname1:port1,hostname2:port2,hostname3:port3。可以填一個或多個，為了提高可靠性，建議都填上。注意，此配置允許我們指定一個zookeeper路徑來存放此kafka集群的所有數據，為了與其他應用集群區分開，建議在此配置中指定本集群存放目錄，格式為：hostname1:port1,hostname2:port2,hostname3:port3/chroot/path 。需要注意的是，消費者的參數要和此參數一致。
message.max.bytes	1000000	服務器可以接收到的最大的消息大小。注意此參數要和consumer的maximum.message.size大小一致，否則會因為生產者生產的消息太大導致消費者無法消費。
num.io.threads	8	服務器用來執行讀寫請求的IO線程數，此參數的數量至少要等于服務器上磁盤的數量。
queued.max.requests	500	I/O線程可以處理請求的隊列大小，若實際請求數超過此大小，網絡線程將停止接收新的請求。
socket.send.buffer.bytes	100 * 1024	The SO_SNDBUFF buffer the server prefers for socket connections.
socket.receive.buffer.bytes	100 * 1024	The SO_RCVBUFF buffer the server prefers for socket connections.
socket.request.max.bytes	100 * 1024 * 1024	服務器允許請求的最大值， 用來防止內存溢出，其值應該小于 Java heap size.
num.partitions	1	默認partition數量，如果topic在創建時沒有指定partition數量，默認使用此值，建議改為5
log.segment.bytes	1024 * 1024 * 1024	Segment文件的大小，超過此值將會自動新建一個segment，此值可以被topic級別的參數覆蓋。
log.roll.{ms,hours}	24 * 7 hours	新建segment文件的時間，此值可以被topic級別的參數覆蓋。
log.retention.{ms,minutes,hours}	7 days	Kafka segment log的保存周期，保存周期超過此時間日志就會被刪除。此參數可以被topic級別參數覆蓋。數據量大時，建議減小此值。
log.retention.bytes	-1	每個partition的最大容量，若數據量超過此值，partition數據將會被刪除。注意這個參數控制的是每個partition而不是topic。此參數可以被log級別參數覆蓋。
log.retention.check.interval.ms	5 minutes	刪除策略的檢查周期
auto.create.topics.enable	true	自動創建topic參數，建議此值設置為false，嚴格控制topic管理，防止生產者錯寫topic。
default.replication.factor	1	默認副本數量，建議改為2。
replica.lag.time.max.ms	10000	在此窗口時間內沒有收到follower的fetch請求，leader會將其從ISR(in-sync replicas)中移除。
replica.lag.max.messages	4000	如果replica節點落后leader節點此值大小的消息數量，leader節點就會將其從ISR中移除。
replica.socket.timeout.ms	30 * 1000	replica向leader發送請求的超時時間。
replica.socket.receive.buffer.bytes	64 * 1024	The socket receive buffer for network requests to the leader for replicating data.
replica.fetch.max.bytes	1024 * 1024	The number of byes of messages to attempt to fetch for each partition in the fetch requests the replicas send to the leader.
replica.fetch.wait.max.ms	500	The maximum amount of time to wait time for data to arrive on the leader in the fetch requests sent by the replicas to the leader.
num.replica.fetchers	1	Number of threads used to replicate messages from leaders. Increasing this value can increase the degree of I/O parallelism in the follower broker.
fetch.purgatory.purge.interval.requests	1000	The purge interval (in number of requests) of the fetch request purgatory.
zookeeper.session.timeout.ms	6000	ZooKeeper session 超時時間。如果在此時間內server沒有向zookeeper發送心跳，zookeeper就會認為此節點已掛掉。 此值太低導致節點容易被標記死亡；若太高，.會導致太遲發現節點死亡。
zookeeper.connection.timeout.ms	6000	客戶端連接zookeeper的超時時間。
zookeeper.sync.time.ms	2000	ZK follower落后 ZK leader的時間。
controlled.shutdown.enable	true	允許broker shutdown。如果啟用，broker在關閉自己之前會把它上面的所有leaders轉移到其它brokers上，建議啟用，增加集群穩定性。
auto.leader.rebalance.enable	true	If this is enabled the controller will automatically try to balance leadership for partitions among the brokers by periodically returning leadership to the “preferred” replica for each partition if it is available.
leader.imbalance.per.broker.percentage	10	The percentage of leader imbalance allowed per broker. The controller will rebalance leadership if this ratio goes above the configured value per broker.
leader.imbalance.check.interval.seconds	300	The frequency with which to check for leader imbalance.
offset.metadata.max.bytes	4096	The maximum amount of metadata to allow clients to save with their offsets.
connections.max.idle.ms	600000	Idle connections timeout: the server socket processor threads close the connections that idle more than this.
num.recovery.threads.per.data.dir	1	The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
unclean.leader.election.enable	true	Indicates whether to enable replicas not in the ISR set to be elected as leader as a last resort, even though doing so may result in data loss.
delete.topic.enable	false	啟用deletetopic參數，建議設置為true。
offsets.topic.num.partitions	50	The number of partitions for the offset commit topic. Since changing this after deployment is currently unsupported, we recommend using a higher setting for production (e.g., 100-200).
offsets.topic.retention.minutes	1440	Offsets that are older than this age will be marked for deletion. The actual purge will occur when the log cleaner compacts the offsets topic.
offsets.retention.check.interval.ms	600000	The frequency at which the offset manager checks for stale offsets.
offsets.topic.replication.factor	3	The replication factor for the offset commit topic. A higher setting (e.g., three or four) is recommended in order to ensure higher availability. If the offsets topic is created when fewer brokers than the replication factor then the offsets topic will be created with fewer replicas.
offsets.topic.segment.bytes	104857600	Segment size for the offsets topic. Since it uses a compacted topic, this should be kept relatively low in order to facilitate faster log compaction and loads.
offsets.load.buffer.size	5242880	An offset load occurs when a broker becomes the offset manager for a set of consumer groups (i.e., when it becomes a leader for an offsets topic partition). This setting corresponds to the batch size (in bytes) to use when reading from the offsets segments when loading offsets into the offset manager’s cache.
offsets.commit.required.acks	-1	The number of acknowledgements that are required before the offset commit can be accepted. This is similar to the producer’s acknowledgement setting. In general, the default should not be overridden.
offsets.commit.timeout.ms	5000	The offset commit will be delayed until this timeout or the required number of replicas have received the offset commit. This is similar to the producer request timeout.

1.5.2 Producer Config

屬性默認值描述

metadata.broker.list	null	啟動時producer查詢brokers的列表，可以是集群中所有brokers的一個子集。注意，這個參數只是用來獲取topic的元信息用，producer會從元信息中挑選合適的broker并與之建立socket連接。格式是：host2:port1,host2:port2。
request.required.acks	0	參見3.2節介紹
request.timeout.ms	10000	Broker等待ack的超時時間，若等待時間超過此值，會返回客戶端錯誤信息。
producer.type	sync	同步異步模式。async表示異步，sync表示同步。如果設置成異步模式，可以允許生產者以batch的形式push數據，這樣會極大的提高broker性能，推薦設置為異步。
serializer.class	kafka.serializer.DefaultEncoder	序列號類，.默認序列化成 byte[] 。
key.serializer.class	kafka.serializer.DefaultEncoder	Key的序列化類，默認同上。
partitioner.class	kafka.producer.DefaultPartitioner	Partition類，默認對key進行hash。
compression.codec	none	指定producer消息的壓縮格式，可選參數為： “none”, “gzip” and “snappy”。關于壓縮參見4.1節
compressed.topics	null	啟用壓縮的topic名稱。若上面參數選擇了一個壓縮格式，那么壓縮僅對本參數指定的topic有效，若本參數為空，則對所有topic有效。
message.send.max.retries	3	Producer發送失敗時重試次數。若網絡出現問題，可能會導致不斷重試。
retry.backoff.ms	100	Before each retry, the producer refreshes the metadata of relevant topics to see if a new leader has been elected. Since leader election takes a bit of time, this property specifies the amount of time that the producer waits before refreshing the metadata.
topic.metadata.refresh.interval.ms	600 * 1000	The producer generally refreshes the topic metadata from brokers when there is a failure (partition missing, leader not available…). It will also poll regularly (default: every 10min so 600000ms). If you set this to a negative value, metadata will only get refreshed on failure. If you set this to zero, the metadata will get refreshed after each message sent (not recommended). Important note: the refresh happen only AFTER the message is sent, so if the producer never sends a message the metadata is never refreshed
queue.buffering.max.ms	5000	啟用異步模式時，producer緩存消息的時間。比如我們設置成1000時，它會緩存1秒的數據再一次發送出去，這樣可以極大的增加broker吞吐量，但也會造成時效性的降低。
queue.buffering.max.messages	10000	采用異步模式時producer buffer 隊列里最大緩存的消息數量，如果超過這個數值，producer就會阻塞或者丟掉消息。
queue.enqueue.timeout.ms	-1	當達到上面參數值時producer阻塞等待的時間。如果值設置為0，buffer隊列滿時producer不會阻塞，消息直接被丟掉。若值設置為-1，producer會被阻塞，不會丟消息。
batch.num.messages	200	采用異步模式時，一個batch緩存的消息數量。達到這個數量值時producer才會發送消息。
send.buffer.bytes	100 * 1024	Socket write buffer size
client.id	“”	The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request.

1.5.3 Consumer Config

屬性默認值描述

group.id	null	Consumer的組ID，相同goup.id的consumer屬于同一個組。
zookeeper.connect	null	Consumer的zookeeper連接串，要和broker的配置一致。
consumer.id	null	如果不設置會自動生成。
socket.timeout.ms	30 * 1000	網絡請求的socket超時時間。實際超時時間由max.fetch.wait + socket.timeout.ms 確定。
socket.receive.buffer.bytes	64 * 1024	The socket receive buffer for network requests.
fetch.message.max.bytes	1024 * 1024	查詢topic-partition時允許的最大消息大小。consumer會為每個partition緩存此大小的消息到內存，因此，這個參數可以控制consumer的內存使用量。這個值應該至少比server允許的最大消息大小大，以免producer發送的消息大于consumer允許的消息。
num.consumer.fetchers	1	The number fetcher threads used to fetch data.
auto.commit.enable	true	如果此值設置為true，consumer會周期性的把當前消費的offset值保存到zookeeper。當consumer失敗重啟之后將會使用此值作為新開始消費的值。
auto.commit.interval.ms	60 * 1000	Consumer提交offset值到zookeeper的周期。
queued.max.message.chunks	2	用來被consumer消費的message chunks 數量， 每個chunk可以緩存fetch.message.max.bytes大小的數據量。
rebalance.max.retries	4	When a new consumer joins a consumer group the set of consumers attempt to “rebalance” the load to assign partitions to each consumer. If the set of consumers changes while this assignment is taking place the rebalance will fail and retry. This setting controls the maximum number of attempts before giving up.
fetch.min.bytes	1	The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request.
fetch.wait.max.ms	100	The maximum amount of time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy fetch.min.bytes.
rebalance.backoff.ms	2000	Backoff time between retries during rebalance.
refresh.leader.backoff.ms	200	Backoff time to wait before trying to determine the leader of a partition that has just lost its leader.
auto.offset.reset	largest	What to do when there is no initial offset in ZooKeeper or if an offset is out of range ;smallest : automatically reset the offset to the smallest offset; largest : automatically reset the offset to the largest offset;anything else: throw exception to the consumer
consumer.timeout.ms	-1	若在指定時間內沒有消息消費，consumer將會拋出異常。
exclude.internal.topics	true	Whether messages from internal topics (such as offsets) should be exposed to the consumer.
zookeeper.session.timeout.ms	6000	ZooKeeper session timeout. If the consumer fails to heartbeat to ZooKeeper for this period of time it is considered dead and a rebalance will occur.
zookeeper.connection.timeout.ms	6000	The max time that the client waits while establishing a connection to zookeeper.
zookeeper.sync.time.ms	2000	How far a ZK follower can be behind a ZK leader

參考文章：
https://blog.csdn.net/suifeng3051/article/details/48053965
https://www.cnblogs.com/frankdeng/p/9310684.html
https://cwiki.apache.org/confluence/display/KAFKA/Index（官網）
https://blog.csdn.net/wqc19920906/article/details/82193316/（消息中間件對比）

總結

以上是生活随笔為你收集整理的Kafka笔记：kafka原理简介以及架构的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。