【Kafka】Kafka 架構深入

Kafka工作流程及文件存儲機制

Kafka 中消息是以topic進行分類的，生產者生產消息，消費者消費消息，都是面向 topic 的。

topic 是邏輯上的概念，而 partition 是物理上的概念，每個 partition 對應于一個 log 文件，該 log 文件中存儲的就是 producer 生產的數據。Producer 生產的數據會被不斷追加到該 log 文件末端，且每條數據都有自己的 offset。 消費者組中的每個消費者，都會實時記錄自己消費到了哪個 offset，以便出錯恢復時，從上次的位置繼續(xù)消費。

由于生產者生產的消息會不斷追加到 log 文件末尾，為防止 log 文件過大導致數據定位效率低下，Kafka 采取了分片和索引機制，將每個 partition 分為多個 segment。每個 segment 對應兩個文件：“.index” 文件和 “.log” 文件。這些文件位于一個文件夾下，該文件夾的命名規(guī)則為：topic名稱+分區(qū)序號。

例如，test 這個 topic 有三個分區(qū)， 則其對應的文件夾為 test-0、test-1、test-2。

index 和 log 文件以當前 segment 的第一條消息的 offset 命名

“.index” 文件存儲大量的索引信息，“.log” 文件存儲大量的數據，索引文件中的元數據指向對應數據文件中 message 的物理偏移地址

數據可靠性保證

為保證 producer 發(fā)送的數據，能可靠的發(fā)送到指定的 topic，topic 的每個 partition 收到 producer 發(fā)送的數據后， 都需要向 producer 發(fā)送 ack（acknowledgement 確認收到），如果 producer 收到 ack，就會進行下一輪的發(fā)送，否則重新發(fā)送數據。

數據一致性問題

LEO：指的是每個副本最大的 offset；
HW：指的是消費者能見到的最大的 offset，所有副本中最小的 LEO

1）follower 故障

follower 發(fā)生故障后會被臨時踢出 ISR（Leader 維護的一個和 Leader 保持同步的 Follower 集合），待該 follower 恢復后，follower 會讀取本地磁盤記錄的上次的 HW，并將 log 文件高于 HW 的部分截取掉，從 HW 開始向 leader 進行同步。等該 follower 的 LEO 大于等于該 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

2）leader 故障

leader 發(fā)生故障之后，會從 ISR 中選出一個新的 leader， 之后，為保證多個副本之間的數據一致性，其余的 follower 會先將各自的 log 文件高于 HW 的部分截掉，然后從新的 leader 同步數據。

注：這只能保證副本之間的數據一致性，并不能保證數據不丟失或者不重復。

ack 應答機制

對于某些不太重要的數據，對數據的可靠性要求不是很高，能夠容忍數據的少量丟失，所以沒必要等 ISR 中的 follower 全部接收成功。所以 Kafka 為用戶提供了三種可靠性級別，用戶根據對可靠性和延遲的要求進行權衡選擇。

當 producer 向 leader 發(fā)送數據時，可以通過 request.required.acks 參數來設置數據可靠性的級別：
●0：這意味著producer無需等待來自broker的確認而繼續(xù)發(fā)送下一批消息。這種情況下數據傳輸效率最高，但是數據可靠性確是最低的。當broker故障時有可能丟失數據。

●1（默認配置）：這意味著producer在ISR中的leader已成功收到的數據并得到確認后發(fā)送下一條message。如果在follower同步成功之前l(fā)eader故障，那么將會丟失數據。

●-1（或者是all）：producer需要等待ISR中的所有follower都確認接收到數據后才算一次發(fā)送完成，可靠性最高。但是如果在 follower 同步完成后，broker 發(fā)送ack 之前，leader 發(fā)生故障，那么會造成數據重復。

三種機制性能依次遞減，數據可靠性依次遞增。

注：在 0.11 版本以前的Kafka，對此是無能為力的，只能保證數據不丟失，再在下游消費者對數據做全局去重。在 0.11 及以后版本的 Kafka，引入了一項重大特性：冪等性。所謂的冪等性就是指 Producer 不論向 Server 發(fā)送多少次重復數據， Server 端都只會持久化一條。

Filebeat+Kafka+ELK

確保node1 上有安裝apache服務來產生日志

環(huán)境準備

node1：192.168.67.11    elasticsearch kibana
node2：192.168.67.12    elasticsearch
apache：192.168.67.10        logstash apache/nginx/mysql
Filebeat節(jié)點：filebeat/192.168.67.13      Filebeat
zk-kfk01：192.168.67.21        zookeeper、kafka
zk-kfk02：192.168.67.22        zookeeper、kafka
zk-kfk03：192.168.67.23        zookeeper、kafka

systemctl stop firewalld
systemctl enable firewalld
setenforce 0

1、部署 Zookeeper+Kafka 集群

重啟服務

systemctl restart elasticsearch.service
netstat -antp | grep 9200

cd /usr/local/src/elasticsearch-head/
npm run start &

2、部署 Filebeat

cd /etc/filebeat
#cd /usr/local/filebeat

vim filebeat.yml
filebeat.prospectors:
- type: log
 enabled: true
 paths:
  - /var/log/httpd/access_log
 tags: ["access"]

- type: log
 enabled: true
 paths:
  - /var/log/httpd/error_log
 tags: ["error"]

......
#添加輸出到 Kafka 的配置
output.kafka:
 enabled: true
 #指定 Kafka 集群配置
 hosts: ["192.168.67.21:9092","192.168.67.22:9092","192.168.67.23:9092"]
 #指定 Kafka 的 topic
 topic: "httpd"

注釋掉logstash出口，留下kafka出口；出口只能有一個

啟動 filebeat

systemctl restart filebeat.service
systemctl status filebeat.service

# ./filebeat -e -c filebeat.yml

報錯：服務起不來；查看日志；

原因：是filebeat.yml中將日志同時輸出到了kafka和logstash

解決：注釋掉logstash即可

3、部署 ELK，在 Logstash 組件所在節(jié)點上新建一個 Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafka.conf
input {
  kafka {
    #kafka集群地址
    bootstrap_servers => "192.168.67.21:9092,192.168.67.22:9092,192.168.67.23:9092"
    #拉取的kafka的指定topic
    topics => "httpd"
    #指定 type 字段
    type => "httpd_kafka"
    #解析json格式的日志數據
    codec => "json"
    #拉取最近數據，earliest為從頭開始拉取
    auto_offset_reset => "latest"
    #傳遞給elasticsearch的數據額外增加kafka的屬性數據
    decorate_events => true
  }
}

output {
 if "access" in [tags] {
  elasticsearch {
   hosts => ["192.168.67.11:9200"]
   index => "httpd_access-%{+YYYY.MM.dd}"
  }
 }

 if "error" in [tags] {
  elasticsearch {
   hosts => ["192.168.67.11:9200"]
   index => "httpd_error-%{+YYYY.MM.dd}"
  }
 }

 stdout { codec => rubydebug }
}

啟動 logstash

`logstash -f kafka.conf`

報錯：路徑重復

解決：指定一個新的路徑

`logstash -f kafka.conf --path.data=/opt`

報錯：配置文件有錯

解決：配置文件刪了重寫

注：生產黑屏操作es時查看所有的索引:

`curl -XGET"192.168.67.11:9200/_cat/indices?v"`

4、瀏覽器訪問

`http://192.168.67.11:9100`

`http://192.168.67.11:5601/`

訪問一下apache再訪問9100

瀏覽器訪問 http://192.168.67.11:5601 登錄 Kibana，單擊“Create Index Pattern”按鈕添加索引“httpd_access-*”，單擊 “create” 按鈕創(chuàng)建，單擊 “Discover” 按鈕可查看圖表信息及日志信息。

鏈接：https://blog.csdn.net/Mo_nor/article/details/137711958?spm=1001.2014.3001.5502