[TOC] # 寫入方式 producer采用推(push)模式將消息發布到broker中,每條消息都被追加(append)到分區(patition)中,屬于順序寫磁盤(順序寫磁盤效率要比隨機寫內存要高,保障kafka吞吐率) # 分區(partition) kafka集群有多個消息代理服務器(broker-server)組成,發布到kafka集群的每條消息都有一個類別,用主題(topic)來表示.通常,不同應用產生不同類型的數據,可以設置不同的主題.一個主題一般會有多個消息的訂閱者,當生產者發布消息到某個主題時,訂閱了這個主題的消費者都可以接收到生產者寫入的新消息. kafka集群為每個主題維護了分布式的分區(partition)日志文件,物理意義上可以把主題(topic)看作進行了分區的日志文件(partition log).主題的每個分區都是一個有序的,不可變的記錄序列,新的消息會不斷追加到日志中.分區的每條消息都會按照時間順序分配到一個單調遞增的順序編號,叫做偏移量(offset),這個偏移量能夠唯一地定位當前分區內的每一條消息 消息發送時都被發送到一個topic,其本質就是一個目錄,而topic是由一些partition logs(分區日志)組成  每個partition中的消息都是**有序**的,生產的消息被不斷追加到partition log上,其中每個消息都被賦予了一個唯一的**offset值** 發布到kafka主題的每條消息包括鍵值和時間戳.消息到達服務器端的指定分區后,都會分配到一個自增的偏移量.**原始的消息內容和分配的偏移量以及其他一些元數據信息最后都會存儲到分區日志文件中.**消息的鍵也可以不用設置,這種情況下消息會均衡地分布到不同的分區. 存儲消息可以指定 `key partition value` 如果指定了partition,那么key就沒有意義了,即使設置了也沒有用. 如果沒有partition,那么指定key,他會對key進行hash來決定分到哪里 ## 分區的原因 1. 方便在集群中擴展,每個partition可以通過調整以適應它所在的機器,而一個topic又可以有多個partition組成,因此整個集群就可以適應任意大小的數據了. 2. 可以提高并發,因為可以以partition為單位讀寫了 傳統消息系統在服務端保持消息的順序,如果有多個消費者消費同一個消息隊列,服務端會以消費存儲的順序依次發送給消費者.但由于消息是異步發送給消費者的,消息到達消費者的順序可能是無序的,這就意味著在并行消費時,傳統消息系統無法很好的保證消息被順序處理.雖然我們可以設置一個專用的消費者只消費一個隊列,以此來解決消息順序的問題,但是這就使得消費處理無法真正執行. kafka比傳統消息系統有更強的順序性保證,它使用主題的分區為消息處理的并行單元.kafka以分區作為最小的粒度,將每個分區分配給消費者組中不同的而且是唯一的消費者,并確保一個分區只屬于一個消費者,即這個消費者就是這個分區的唯一讀取線程.那么,只要分區的消息是有序的,消費者處理的消息順序性就能保證.每個主題有多個分區,不同的消費者處理不同的分區,所以kafka不僅保證了消息的有序性,也做到了消費者的負載均衡 ## 分區的原則 1. 指定了partition,則直接使用 2. 未指定partition但指定key,通過對key的value進行hash出一個partition 3. partition和key都未指定,使用輪詢選出一個partition ~~~ DefaultPartitioner類 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = nextValue(topic); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } } ~~~ # 副本(Replication) 同一個partition可能會有多個replication(對應server.properties配置中的default.replication.factor=N).沒有replication的情況下,一旦broker宕機,其上所有partition的數據都不可被消費,同時producer也不能再將數據存于其上的partition.引入replication之后,同一個partition可能會有多個replication,而這時需要在這些replication之間選出一個leader,producer和consumer只與這個leader交互,其它replication作為follower從leader中復制數據 # Producer寫入流程  1. producer先從zookeeper的`"/brokers/../state"`節點找到該partition的leader 2. producer將消息發送給該leader 3. leader將消息寫入本地log 4. followers從leader pull消息,寫入本地log后向leader發送ACK 5. leader收到所有ISR中的replication的ACK后,增加HW(high watermark,最后commit的offset)并向producer發送ACK