1. Kafka介紹

1.1 Kafka是什么

kafka使用scala開(kāi)發(fā)，支持多語(yǔ)言客戶端（c++、java、python、go等）

Kafka最先由LinkedIn公司開(kāi)發(fā)，之后成為Apache的頂級(jí)項(xiàng)目。

Kafka是一個(gè)分布式的、分區(qū)化、可復(fù)制提交的日志服務(wù)

LinkedIn使用Kafka實(shí)現(xiàn)了公司不同應(yīng)用程序之間的松耦和，那么作為一個(gè)可擴(kuò)展、高可靠的消息系統(tǒng) 支持高Throughput的應(yīng)用

scale out：無(wú)需停機(jī)即可擴(kuò)展機(jī)器

持久化：通過(guò)將數(shù)據(jù)持久化到硬盤(pán)以及replication防止數(shù)據(jù)丟失

支持online和offline的場(chǎng)景

1.2 Kafka的特點(diǎn)

Kafka是分布式的，其所有的構(gòu)件borker(服務(wù)端集群)、producer(消息生產(chǎn))、consumer(消息消費(fèi)者)都可以是分布式的。

在消息的生產(chǎn)時(shí)可以使用一個(gè)標(biāo)識(shí)topic來(lái)區(qū)分，且可以進(jìn)行分區(qū)；每一個(gè)分區(qū)都是一個(gè)順序的、不可變的消息隊(duì)列， 并且可以持續(xù)的添加。

同時(shí)為發(fā)布和訂閱提供高吞吐量。據(jù)了解，Kafka每秒可以生產(chǎn)約25萬(wàn)消息（50 MB），每秒處理55萬(wàn)消息（110 MB）。

消息被處理的狀態(tài)是在consumer端維護(hù)，而不是由server端維護(hù)。當(dāng)失敗時(shí)能自動(dòng)平衡

1.3 常用的場(chǎng)景

監(jiān)控：主機(jī)通過(guò)Kafka發(fā)送與系統(tǒng)和應(yīng)用程序健康相關(guān)的指標(biāo)，然后這些信息會(huì)被收集和處理從而創(chuàng)建監(jiān)控儀表盤(pán)并發(fā)送警告。

消息隊(duì)列： 應(yīng)用程度使用Kafka作為傳統(tǒng)的消息系統(tǒng)實(shí)現(xiàn)標(biāo)準(zhǔn)的隊(duì)列和消息的發(fā)布—訂閱，例如搜索和內(nèi)容提要（Content Feed）。比起大多數(shù)的消息系統(tǒng)來(lái)說(shuō)，Kafka有更好的吞吐量，內(nèi)置的分區(qū)，冗余及容錯(cuò)性，這讓Kafka成為了一個(gè)很好的大規(guī)模消息處理應(yīng)用的解決方案。消息系統(tǒng) 一般吞吐量相對(duì)較低，但是需要更小的端到端延時(shí)，并嘗嘗依賴于Kafka提供的強(qiáng)大的持久性保障。在這個(gè)領(lǐng)域，Kafka足以媲美傳統(tǒng)消息系統(tǒng)，如ActiveMR或RabbitMQ

站點(diǎn)的用戶活動(dòng)追蹤: 為了更好地理解用戶行為，改善用戶體驗(yàn)，將用戶查看了哪個(gè)頁(yè)面、點(diǎn)擊了哪些內(nèi)容等信息發(fā)送到每個(gè)數(shù)據(jù)中心的Kafka集群上，并通過(guò)Hadoop進(jìn)行分析、生成日常報(bào)告。

流處理：保存收集流數(shù)據(jù)，以提供之后對(duì)接的Storm或其他流式計(jì)算框架進(jìn)行處理。很多用戶會(huì)將那些從原始topic來(lái)的數(shù)據(jù)進(jìn)行 階段性處理，匯總，擴(kuò)充或者以其他的方式轉(zhuǎn)換到新的topic下再繼續(xù)后面的處理。例如一個(gè)文章推薦的處理流程，可能是先從RSS數(shù)據(jù)源中抓取文章的內(nèi) 容，然后將其丟入一個(gè)叫做“文章”的topic中；后續(xù)操作可能是需要對(duì)這個(gè)內(nèi)容進(jìn)行清理，比如回復(fù)正常數(shù)據(jù)或者刪除重復(fù)數(shù)據(jù)，最后再將內(nèi)容匹配的結(jié)果返 還給用戶。這就在一個(gè)獨(dú)立的topic之外，產(chǎn)生了一系列的實(shí)時(shí)數(shù)據(jù)處理的流程。

日志聚合:使用Kafka代替日志聚合（log aggregation）。日志聚合一般來(lái)說(shuō)是從服務(wù)器上收集日志文件，然后放到一個(gè)集中的位置（文件服務(wù)器或HDFS）進(jìn)行處理。然而Kafka忽略掉 文件的細(xì)節(jié)，將其更清晰地抽象成一個(gè)個(gè)日志或事件的消息流。這就讓Kafka處理過(guò)程延遲更低，更容易支持多數(shù)據(jù)源和分布式數(shù)據(jù)處理。比起以日志為中心的 系統(tǒng)比如Scribe或者Flume來(lái)說(shuō)，Kafka提供同樣高效的性能和因?yàn)閺?fù)制導(dǎo)致的更高的耐用性保證，以及更低的端到端延遲

持久性日志：Kafka可以為一種外部的持久性日志的分布式系統(tǒng)提供服務(wù)。這種日志可以在節(jié)點(diǎn)間備份數(shù)據(jù)，并為故障節(jié)點(diǎn)數(shù)據(jù)回復(fù)提供一種重新同步的機(jī)制。Kafka中日志壓縮功能為這種用法提供了條件。在這種用法中，Kafka類似于Apache BookKeeper項(xiàng)目。

1.4 Kafka中包含以下基礎(chǔ)概念

1.Topic(話題)：Kafka中用于區(qū)分不同類別信息的類別名稱。由producer指定

2.Producer(生產(chǎn)者)：將消息發(fā)布到Kafka特定的Topic的對(duì)象(過(guò)程)

3.Consumers(消費(fèi)者)：訂閱并處理特定的Topic中的消息的對(duì)象(過(guò)程)

4.Broker(Kafka服務(wù)集群)：已發(fā)布的消息保存在一組服務(wù)器中，稱之為Kafka集群。集群中的每一個(gè)服務(wù)器都是一個(gè)代理(Broker). 消費(fèi)者可以訂閱一個(gè)或多個(gè)話題，并從Broker拉數(shù)據(jù)，從而消費(fèi)這些已發(fā)布的消息。

5.Partition(分區(qū))：Topic物理上的分組，一個(gè)topic可以分為多個(gè)partition，每個(gè)partition是一個(gè)有序的隊(duì)列。partition中的每條消息都會(huì)被分配一個(gè)有序的id（offset）

Message：消息，是通信的基本單位，每個(gè)producer可以向一個(gè)topic（主題）發(fā)布一些消息。

1.5 消息

消息由一個(gè)固定大小的報(bào)頭和可變長(zhǎng)度但不透明的字節(jié)陣列負(fù)載。報(bào)頭包含格式版本和CRC32效驗(yàn)和以檢測(cè)損壞或截?cái)?/p>

1.6 消息格式

    1. 4 byte CRC32 of the message
    2. 1 byte "magic" identifier to allow format changes, value is 0 or 1
    3. 1 byte "attributes" identifier to allow annotations on the message independent of the version
       bit 0 ~ 2 : Compression codec
           0 : no compression
           1 : gzip
           2 : snappy
           3 : lz4
       bit 3 : Timestamp type
           0 : create time
           1 : log append time
       bit 4 ~ 7 : reserved
    4. (可選) 8 byte timestamp only if "magic" identifier is greater than 0
    5. 4 byte key length, containing length K
    6. K byte key
    7. 4 byte payload length, containing length V
    8. V byte payload

2. Kafka深層介紹

2.1 架構(gòu)介紹

Producer：Producer即生產(chǎn)者，消息的產(chǎn)生者，是消息的?口。

kafka cluster：kafka集群，一臺(tái)或多臺(tái)服務(wù)?組成

• Broker：Broker是指部署了Kafka實(shí)例的服務(wù)?節(jié)點(diǎn)。每個(gè)服務(wù)?上有一個(gè)或多個(gè)kafka的實(shí) 例，我們姑且認(rèn)為每個(gè)broker對(duì)應(yīng)一臺(tái)服務(wù)?。每個(gè)kafka集群內(nèi)的broker都有一個(gè)不重復(fù)的 編號(hào)，如圖中的broker-0、broker-1等……
• Topic：消息的主題，可以理解為消息的分類，kafka的數(shù)據(jù)就保存在topic。在每個(gè)broker上 都可以創(chuàng)建多個(gè)topic。實(shí)際應(yīng)用中通常是一個(gè)業(yè)務(wù)線建一個(gè)topic。
• Partition：Topic的分區(qū)，每個(gè)topic可以有多個(gè)分區(qū)，分區(qū)的作用是做負(fù)載，提高kafka的吞 吐量。同一個(gè)topic在不同的分區(qū)的數(shù)據(jù)是不重復(fù)的，partition的表現(xiàn)形式就是一個(gè)一個(gè)的?件夾！
• Replication:每一個(gè)分區(qū)都有多個(gè)副本，副本的作用是做備胎。當(dāng)主分區(qū)（Leader）故障的 時(shí)候會(huì)選擇一個(gè)備胎（Follower）上位，成為L(zhǎng)eader。在kafka中默認(rèn)副本的最大數(shù)量是10 個(gè)，且副本的數(shù)量不能大于Broker的數(shù)量，follower和leader絕對(duì)是在不同的機(jī)器，同一機(jī) ?對(duì)同一個(gè)分區(qū)也只可能存放一個(gè)副本（包括自己）。

Consumer：消費(fèi)者，即消息的消費(fèi)方，是消息的出口。

• Consumer Group：我們可以將多個(gè)消費(fèi)組組成一個(gè)消費(fèi)者組，在kafka的設(shè)計(jì)中同一個(gè)分 區(qū)的數(shù)據(jù)只能被消費(fèi)者組中的某一個(gè)消費(fèi)者消費(fèi)。同一個(gè)消費(fèi)者組的消費(fèi)者可以消費(fèi)同一個(gè) topic的不同分區(qū)的數(shù)據(jù)，這也是為了提高kafka的吞吐量！

2.2 ?作流程

我們看上?的架構(gòu)圖中，producer就是生產(chǎn)者，是數(shù)據(jù)的入口。Producer在寫(xiě)入數(shù)據(jù)的時(shí)候會(huì)把數(shù)據(jù) 寫(xiě)入到leader中，不會(huì)直接將數(shù)據(jù)寫(xiě)入follower！那leader怎么找呢？寫(xiě)入的流程又是什么樣的呢？我 們看下圖：

1.?產(chǎn)者從Kafka集群獲取分區(qū)leader信息

2.?產(chǎn)者將消息發(fā)送給leader

3.leader將消息寫(xiě)入本地磁盤(pán)

4.follower從leader拉取消息數(shù)據(jù)

5.follower將消息寫(xiě)入本地磁盤(pán)后向leader發(fā)送ACK

6.leader收到所有的follower的ACK之后向生產(chǎn)者發(fā)送ACK

2.3 選擇partition的原則

那在kafka中，如果某個(gè)topic有多個(gè)partition，producer?怎么知道該將數(shù)據(jù)發(fā)往哪個(gè)partition呢？ kafka中有幾個(gè)原則：

1.partition在寫(xiě)入的時(shí)候可以指定需要寫(xiě)入的partition，如果有指定，則寫(xiě)入對(duì)應(yīng)的partition。

2.如果沒(méi)有指定partition，但是設(shè)置了數(shù)據(jù)的key，則會(huì)根據(jù)key的值hash出一個(gè)partition。

3.如果既沒(méi)指定partition，又沒(méi)有設(shè)置key，則會(huì)采用輪詢?式，即每次取一小段時(shí)間的數(shù)據(jù)寫(xiě)入某partition，下一小段的時(shí)間寫(xiě)入下一個(gè)partition

2.4 ACK應(yīng)答機(jī)制

producer在向kafka寫(xiě)入消息的時(shí)候，可以設(shè)置參數(shù)來(lái)確定是否確認(rèn)kafka接收到數(shù)據(jù)，這個(gè)參數(shù)可設(shè)置 的值為 0,1,all

• 0代表producer往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性最低但是效 率最高。
• 1代表producer往集群發(fā)送數(shù)據(jù)只要leader應(yīng)答就可以發(fā)送下一條，只確保leader發(fā)送成功。
• all代表producer往集群發(fā)送數(shù)據(jù)需要所有的follower都完成從leader的同步才會(huì)發(fā)送下一條，確保 leader發(fā)送成功和所有的副本都完成備份。安全性最?高，但是效率最低。

最后要注意的是，如果往不存在的topic寫(xiě)數(shù)據(jù)，kafka會(huì)?動(dòng)創(chuàng)建topic，partition和replication的數(shù)量 默認(rèn)配置都是1。

2.5 Topic和數(shù)據(jù)?志

topic 是同?類別的消息記錄（record）的集合。在Kafka中，?個(gè)主題通常有多個(gè)訂閱者。對(duì)于每個(gè) 主題，Kafka集群維護(hù)了?個(gè)分區(qū)數(shù)據(jù)?志?件結(jié)構(gòu)如下：

每個(gè)partition都是?個(gè)有序并且不可變的消息記錄集合。當(dāng)新的數(shù)據(jù)寫(xiě)?時(shí)，就被追加到partition的末 尾。在每個(gè)partition中，每條消息都會(huì)被分配?個(gè)順序的唯?標(biāo)識(shí)，這個(gè)標(biāo)識(shí)被稱為offset，即偏移 量。注意，Kafka只保證在同?個(gè)partition內(nèi)部消息是有序的，在不同partition之間，并不能保證消息 有序。

Kafka可以配置?個(gè)保留期限，?來(lái)標(biāo)識(shí)?志會(huì)在Kafka集群內(nèi)保留多?時(shí)間。Kafka集群會(huì)保留在保留 期限內(nèi)所有被發(fā)布的消息，不管這些消息是否被消費(fèi)過(guò)。?如保留期限設(shè)置為兩天，那么數(shù)據(jù)被發(fā)布到 Kafka集群的兩天以內(nèi)，所有的這些數(shù)據(jù)都可以被消費(fèi)。當(dāng)超過(guò)兩天，這些數(shù)據(jù)將會(huì)被清空，以便為后 續(xù)的數(shù)據(jù)騰出空間。由于Kafka會(huì)將數(shù)據(jù)進(jìn)?持久化存儲(chǔ)（即寫(xiě)?到硬盤(pán)上），所以保留的數(shù)據(jù)??可 以設(shè)置為?個(gè)?較?的值。

2.6 Partition結(jié)構(gòu)

Partition在服務(wù)器上的表現(xiàn)形式就是?個(gè)?個(gè)的?件夾，每個(gè)partition的?件夾下?會(huì)有多組segment ?件，每組segment?件?包含 .index ?件、 .log ?件、 .timeindex ?件三個(gè)?件，其中 .log ? 件就是實(shí)際存儲(chǔ)message的地?，? .index 和 .timeindex ?件為索引?件，?于檢索消息。

2.7 消費(fèi)數(shù)據(jù)

多個(gè)消費(fèi)者實(shí)例可以組成?個(gè)消費(fèi)者組，并??個(gè)標(biāo)簽來(lái)標(biāo)識(shí)這個(gè)消費(fèi)者組。?個(gè)消費(fèi)者組中的不同消 費(fèi)者實(shí)例可以運(yùn)?在不同的進(jìn)程甚?不同的服務(wù)器上。

如果所有的消費(fèi)者實(shí)例都在同?個(gè)消費(fèi)者組中，那么消息記錄會(huì)被很好的均衡的發(fā)送到每個(gè)消費(fèi)者實(shí) 例。

如果所有的消費(fèi)者實(shí)例都在不同的消費(fèi)者組，那么每?條消息記錄會(huì)被?播到每?個(gè)消費(fèi)者實(shí)例。

舉個(gè)例?，如上圖所示?個(gè)兩個(gè)節(jié)點(diǎn)的Kafka集群上擁有?個(gè)四個(gè)partition（P0-P3）的topic。有兩個(gè) 消費(fèi)者組都在消費(fèi)這個(gè)topic中的數(shù)據(jù)，消費(fèi)者組A有兩個(gè)消費(fèi)者實(shí)例，消費(fèi)者組B有四個(gè)消費(fèi)者實(shí)例。 從圖中我們可以看到，在同?個(gè)消費(fèi)者組中，每個(gè)消費(fèi)者實(shí)例可以消費(fèi)多個(gè)分區(qū)，但是每個(gè)分區(qū)最多只 能被消費(fèi)者組中的?個(gè)實(shí)例消費(fèi)。也就是說(shuō)，如果有?個(gè)4個(gè)分區(qū)的主題，那么消費(fèi)者組中最多只能有4 個(gè)消費(fèi)者實(shí)例去消費(fèi)，多出來(lái)的都不會(huì)被分配到分區(qū)。其實(shí)這也很好理解，如果允許兩個(gè)消費(fèi)者實(shí)例同 時(shí)消費(fèi)同?個(gè)分區(qū)，那么就?法記錄這個(gè)分區(qū)被這個(gè)消費(fèi)者組消費(fèi)的offset了。如果在消費(fèi)者組中動(dòng)態(tài) 的上線或下線消費(fèi)者，那么Kafka集群會(huì)?動(dòng)調(diào)整分區(qū)與消費(fèi)者實(shí)例間的對(duì)應(yīng)關(guān)系。

3. 操作Kafka

3.1 sarama

Go語(yǔ)言中連接kafka使用第三方庫(kù): github.com/Shopify/sarama。

3.2 下載及安裝

    go get github.com/Shopify/sarama

注意事項(xiàng): sarama v1.20之后的版本加入了zstd壓縮算法，需要用到cgo，在Windows平臺(tái)編譯時(shí)會(huì)提示類似如下錯(cuò)誤： github.com/DataDog/zstd exec: "gcc":executable file not found in %PATH% 所以在Windows平臺(tái)請(qǐng)使用v1.19版本的sarama。(如果不會(huì)版本控制請(qǐng)查看博客里面的go module章節(jié))

3.3 連接kafka發(fā)送消息

package main
import (
    "fmt"
    "github.com/Shopify/sarama"
)
// 基于sarama第三方庫(kù)開(kāi)發(fā)的kafka client
func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 發(fā)送完數(shù)據(jù)需要leader和follow都確認(rèn)
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 新選出一個(gè)partition
    config.Producer.Return.Successes = true                   // 成功交付的消息將在success channel返回
    // 構(gòu)造一個(gè)消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = "web_log"
    msg.Value = sarama.StringEncoder("this is a test log")
    // 連接kafka
    client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    defer client.Close()
    // 發(fā)送消息
    pid, offset, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send msg failed, err:", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

3.4 連接kafka消費(fèi)消息

package main
import (
    "fmt"
    "github.com/Shopify/sarama"
)
// kafka consumer
func main() {
    consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
    if err != nil {
        fmt.Printf("fail to start consumer, err:%v\n", err)
        return
    }
    partitionList, err := consumer.Partitions("web_log") // 根據(jù)topic取到所有的分區(qū)
    if err != nil {
        fmt.Printf("fail to get list of partition:err%v\n", err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList { // 遍歷所有的分區(qū)
        // 針對(duì)每個(gè)分區(qū)創(chuàng)建一個(gè)對(duì)應(yīng)的分區(qū)消費(fèi)者
        pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
            return
        }
        defer pc.AsyncClose()
        // 異步從每個(gè)分區(qū)消費(fèi)信息
        go func(sarama.PartitionConsumer) {
            for msg := range pc.Messages() {
                fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
            }
        }(pc)
    }
}

以上就是go操作Kfaka使用示例詳解的詳細(xì)內(nèi)容，更多關(guān)于go操作Kfaka的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論