[TOC] # 1 kafka架構進階 ## 1.1 Kafka底層資料的同步機制（面試常問） 1、Kafka的Topic被分為多個分割槽，分割槽是是按照Segments（檔案段）儲存檔案塊。分割槽日誌是儲存在磁碟上的日誌序列，Kafka可以保證分割槽裡的事件是有序的。其中Leader負責對應分割槽的讀寫、Follower負責同步分割槽的資料，0.11 版本之前Kafka使用highwatermarker機制保證資料的同步，但是基於highwatermarker的同步資料可能會導致資料的不一致或者是亂序。在Kafka資料同步有以下概念。 2、LEO：log end offset 標識的是每個分割槽中最後一條訊息的下一個位置，分割槽的每個副本都有自己的LEO. 3、HW: high watermarker稱為高水位線，所有HW之前的的資料都理解是已經備份的,當所有節點都備 份成功，Leader會更新水位線。 4、ISR:In-sync-replicas,kafka的leader會維護一份處於同步的副本集，如果在`replica.lag.time.max.ms`時間內系統沒有傳送fetch請求，或者已然在傳送請求，但是在該限定時間內沒有趕上Leader的資料就被剔除ISR列表。在Kafka-0.9.0版本剔除`replica.lag.max.messages`訊息個數限定，因為這個會導致其他的Broker節點頻繁的加入和退出ISR。即如果某個從機總是趕不上leader的leo,該節點在達到限制時間後，會被剔除isr列表  ### 1.1.1 高水位截斷的同步方式可能帶來資料丟失（Kafka 0.11版本前的問題） > 場景1丟資料： B為leader，A同步B的資料，同步到m1之後，A宕機重啟，A恢復後B宕機，B的m2沒人同步，B宕機後zk選舉A為leader。此時A只有m1的資料，B恢復後，同步A的資料，此時m2會丟失。  > 場景2資料不一致：開始B為leader，寫入了兩條資料m1和m2。A同步B的資料只同步到m1時，A和B同時宕機。A啟動恢復的快一點，選舉A為leader，此時有使用者往A中寫了一條m3，A的水位線變為1，當B啟動後，B為fowller。B檢查自己的水位線也是1，不會被截斷。產生高水位相同，但是資料不一致的問題  > 為了避免這個問題，建議把Kafka的版本升級到kafka-0.11+版本 ### 1.1.2 解決高水位截斷資料丟失和不一致問題（leaderEpoch） 可以看出0.11版本之前Kafka的副本備份機制的設計存在問題。依賴HW的概念實現資料同步，但是存在資料不一致問題和丟失資料問題，因此Kafka-0.11版本引入了 Leader Epoch解決這個問題，不在使用HW作為資料截斷的依據。而是已引入了Leader epoch的概念，任意一個Leader持有一個LeaderEpoch。該LeaderEpoch這是一個由Controller管理的32位數字，儲存在Zookeeper的分割槽狀態資訊中，並作為LeaderAndIsrRequest的一部分傳遞給每個新的Leader。Leader接受Producer請求資料上使用LeaderEpoch標記每個Message。然後，該LeaderEpoch編號將通過複製協議傳播，並用於替換HW標記，作為訊息截斷的參考點。 > 生產者傳送給leader訊息，都會存下來每條訊息的leader epoch。所有從機同步leader資料的時候，都會同步一份leaderrEpoch存下來。leaderEpoch中儲存了重新選取哪個機器作為leader的資訊等  改進訊息格式，以便每個訊息集都帶有一個4位元組的Leader Epoch號。在每個日誌目錄中，會建立一個新的Leader Epoch Sequence檔案，在其中儲存Leader Epoch的序列和在該Epoch中生成的訊息的Start Offset。它也快取在每個副本中，也快取在記憶體中。 follower變成Leader 當Follower成為Leader時，它首先將新的Leader Epoch和副本的LEO新增到Leader Epoch Sequence序列檔案的末尾並重新整理資料。給Leader產生的每個新訊息集都帶有新的“Leader Epoch”標記。 Leader變成Follower 如果需要需要從本地的Leader Epoch Sequence載入資料，將資料儲存在記憶體中，給相應的分割槽的Leader傳送epoch 請求，該請求包含最新的EpochID,StartOffset資訊.Leader接收到資訊以後返回該EpochID所對應的LastOffset資訊。該資訊可能是最新EpochID的StartOffset或者是當前EpochID的Log End Offset資訊. - 情形1：Fllower的Offset比Leader的小  - 情形2：使用者的Leader Epoch的資訊startOffset資訊比Leader返回的LastOffset要大，Follower回去重置自己的Leader Epoch檔案，將Offset修改為Leader的LastOffset資訊，**並且截斷自己的日誌資訊**。  Follower在提取過程中，如果關注者看到的LeaderEpoch訊息集大於其最新的LeaderEpoch，則會在其LeaderEpochSequence中新增新的LeaderEpoch和起始偏移量，並將Epoch資料檔案重新整理到磁碟。同時將Fetch的日誌資訊重新整理到本地日誌檔案。 ### 1.1.3 LeaderEpoch解決資料丟失  ### 1.1.4 LeaderEpoch解決資料不一致  ## 1.2 kafka監控之Kafka-Eagle ### 1.2.1 Kafka-Eagle安裝 > 原始碼地址：github.com/smartloli/kafka-eagle > 下載地址： download.kafka-eagle.org 這是一個監視系統，監視您的kafka群集以及可視的使用者執行緒，偏移量，所有者等。當您安裝Kafka Eagle時，使用者可以看到當前的使用者組，對於每個組，他們正在消耗的Topic以及該組在每個主題中的偏移量，滯後，日誌大小和位置。這對於瞭解使用者從訊息佇列消耗的速度以及訊息佇列增加的速度很有用。 ``` [root@CentOSB ~]# tar -zxf kafka-eagle-web-1.4.0-bin.tar.gz -C /usr/ [root@CentOSB ~]# mv /usr/kafka-eagle-web-1.4.0 /usr/kafka-eagle [root@CentOSB ~]# vi .bashrc KE_HOME=/usr/kafka-eagle JAVA_HOME=/usr/java/latest PATH=$PATH:$JAVA_HOME/bin:$KE_HOME/bin CLASSPATH=. export JAVA_HOME export PATH export CLASSPATH export KE_HOME [root@CentOSB ~]# source .bashrc [root@CentOSB ~]# cd /usr/kafka-eagle/ [root@CentOSB kafka-eagle]# vi conf/system-config.properties kafka.eagle.zk.cluster.alias=cluster1 cluster1.zk.list=CentOSA:2181,CentOSB:2181,CentOSC:2181 cluster1.kafka.eagle.offset.storage=kafka kafka.eagle.metrics.charts=true kafka.eagle.driver=com.mysql.jdbc.Driver kafka.eagle.url=jdbc:mysql://192.168.52.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull kafka.eagle.username=root kafka.eagle.password=root [root@CentOSB kafka-eagle]# chmod u+x bin/ke.sh [root@CentOSB kafka-eagle]# ./bin/ke.sh start ``` 如果需要檢測Kafka效能指標需要修改Kafka啟動檔案 ``` vi kafka-server-start.sh ... if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70" export JMX_PORT="9999" #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" fi ``` > 視覺化操作和監控kafka叢集狀態 ## 1.3 Kafka-Flume整合 > flume也是日誌採集器，類似於elk中的logstash。通過flume採集過來的資料，一方面我們可以把資料寫入hdfs,我們也可以把採集的資料寫入到mq當中，例如kafka，kafka在寫入到主流的流處理中。 元件1：r1-採集元件，測試使用netcat 元件2：k1-輸出元件 元件3：c1-快取元件，測試使用記憶體做緩衝 > flume提供了kafka整合方案，在flume中新增kafka的配置資訊即可 - jdk8+環境，準備flume的安裝包，安裝flume > 參考 flume.apache.org 官方文件 `tar -zxf apache-flume-1.9.0-bin.tar.gz -C /usr/` `cd /usr/apache-flume-1.9.0-bin` `vi conf/kafka.properties` ``` # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = CentOS a1.sources.r1.port = 44444 # Describe the sink a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = topic01 a1.sinks.k1.kafka.bootstrap.servers = CentOSA:9092,CentOSB:9092,CentOSC:9092 a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = -1 a1.sinks.k1.kafka.producer.linger.ms = 100 a1.sinks.k1.kafka.producer.compression.type = snappy # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 ``` ``` a1.sources = r1 a1.sinks = k1 a1.channels = c1 a1.sources.r1.type = avro a1.sources.r1.bind = CentOS a1.sources.r1.port = 44444 a1.channels.c1.type = memory a1.channels.c1.capacity = 10000 a1.channels.c1.transactionCapacity = 10000 a1.channels.c1.byteCapacityBufferPercentage = 20 a1.channels.c1.byteCapacity = 800000 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = topic01 a1.sinks.k1.kafka.bootstrap.servers = CentOS:9092 a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = -1 a1.sinks.k1.kafka.producer.linger.ms = 1 a1.sinks.k1.kafka.producer.compression.type = snappy a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 ``` ## 1.4 Kafka-SpringBoot整合 - 依賴引入(pom.xml) ``` ``` - 配置(application.properties) ``` # 連線 spring.kafka.bootstrap-servers=CentOSA:9092,CentOSB:9092,CentOSC:9092 # 重試次數 spring.kafka.producer.retries=5 # 開啟應答 spring.kafka.producer.acks=all # 緩衝區大小 spring.kafka.producer.batch-size=16384 spring.kafka.producer.buffer-memory=33554432 # 事務控制，開啟後KafkaTemplate傳送訊息，要處於事務控制中@Transation spring.kafka.producer.transaction-id-prefix=transaction-id- spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer # 開啟冪等 spring.kafka.producer.properties.enable.idempotence=true spring.kafka.consumer.group-id=group1 spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100 spring.kafka.consumer.properties.isolation.level=read_committed spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer ``` - 配置日誌 ``` ``` - 整合後Spring後KafkaTemplate類專門用來發送資料 ``` @Transactional @Service public class OrderService implements IOrderService { @Autowired private KafkaTemplate kafkaTemplate; @Override public void saveOrder(String id,Object message) { //傳送訊息給伺服器 } } ``` - 通過監聽實現傳送端到訊息處理的轉發 ``` @KafkaListeners(value = {@KafkaListener(topics = {"topic04"})}) @SendTo(value = {"topic05"}) public String listenner(ConsumerRecord cr) { return cr.value()+" mashibing edu"; } ``` - 開啟事務後，KafkaTemplate傳送訊息的方式1 ``` kafkaTemplate.executeInTransaction(new KafkaOperation.Operation