JavaAPI方式调用Kafka各种协议的方法

  org.apache.kafka
  kafka-clients
  0.10.2.0

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
   * 由于只是样例代码，有些东西就硬编码写到程序里面了(比如主机名和端口)，各位看官自行修改即可
   * @param topicName       topic名
   * @param partitions      分区数
   * @param replicationFactor   副本数
   * @throws IOException
   */
  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());
  }

/**
   * 获取某个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 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();
  }