kafka 通信報文格式

1. 序列化一條消息

消息有 key 和 value

kafka 提供了基礎數據類型的序列化工具，對於業務的自定義類需要自行實現序列化

ProducerRecord 是對象，含 KV 和 headers，此時的 KV 還是對象

在 KafkaProducer#doSend 中會對 KV 進行序列化，得到 KV 的 byte 數組

然后把 byte 數組和 headers 加入到 ProducerBatch 中

代碼見:

org.apache.kafka.clients.producer.internals.ProducerBatch#recordsBuilder
org.apache.kafka.common.record.MemoryRecordsBuilder#appendStream

 

2. kafka 的 tcp 報文

利用 Struct 和 Schema 把 ProducerBatch 的數據轉換成符合 kafka 格式的 tcp 報文

以發送消息為例

org.apache.kafka.common.requests.AbstractRequest#toSend
org.apache.kafka.common.requests.AbstractRequest#serialize
org.apache.kafka.common.requests.AbstractRequestResponse#serialize
org.apache.kafka.common.requests.ProduceRequest#toStruct
org.apache.kafka.common.protocol.types.Schema#write

org.apache.kafka.common.requests.RequestHeader#toStruct

public Struct toStruct() {
    Schema schema = schema(apiKey.id, apiVersion);
    Struct struct = new Struct(schema);
    struct.set(API_KEY_FIELD_NAME, apiKey.id);
    struct.set(API_VERSION_FIELD_NAME, apiVersion);

    // only v0 of the controlled shutdown request is missing the clientId
    if (struct.hasField(CLIENT_ID_FIELD_NAME))
        struct.set(CLIENT_ID_FIELD_NAME, clientId);
    struct.set(CORRELATION_ID_FIELD_NAME, correlationId);
    return struct;
}

org.apache.kafka.common.requests.ProduceRequest#toStruct

public Struct toStruct() {
    // Store it in a local variable to protect against concurrent updates
    Map<TopicPartition, MemoryRecords> partitionRecords = partitionRecordsOrFail();
    short version = version();
    Struct struct = new Struct(ApiKeys.PRODUCE.requestSchema(version));
    Map<String, Map<Integer, MemoryRecords>> recordsByTopic = CollectionUtils.groupDataByTopic(partitionRecords);
    struct.set(ACKS_KEY_NAME, acks);
    struct.set(TIMEOUT_KEY_NAME, timeout);
    struct.setIfExists(NULLABLE_TRANSACTIONAL_ID, transactionalId);

    List<Struct> topicDatas = new ArrayList<>(recordsByTopic.size());
    for (Map.Entry<String, Map<Integer, MemoryRecords>> topicEntry : recordsByTopic.entrySet()) {
        Struct topicData = struct.instance(TOPIC_DATA_KEY_NAME);
        topicData.set(TOPIC_NAME, topicEntry.getKey());
        List<Struct> partitionArray = new ArrayList<>();
        for (Map.Entry<Integer, MemoryRecords> partitionEntry : topicEntry.getValue().entrySet()) {
            MemoryRecords records = partitionEntry.getValue();
            Struct part = topicData.instance(PARTITION_DATA_KEY_NAME)
                    .set(PARTITION_ID, partitionEntry.getKey())
                    .set(RECORD_SET_KEY_NAME, records);
            partitionArray.add(part);
        }
        topicData.set(PARTITION_DATA_KEY_NAME, partitionArray.toArray());
        topicDatas.add(topicData);
    }
    struct.set(TOPIC_DATA_KEY_NAME, topicDatas.toArray());
    return struct;
}

組裝報文

public abstract class AbstractRequestResponse {
    /**
     * Visible for testing.
     */
    public static ByteBuffer serialize(Struct headerStruct, Struct bodyStruct) {
        ByteBuffer buffer = ByteBuffer.allocate(headerStruct.sizeOf() + bodyStruct.sizeOf());
        headerStruct.writeTo(buffer);
        bodyStruct.writeTo(buffer);
        buffer.rewind();
        return buffer;
    }
}

public class NetworkSend extends ByteBufferSend {

    public NetworkSend(String destination, ByteBuffer buffer) {
        super(destination, sizeDelimit(buffer));
    }

    private static ByteBuffer[] sizeDelimit(ByteBuffer buffer) {
        return new ByteBuffer[] {sizeBuffer(buffer.remaining()), buffer};
    }

    private static ByteBuffer sizeBuffer(int size) {
        ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
        sizeBuffer.putInt(size);
        sizeBuffer.rewind();
        return sizeBuffer;
    }

}

所以能推斷出，kafka 報文格式：4 字節存儲長度，headerStruct，bodyStruct

當然通過 NetworkSend 和 NetworkReceive 的注釋也能看出來