Java API方式調用Kafka各種協議

　　眾所周知，Kafka自己實現了一套二進制協議(binary protocol)用於各種功能的實現，比如發送消息，獲取消息，提交位移以及創建topic等。具體協議規范參見：Kafka協議  這套協議的具體使用流程為：

1. 客戶端創建對應協議的請求
2. 客戶端發送請求給對應的broker
3. broker處理請求，並發送response給客戶端

　　雖然Kafka提供的大量的腳本工具用於各種功能的實現，但很多時候我們還是希望可以把某些功能以編程的方式嵌入到另一個系統中。這時使用Java API的方式就顯得異常地靈活了。本文我將嘗試給出Java API底層框架的一個范例，同時也會針對“創建topic”和“查看位移”這兩個主要功能給出對應的例子。 需要提前說明的是，本文給出的范例並沒有考慮Kafka集群開啟安全的情況。另外Kafka的KIP4應該一直在優化命令行工具以及各種管理操作，有興趣的讀者可以關注這個KIP。

　　本文中用到的API依賴於kafka-clients，所以如果你使用Maven構建的話，請加上：

 

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.0</version>
</dependency>

 

如果是gradle，請加上：

compile group: 'org.apache.kafka', name: 'kafka-clients', version: '0.10.2.0'

底層框架

/**
     * 發送請求主方法
     * @param host          目標broker的主機名
     * @param port          目標broker的端口
     * @param request       請求對象
     * @param apiKey        請求類型
     * @return              序列化后的response
     * @throws IOException
     */
    public ByteBuffer send(String host, int port, AbstractRequest request, ApiKeys apiKey) throws IOException {
        Socket socket = connect(host, port);
        try {
            return send(request, apiKey, socket);
        } finally {
            socket.close();
        }
    }

    /**
     * 發送序列化請求並等待response返回
     * @param socket            連向目標broker的socket
     * @param request           序列化后的請求
     * @return                  序列化后的response
     * @throws IOException
     */
    private byte[] issueRequestAndWaitForResponse(Socket socket, byte[] request) throws IOException {
        sendRequest(socket, request);
        return getResponse(socket);
    }

    /**
     * 發送序列化請求給socket
     * @param socket            連向目標broker的socket
     * @param request           序列化后的請求
     * @throws IOException
     */
    private void sendRequest(Socket socket, byte[] request) throws IOException {
        DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
        dos.writeInt(request.length);
        dos.write(request);
        dos.flush();
    }

    /**
     * 從給定socket處獲取response
     * @param socket            連向目標broker的socket
     * @return                  獲取到的序列化后的response
     * @throws IOException
     */
    private byte[] getResponse(Socket socket) throws IOException {
        DataInputStream dis = null;
        try {
            dis = new DataInputStream(socket.getInputStream());
            byte[] response = new byte[dis.readInt()];
            dis.readFully(response);
            return response;
        } finally {
            if (dis != null) {
                dis.close();
            }
        }
    }

    /**
     * 創建Socket連接
     * @param hostName          目標broker主機名
     * @param port              目標broker服務端口, 比如9092
     * @return                  創建的Socket連接
     * @throws IOException
     */
    private Socket connect(String hostName, int port) throws IOException {
        return new Socket(hostName, port);
    }

    /**
     * 向給定socket發送請求
     * @param request       請求對象
     * @param apiKey        請求類型, 即屬於哪種請求
     * @param socket        連向目標broker的socket
     * @return              序列化后的response
     * @throws IOException
     */
    private ByteBuffer send(AbstractRequest request, ApiKeys apiKey, Socket socket) throws IOException {
        RequestHeader header = new RequestHeader(apiKey.id, request.version(), "client-id", 0);
        ByteBuffer buffer = ByteBuffer.allocate(header.sizeOf() + request.sizeOf());
        header.writeTo(buffer);
        request.writeTo(buffer);
        byte[] serializedRequest = buffer.array();
        byte[] response = issueRequestAndWaitForResponse(socket, serializedRequest);
        ByteBuffer responseBuffer = ByteBuffer.wrap(response);
        ResponseHeader.parse(responseBuffer);
        return responseBuffer;
    }

　　有了這些方法的鋪墊，我們就可以創建具體的請求了。

創建topic

/**
     * 創建topic
     * 由於只是樣例代碼，有些東西就硬編碼寫到程序里面了(比如主機名和端口)，各位看官自行修改即可
     * @param topicName             topic名
     * @param partitions            分區數
     * @param replicationFactor     副本數
     * @throws IOException
     */
    public void createTopics(String topicName, int partitions, short replicationFactor) throws IOException {
        Map<String, CreateTopicsRequest.TopicDetails> topics = new HashMap<>();
        // 插入多個元素便可同時創建多個topic
        topics.put(topicName, new CreateTopicsRequest.TopicDetails(partitions, replicationFactor));
        int creationTimeoutMs = 60000;
        CreateTopicsRequest request = new CreateTopicsRequest.Builder(topics, creationTimeoutMs).build();
        ByteBuffer response = send("localhost", 9092, request, ApiKeys.CREATE_TOPICS);
        CreateTopicsResponse.parse(response, request.version());
    }

查看位移

/**
     * 獲取某個consumer group下的某個topic分區的位移
     * @param groupID           group id
     * @param topic             topic名
     * @param parititon         分區號
     * @throws IOException
     */
    public void getOffsetForPartition(String groupID, String topic, int parititon) throws IOException {
        TopicPartition tp = new TopicPartition(topic, parititon);
        OffsetFetchRequest request = new OffsetFetchRequest.Builder(groupID, singletonList(tp))
                .setVersion((short)2).build();
        ByteBuffer response = send("localhost", 9092, request, ApiKeys.OFFSET_FETCH);
        OffsetFetchResponse resp = OffsetFetchResponse.parse(response, request.version());
        OffsetFetchResponse.PartitionData partitionData = resp.responseData().get(tp);
        System.out.println(partitionData.offset);
    }

/**
     * 獲取某個consumer group下所有topic分區的位移信息
     * @param groupID           group id
     * @return                  (topic分區 --> 分區信息)的map
     * @throws IOException
     */
    public Map<TopicPartition, OffsetFetchResponse.PartitionData> getAllOffsetsForGroup(String groupID) throws IOException {
        OffsetFetchRequest request = new OffsetFetchRequest.Builder(groupID, null).setVersion((short)2).build();
        ByteBuffer response = send("localhost", 9092, request, ApiKeys.OFFSET_FETCH);
        OffsetFetchResponse resp = OffsetFetchResponse.parse(response, request.version());
        return resp.responseData();
    }

okay, 上面就是“創建topic”和“查看位移”的樣例代碼，各位看官可以參考着這兩個例子構建其他類型的請求。