• 先了解Reqeust和Response的構(gòu)成, 有助于我們分析各種請求的處理過程;
• Kafka的Request基本上分為client->server和server->server兩大類;

基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)類:

Type類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
• 這是一個abstrace class, 主要是定義了ByteBuffer與各種Object之間的序列化和反序列化;

public abstract void write(ByteBuffer buffer, Object o);
public abstract Object read(ByteBuffer buffer);
public abstract Object validate(Object o);
public abstract int sizeOf(Object o);
public boolean isNullable();

• 定義了若干Type類的實現(xiàn)類:

public static final Type INT8
public static final Type INT16
public static final Type INT32
public static final Type INT64
public static final Type STRING
public static final Type BYTES
public static final Type NULLABLE_BYTES

ArrayOf類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java
• Type類的具體實現(xiàn), 是Type對象的數(shù)組類型;

Field類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/types/Field.java
• 定義了在這個schema中的一個字段;
• 成員:

    final int index;
    public final String name;
    public final Type type;
    public final Object defaultValue;
    public final String doc;
    final Schema schema;

Schema類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/types/schema.java
• Schema類本身實現(xiàn)了Type類, 又包含了一個Field類對象的數(shù)組, 構(gòu)成了記錄的Schema;

Sturct類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/types/struct.java
• 包括了一個Schema對象; 一個Object[] values數(shù)組,用于存放Schema描述的所有Field對應(yīng)的值;

    private final Schema schema;
    private final Object[] values;

• 定義了一系列getXXX方法, 用來獲取schema中某個Field對應(yīng)的值;
• 定義了set方法, 用來設(shè)置schema中某個Field對應(yīng)的值;
• writeTo 用來將Stuct對象序列華到ByteBuffer;
• Schema就是模板,Struct負(fù)責(zé)特化這個模板,向模板里添數(shù)據(jù),構(gòu)造出具體的request對象, 并可以將這個對象與ByteBuffer互相轉(zhuǎn)化;

協(xié)議相關(guān)類型:

Protocol類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
• 定義了各種Schema:

public static final Schema REQUEST_HEADER = new Schema(new Field("api_key", INT16, "The id of the request type."),
                                                           new Field("api_version", INT16, "The version of the API."),
                                                           new Field("correlation_id",
                                                                     INT32,
                                                                     "A user-supplied integer value that will be passed back with the response"),
                                                           new Field("client_id",
                                                                     STRING,
                                                                     "A user specified identifier for the client making the request."));
...

ApiKeys類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
• 定義了所有Kafka Api 的ID和名字
• 如下:

    PRODUCE(0, "Produce"),
    FETCH(1, "Fetch"),
    LIST_OFFSETS(2, "Offsets"),
    METADATA(3, "Metadata"),
    LEADER_AND_ISR(4, "LeaderAndIsr"),
    STOP_REPLICA(5, "StopReplica"),
    UPDATE_METADATA_KEY(6, "UpdateMetadata"),
    CONTROLLED_SHUTDOWN_KEY(7, "ControlledShutdown"),
    OFFSET_COMMIT(8, "OffsetCommit"),
    OFFSET_FETCH(9, "OffsetFetch"),
    GROUP_COORDINATOR(10, "GroupCoordinator"),
    JOIN_GROUP(11, "JoinGroup"),
    HEARTBEAT(12, "Heartbeat"),
    LEAVE_GROUP(13, "LeaveGroup"),
    SYNC_GROUP(14, "SyncGroup"),
    DESCRIBE_GROUPS(15, "DescribeGroups"),
    LIST_GROUPS(16, "ListGroups");

Request和Response相關(guān)類型

每個Request和Response都由RequestHeader(ResponseHeader) + 具體的消費體構(gòu)成;

AbstractRequestResponse類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/request/AbstractRequestResponse.java
• 所有Request和Response的抽象基類
• 主要數(shù)據(jù)成員: protected final Struct struct
• 主要接口:

public int sizeOf()
public void writeTo(ByteBuffer buffer)
public String toString()
public int hashCode()
public boolean equals(Object obj)

AbstractRequest類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/request/AbstractRequest.java
• 繼承自AbstractReqeustResponse類, 增加了接口:
  public abstract AbstractRequestResponse getErrorResponse(int versionId, Throwable e)
• 最重要的是它提供了一個工廠方法用于從ByteBuffer來產(chǎn)生不同類型的具體的Request;

public static AbstractRequest getRequest(int requestId, int versionId, ByteBuffer buffer) {
        switch (ApiKeys.forId(requestId)) {
            case PRODUCE:
                return ProduceRequest.parse(buffer, versionId);
            case FETCH:
                return FetchRequest.parse(buffer, versionId);
            case LIST_OFFSETS:
                return ListOffsetRequest.parse(buffer, versionId);
            case METADATA:
                return MetadataRequest.parse(buffer, versionId);
            case OFFSET_COMMIT:
                return OffsetCommitRequest.parse(buffer, versionId);
            case OFFSET_FETCH:
                return OffsetFetchRequest.parse(buffer, versionId);
            case GROUP_COORDINATOR:
                return GroupCoordinatorRequest.parse(buffer, versionId);
            case JOIN_GROUP:
                return JoinGroupRequest.parse(buffer, versionId);
            case HEARTBEAT:
                return HeartbeatRequest.parse(buffer, versionId);
            case LEAVE_GROUP:
                return LeaveGroupRequest.parse(buffer, versionId);
            case SYNC_GROUP:
                return SyncGroupRequest.parse(buffer, versionId);
            case STOP_REPLICA:
                return StopReplicaRequest.parse(buffer, versionId);
            case CONTROLLED_SHUTDOWN_KEY:
                return ControlledShutdownRequest.parse(buffer, versionId);
            case UPDATE_METADATA_KEY:
                return UpdateMetadataRequest.parse(buffer, versionId);
            case LEADER_AND_ISR:
                return LeaderAndIsrRequest.parse(buffer, versionId);
            case DESCRIBE_GROUPS:
                return DescribeGroupsRequest.parse(buffer, versionId);
            case LIST_GROUPS:
                return ListGroupsRequest.parse(buffer, versionId);
            default:
                return null;
        }
    }

實現(xiàn)上是調(diào)用各個具體Request對象的parse方法根據(jù)bytebuffer和versionid來產(chǎn)生具體的Request對象;

ProduceRequest類:

• 我們找其中一個ProduceRqeust類來分析一下, 這個類是客戶端提交消息到broker時使用的請求;
• 所在文件: clients/src/main/java/org/apache/kafka/common/request/ProduceRequest.java
• 一個ProduceRequest包括下列字段:

    private final short acks;
    private final int timeout;
    private final Map<TopicPartition, ByteBuffer> partitionRecords;

• 構(gòu)造函數(shù)public ProduceRequest(Struct struct), 利用Struct里定義的Schame來從ByteBuffer反序列化出ProduceRequest對象;

public ProduceRequest(Struct struct) {
        super(struct);
        partitionRecords = new HashMap<TopicPartition, ByteBuffer>();
        for (Object topicDataObj : struct.getArray(TOPIC_DATA_KEY_NAME)) {
            Struct topicData = (Struct) topicDataObj;
            String topic = topicData.getString(TOPIC_KEY_NAME);
            for (Object partitionResponseObj : topicData.getArray(PARTITION_DATA_KEY_NAME)) {
                Struct partitionResponse = (Struct) partitionResponseObj;
                int partition = partitionResponse.getInt(PARTITION_KEY_NAME);
                ByteBuffer records = partitionResponse.getBytes(RECORD_SET_KEY_NAME);
                partitionRecords.put(new TopicPartition(topic, partition), records);
            }
        }
        acks = struct.getShort(ACKS_KEY_NAME);
        timeout = struct.getInt(TIMEOUT_KEY_NAME);
    }

RequestHeader類:

• 所在文件: clients/src/main/java/org/apache/kafka/common/request/RequestHeader.java
• Request的消息頭
• 主要成員:

    private static final Field API_KEY_FIELD = REQUEST_HEADER.get("api_key");
    private static final Field API_VERSION_FIELD = REQUEST_HEADER.get("api_version");
    private static final Field CLIENT_ID_FIELD = REQUEST_HEADER.get("client_id");
    private static final Field CORRELATION_ID_FIELD = REQUEST_HEADER.get("correlation_id");

關(guān)系圖:

request_response.png

實際上在 core/src/main/scala/kafka/api下也定義了各種Request和Response:

• 代碼中的注釋:

NOTE: this map only includes the server-side request/response handlers. Newer
request types should only use the client-side versions which are parsed with
o.a.k.common.requests.AbstractRequest.getRequest()

     val keyToNameAndDeserializerMap: Map[Short, (String, (ByteBuffer) =>       
     RequestOrResponse)]=
     Map(ProduceKey -> ("Produce", ProducerRequest.readFrom),
        FetchKey -> ("Fetch", FetchRequest.readFrom),
        OffsetsKey -> ("Offsets", OffsetRequest.readFrom),
        MetadataKey -> ("Metadata", TopicMetadataRequest.readFrom),
        LeaderAndIsrKey -> ("LeaderAndIsr", LeaderAndIsrRequest.readFrom),
        StopReplicaKey -> ("StopReplica", StopReplicaRequest.readFrom),
        UpdateMetadataKey -> ("UpdateMetadata", UpdateMetadataRequest.readFrom),
        ControlledShutdownKey -> ("ControlledShutdown",   
        ControlledShutdownRequest.readFrom),
        OffsetCommitKey -> ("OffsetCommit", OffsetCommitRequest.readFrom),
        OffsetFetchKey -> ("OffsetFetch", OffsetFetchRequest.readFrom)

• 這部分作解析, 沒有采用schema的形式, 是采用的直接讀取方式:

def readFrom(buffer: ByteBuffer): ProducerRequest = {
    val versionId: Short = buffer.getShort
    val correlationId: Int = buffer.getInt
    val clientId: String = readShortString(buffer)
    val requiredAcks: Short = buffer.getShort
    val ackTimeoutMs: Int = buffer.getInt
    //build the topic structure
    val topicCount = buffer.getInt
    val partitionDataPairs = (1 to topicCount).flatMap(_ => {
      // process topic
      val topic = readShortString(buffer)
      val partitionCount = buffer.getInt
      (1 to partitionCount).map(_ => {
        val partition = buffer.getInt
        val messageSetSize = buffer.getInt
        val messageSetBuffer = new Array[Byte](messageSetSize)
        buffer.get(messageSetBuffer,0,messageSetSize)
        (TopicAndPartition(topic, partition), new ByteBufferMessageSet(ByteBuffer.wrap(messageSetBuffer)))
      })
    })

請求生成與保存

• 所有進(jìn)來的請求最終會轉(zhuǎn)換成 RequestChannel::Request, 保存在RequestChannel的ArrayBlockingQueue[RequestChannel.Request]中, 這個前面章節(jié)已經(jīng)講過;

Kafka協(xié)議官網(wǎng)地址

下一篇Kafka初始化流程與請求處理

Kafka源碼分析-匯總