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构建的话，请加上：

  org.apache.kafka
  kafka-clients
  0.10.2.0

如果是gradle，请加上：

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

底层框架

  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();
    }
  }

  
  private byte[] issueRequestAndWaitForResponse(Socket socket, byte[] request) throws IOException {
    sendRequest(socket, request);
    return getResponse(socket);
  }

  
  private void sendRequest(Socket socket, byte[] request) throws IOException {
    DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
    dos.writeInt(request.length);
    dos.write(request);
    dos.flush();
  }

  
  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();
      }
    }
  }

  
  private Socket connect(String hostName, int port) throws IOException {
    return new Socket(hostName, port);
  }

  
  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

  public void createTopics(String topicName, int partitions, short replicationFactor) throws IOException {
    Map 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());
  }

查看位移

  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);
  }

  public Map 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”和“查看位移”的样例代码，各位看官可以参考着这两个例子构建其他类型的请求。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持考高分网。