Kafka系列之-Kafka Protocol例項分析

阿新 • • 發佈：2019-01-21

　　本文基於A Guide To The Kafka Protocol文件，以及Spark Streaming中實現的org.apache.spark.streaming.kafka.KafkaCluster類。整理出Kafka中有關

Metadata API
Produce API
Fetch API
Offset API(Aka ListOffset)
Offset Commit/Fetch API
Group Membership API
Administrative API

零、準備工作

　　需要執行以下部分的示例程式碼時，需要提前建好需要的topic，寫入一些message

，再用consumer消費一下。

1、新建topic

[[email protected] kafka]$ bin/kafka-topics.sh --zookeeper kafka001:2181 --create --topic kafka_protocol_test --replication-factor 3 --partitions 4
Created topic "kafka_protocol_test".
[[email protected] kafka]$ bin/kafka-topics.sh --zookeeper kafka001:2181 --describe 
 --topic kafka_protocol_test
Topic:kafka_protocol_test   PartitionCount:4    ReplicationFactor:3 Configs:
    Topic: kafka_protocol_test  Partition: 0    Leader: 1   Replicas: 1,2,3 Isr: 1,2,3
    Topic: kafka_protocol_test  Partition: 1    Leader: 2   Replicas: 2,3,4 Isr: 2,3,4
    Topic: kafka_protocol_test 
  Partition: 2    Leader: 3   Replicas: 3,4,1 Isr: 3,4,1
    Topic: kafka_protocol_test  Partition: 3    Leader: 4   Replicas: 4,1,2 Isr: 4,1,2

2、produce message

　　使用Kafka系列之-自定義Producer中提到的KafkaProducerTool往指定kafka_protocol_test中傳送訊息，

public class ProducerTest2 {
    public static void main(String[] args) throws InterruptedException {
        KafkaProducerTool kafkaProducerTool = new KafkaProducerToolImpl("D:\\Files\\test\\kafkaconfig.properties");
        int i = 1;
        while(true) {
            kafkaProducerTool.publishMessage("message" + (i++));
        }
    }
}

　　執行一段時間後停止寫入。執行一個console-consumer從kafka_protocol_test消費message。

3、consume message

　　執行一個console-consumer從kafka_protocol_test消費。注意觀察該topic每個partition中的messages數。

[[email protected] kafka]$ bin/kafka-console-consumer.sh --zookeeper kafka001:2181 --topic kafka_protocol_test --from-beginning

　　這裡寫圖片描述

一、Metadata API

　　這個API是通過向Kafka叢集傳送一個TopicMetadaaRequest請求，得到MetadataResponse響應後從MetadataResponse中解析出Metadata相關資訊。
　　TopicMetadataRequest的結構和示例如下
　　

case class TopicMetadataRequest(val versionId: Short,
                                val correlationId: Int,
                                val clientId: String,
                                val topics: Seq[String])

TopicMetadataRequest(TopicMetadataRequest.CurrentVersion, 0, config.clientId, topics)

　　得到的MetadataResponse包含的資訊如下，可以從PartitionMetadata中獲取到Partition相關資訊，從TopicMetadata中獲取到Topic相關資訊，Broker中記錄了Broker的ip和埠號等。

MetadataResponse => [Broker][TopicMetadata]
  Broker => NodeId Host Port  (any number of brokers may be returned)
    NodeId => int32
    Host => string
    Port => int32
  TopicMetadata => TopicErrorCode TopicName [PartitionMetadata]
    TopicErrorCode => int16
  PartitionMetadata => PartitionErrorCode PartitionId Leader Replicas Isr
    PartitionErrorCode => int16
    PartitionId => int32
    Leader => int32
    Replicas => [int32]
    Isr => [int32]

1、所包含的資訊

可以查詢指定Topic是否存在，
指定topic有多少個partition，
每個partition當前哪個broker處於leader狀態，
每個broker的host和port是什麼

　　如果設定了auto.create.topics.enable引數，遇到不存在的topic時，就會按預設replication和partition新建該不存在的topic。
　　

2、示例

　　生成一個TopicMetadataRequest物件

// 封裝一個TopicMetadataRequest型別的請求物件
val req = TopicMetadataRequest(TopicMetadataRequest.CurrentVersion, 0, config.clientId, topics)
// 傳送該請求
val resp: TopicMetadataResponse = consumer.send(req)
// 其中consumer物件是SimpleConsumer型別的
new SimpleConsumer(host, port, config.socketTimeoutMs,
      config.socketReceiveBufferBytes, config.clientId)

（1）查詢topic是否存在
　　由於在TopicMetadataRequest中可以傳送一組Seq[String]型別的topics，所以獲取到的TopicMetadataResponse.topicsMetadata是Set[TopicMetadata]型別的。
　　對每個TopicMetadata物件，如果其errorCode不為ErrorMapping.NoError即表示該Topic不正常。
　　

topicMetadatas.foreach { topic =>
  if (topic.errorCode == ErrorMapping.NoError)
    println(s"topic: ${topic.topic}存在")
  else
    println(s"topic: ${topic.topic}不存在")
}

（2）獲取Topic的Partition個數
　　首先將所有TopicMetadata中正常的Topic過濾出來，然後遍歷每一個TopicMetadata物件，獲取其partitionsMetadata資訊，其長度即Partition的個數

val existsTopicMetadatas = topicMetadatas.filter(tm => tm.errorCode == ErrorMapping.NoError)
existsTopicMetadatas.foreach { topic =>
   val numPartitions = topic.partitionsMetadata.length
   println(s"topic: ${topic.topic} 有${numPartitions}個partition")
}

（3）獲取Partition具體情況
　　以下程式碼可以獲取到Topic的每個Partition中的Leader Partition以及replication節點的資訊。

existsTopicMetadatas.foreach { topic =>
    println(s"topic:${topic.topic}的Partition資訊：")
    topic.partitionsMetadata.foreach { pm =>
    val leaderPartition = pm.leader
    println(s"\tpartition: ${pm.partitionId}")
    println(s"\tleader節點：$leaderPartition")
    val replicas = pm.replicas
    println(s"\treplicas節點：$replicas")
  }
}

3、執行結果

　　傳入上面新建的kafka_protocol_test以及一個不存在的topic kafka_protocol_test1，以上程式碼的執行結果如下：

=============Topic相關資訊===========
topic: kafka_protocol_test存在
topic: kafka_protocol_test1不存在
topic: kafka_protocol_test 有4個partition
=============Partition相關資訊===========
topic:kafka_protocol_test的Partition資訊：
    partition: 0
    leader節點：Some(id:1,host:kafka001,port:9092)
    replicas節點：Vector(id:1,host:kafka001,port:9092, id:2,host:kafka002,port:9092, id:3,host:kafka003,port:9092)
    partition: 1
    leader節點：Some(id:2,host:kafka002,port:9092)
    replicas節點：Vector(id:2,host:kafka002,port:9092, id:3,host:kafka003,port:9092, id:4,host:kafka004,port:9092)
    partition: 2
    leader節點：Some(id:3,host:kafka003,port:9092)
    replicas節點：Vector(id:3,host:kafka003,port:9092, id:4,host:kafka004,port:9092, id:1,host:kafka001,port:9092)
    partition: 3
    leader節點：Some(id:4,host:kafka004,port:9092)
    replicas節點：Vector(id:4,host:kafka004,port:9092, id:1,host:kafka001,port:9092, id:2,host:kafka002,port:9092)

二、Produce API

三、Fetch API

四、Offset API(Aka ListOffset)

　　這個API通過向Kafka叢集傳送一個OffsetRequest物件，從返回的OffsetResponse物件中獲取Offset相關資訊。
　　OffsetRequest物件描述如下

OffsetRequest => ReplicaId [TopicName [Partition Time MaxNumberOfOffsets]]
  ReplicaId => int32
  TopicName => string
  Partition => int32
  Time => int64
  MaxNumberOfOffsets => int32

　　上面Time的作用是，獲取特定時間(單位為ms)之前的所有messages。如果設定為-1則獲取最新的offset，即下一條messages的offset位置；如果設定為-2則獲取第一條message的offset位置，即當前partition中的offset起始位置。

　　OffsetResponse物件描述如下

OffsetResponse => [TopicName [PartitionOffsets]]
  PartitionOffsets => Partition ErrorCode [Offset]
  Partition => int32
  ErrorCode => int16
  Offset => int64

1、所包含的資訊

　　通過該API可以獲取指定topic-partition集合的合法offset的範圍，需要直接連線到Partition的Leader節點。

2、示例

　　獲取指定topic下所有partition的offset範圍
　　封裝一個getLeaderOffsets方法，在此方法的基礎上分別封裝一個getEarliestLeaderOffsets方法用於獲取最小offset，getLatestLeaderOffsets用於獲取最大offset。
　　分別傳入的關鍵引數是前面提到的Time，

def getLatestLeaderOffsets(
       topicAndPartitions: Set[TopicAndPartition]
       ): Either[Err, Map[TopicAndPartition, LeaderOffset]] =
    getLeaderOffsets(topicAndPartitions, OffsetRequest.LatestTime) // -1L
def getEarliestLeaderOffsets(
       topicAndPartitions: Set[TopicAndPartition]
       ): Either[Err, Map[TopicAndPartition, LeaderOffset]] =
    getLeaderOffsets(topicAndPartitions, OffsetRequest.EarliestTime) // -2L

　　在getLeaderOffsets中，查詢到當前partition的leader節點，
  def findLeaders(topicAndPartitions: Set[TopicAndPartition]): Either[Err, Map[TopicAndPartition, (String, Int)]] = {
    // 獲取當前topicAndPartitions中的所有topic
    val topics = topicAndPartitions.map(_.topic)
    // 獲取topic對應的MetadataResp物件，之前已過濾不存在的topic，所以這裡無需進一步過濾
    val topicMetadatas = getMetadataResp(topics.toSeq).left.get

    val leaderMap = topicMetadatas.flatMap { topic =>
      topic.partitionsMetadata.flatMap { pm =>
        val tp = TopicAndPartition(topic.topic, pm.partitionId)
        // 獲取對應PartitionMedatada的leader節點資訊
        pm.leader.map { l =>
          tp -> (l.host -> l.port)
        }
      }
    }.toMap
    Right(leaderMap)
  }

　　然後在這些節點中，封裝一個OffsetRequest物件，向Kafka叢集獲得OffsetResponse物件。

val resp = consumer.getOffsetsBefore(req)
val respMap = resp.partitionErrorAndOffsets

　　最後從OffsetResponse物件中獲取offset範圍，

val resp = getMetadataResp(topics.toSeq)
    // 如果獲取的resp是left，則處理返回的Set[TopicMetadata]
val topicAndPartitions = processRespInfo(resp) { resp =>
val topicMetadatas = resp.left.get.asInstanceOf[Set[TopicMetadata]]
val existsTopicMetadatas = topicMetadatas.filter(tm => tm.errorCode == ErrorMapping.NoError)
   getPartitions(existsTopicMetadatas)
}.asInstanceOf[Set[TopicAndPartition]]

// 獲取指定topic-partition最早的offset
val offsetBegin = getEarliestLeaderOffsets(topicAndPartitions).right.get
// 獲取指定topic-partition最晚的offset
val offsetEnd = getLatestLeaderOffsets(topicAndPartitions).right.get

print("=============Offset範圍資訊===========")
topicAndPartitions.foreach { tp =>
   println(s"topic: ${tp.topic}, Partition: ${tp.partition} 的Offset範圍：")
   println(s"\t${offsetBegin(tp).offset} ~ ${offsetEnd(tp).offset}")
}

3、執行結果

　　連線到kafka_protocol_test，執行結果如下

topic: kafka_protocol_test, Partition: 0 的Offset範圍：
    0 ~ 9000
topic: kafka_protocol_test, Partition: 1 的Offset範圍：
    0 ~ 598134
topic: kafka_protocol_test, Partition: 2 的Offset範圍：
    0 ~ 0
topic: kafka_protocol_test, Partition: 3 的Offset範圍：
    0 ~ 91000

　　和第零節中圖片顯示結果一致。

五、Offset Commit/Fetch API

　　首先參考Offset Management文件中的描述，分析一下Kafka中有關Offset管理的文件。
　　在這篇文件中主要提供了OffsetFetch和OffsetCommit兩個API，其中
　　

1、OffsetFetch API

　　這個API可以獲取一個Consumer讀取message的offset資訊。傳送的請求是OffsetFetchRequest型別的物件，接收到的是OffsetFetchResponse型別的響應。具體offset資訊可以從OffsetFetchResponse物件中解析。
　　傳送的Request請求為，需要指定consumer所屬的group，以及需要獲取offset的所有TopicAndPartitions。

val req = OffsetFetchRequest(groupId, topicAndPartitions.toSeq, 0)

　　或得到的響應為OffsetFetchResponse型別的物件。
val resp = consumer.fetchOffsets(req)
　　其中consumer物件是SimpleConsumer型別的
new SimpleConsumer(host, port, config.socketTimeoutMs,
      config.socketReceiveBufferBytes, config.clientId)

　　具體獲取offset的邏輯如下，
withBrokers(Random.shuffle(config.seedBrokers)) { consumer =>
  // 連線consumer，傳送該OffsetFetchRequest請求
  val resp = consumer.fetchOffsets(req)
  val respMap = resp.requestInfo
  // 從傳入的topicAndPartitions中取出不包含在result中的topicAndPartition
  val needed = topicAndPartitions.diff(result.keySet)
  // 遍歷每一個需要獲取offset的topic-partition
  needed.foreach { tp: TopicAndPartition =>
    respMap.get(tp).foreach { ome: OffsetMetadataAndError =>
      // 如果沒有錯誤
      if (ome.error == ErrorMapping.NoError) {
        result += tp -> ome
      } else {
        errs.append(ErrorMapping.exceptionFor(ome.error))
      }
    }
  }
  if (result.keys.size == topicAndPartitions.size) {
    return Right(result)
  }
}

2、OffsetCommit API

　　當最終呼叫commit()方法，或者如果啟用了autocommit引數時，這個API可以使consumer儲存其消費的offset資訊。
　　傳送的Request請求為OffsetCommitRequest型別。

　　OffsetCommitRequest需要傳入的引數如下，

val offsetEnd = getLatestLeaderOffsets(topicAndPartitions).right.get
val resetOffsets = offsetsFetch.right.get.map { offsetInfo =>
val plus10Offset = offsetInfo._2.offset + 10
   offsetInfo._1 -> OffsetAndMetadata(if (offsetEnd(offsetInfo._1).offset >= plus10Offset) plus10Offset else offsetEnd(offsetInfo._1).offset)
    }
// resetOffsets型別為Map[TopicAndPartition, OffsetAndMetadata]
val req = OffsetCommitRequest(groupId, resetOffsets, 0)
// 傳送該請求的方式如下
val resp = consumer.commitOffsets(req)

3、GroupCoordinator API

　　需要注意的是這個API在Kafka-0.9以後的版本中才提供。指定Consumer Group的offsets資料儲存在某個特定的Broker中。
　　向Kafka叢集傳送一個GroupCoordinatorRequest型別的請求引數，該request物件中只需要指定一個groupId即可。如下所示，

val req = new GroupCoordinatorRequest(groupId)
val resp = consumer.send(req)

　　獲取到的Response物件是GroupCoordinatorResponse型別的，在resp.coordinatorOpt中返回一個BrokerEndpoint物件，可以獲取該Broker對應的Id, Ip, Port等資訊。

4、示例

（1）執行OffsetFetch API
(a) 獲取kafka_protocol_test的consumer group消費狀態
　　啟動一個console-consumer從kafka_protocol_test topic消費messages。需要指定一個特定的group.id引數，如下所示，使用預設的consumer.properties配置檔案即可。

bin/kafka-console-consumer.sh --zookeeper kafka001:2181 --topic kafka_protocol_test --from-beginning --consumer.config ./config/consumer.properties

　　執行後，將其停止，檢視當前console-consumer的消費狀態

[[email protected] kafka]$  bin/kafka-consumer-offset-checker.sh --zookeeper kafka001:2181 --topic kafka_protocol_test --group test-consumer-group
Group           Topic                          Pid Offset          logSize         Lag             Owner
test-consumer-group kafka_protocol_test            0   9000            9000            0               none
test-consumer-group kafka_protocol_test            1   26886           598134          571248          none
test-consumer-group kafka_protocol_test            2   0               0               0               none
test-consumer-group kafka_protocol_test            3   18296           91000           72704           none

(b) 執行OffsetFetch程式碼，檢視執行結果

　　執行時仍然傳入test-consumer-group，執行結果如下

Topic: kafka_protocol_test, Partition: 0
    Offset: 9000
Topic: kafka_protocol_test, Partition: 1
    Offset: 26886
Topic: kafka_protocol_test, Partition: 2
    Offset: 0
Topic: kafka_protocol_test, Partition: 3
    Offset: 18296

　　對比後發現，兩個offset資訊保持一致。

（2）執行OffsetCommit API
　　在這裡，將OffsetFetch獲取到的每個TopicAndPartition對應的Offset加10，如果加10後超過其最大Offset，則取最大Offset。
　　在Commit前後，兩次呼叫OffsetFetch API的程式碼，前後執行結果如下，
更新前的offset：

Topic: kafka_protocol_test, Partition: 0
    Offset: 9000
Topic: kafka_protocol_test, Partition: 1
    Offset: 26886
Topic: kafka_protocol_test, Partition: 2
    Offset: 0
Topic: kafka_protocol_test, Partition: 3
    Offset: 18296
更新後的offset：（partition 0和partition 2沒有變化是由於加10後超過了該partition的offset範圍最大值）
Topic: kafka_protocol_test, Partition: 0
    Offset: 9000
Topic: kafka_protocol_test, Partition: 1
    Offset: 26896
Topic: kafka_protocol_test, Partition: 2
    Offset: 0
Topic: kafka_protocol_test, Partition: 3
    Offset: 18306

（3）執行Group Coordinator API
　　傳入一個consumer group後，檢視其執行結果

Comsuner Group : test-consumer-group, coordinator broker is:
    id: 1, host: kafka001, port: 9092

六、Group Membership API

　　這個API從Kafka-0.9.0.0版本開始出現。
　　在0.9以前的client api中，consumer是要依賴Zookeeper的。因為同一個consumer group中的所有consumer需要進行協同，進行下面所講的rebalance。但是因為zookeeper的“herd”與“split brain”，導致一個group裡面，不同的consumer擁有了同一個partition，進而會引起訊息的消費錯亂。為此，在0.9中，不再用zookeeper，而是Kafka叢集本身來進行consumer之間的同步。下面引自kafka設計的原文：
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Failuredetectionprotocol

七、Administrative API

　　注意，這個API也是從Kafka-0.9之後的client版本中才提供。通過這個API可以對Kafka叢集進行一些管理方面的操作，比如獲取所有的Consumer Groups資訊。想要獲取叢集中所有Consumer Groups資訊，需要傳送一個ListGroupRequest請求到所有的Brokers節點。
　　還可以通過傳送一個DescribeGroupsRequest型別的請求物件，獲取對特定Consumer Group的描述。

　　在Kafka-0.9之後的client中，提供了一個kafka.admin.AdminClient類，呼叫createSimplePlaintext方法，傳入一個broker list字val client = AdminClient.createSimplePlaintext(“kafka001:9092,kafka002:9092,kafka003:9092,kafka004:9092”)AdminClient`提供了很多方法，比如

def findCoordinator(groupId: String): Node
def findAllBrokers(): List[Node]
def listAllGroups(): Map[Node, List[GroupOverview]]
def listAllConsumerGroups(): Map[Node, List[GroupOverview]]

　　等等。

Kafka系列之-Kafka Protocol例項分析

零、準備工作

1、新建topic

2、produce message

3、consume message

一、Metadata API

1、所包含的資訊

2、示例

3、執行結果

二、Produce API

三、Fetch API

四、Offset API(Aka ListOffset)

1、所包含的資訊

2、示例

3、執行結果

五、Offset Commit/Fetch API

1、OffsetFetch API

2、OffsetCommit API

3、GroupCoordinator API

4、示例

六、Group Membership API

七、Administrative API

相關推薦