【Kafka】消费者客户端小结（java）

public class KafkaConsumerDemo {
    private static final String BROKERLIST = "10.128.123.250:9092";
    private static final String TOPIC = "topic-demo";
    private static final String GROUP_ID = "group.demo";
    private static final AtomicBoolean IS_RUNNING = new AtomicBoolean(true);
    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList(TOPIC));
        try {
            while (IS_RUNNING.get()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("topic=" + record.topic() + ", partition=" + record.partition() + ", offset=" + record.offset());
                    System.out.println("key=" + record.key() + ", value=" + record.value());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
    public static Properties initConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.client.id.demo");
        return props;
    }
}

// 订阅主题集合
public void subscribe(Collection<String> topics) {
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
// 通过正则表达式匹配主题
public void subscribe(Pattern pattern) {
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {

consumer.subscribe(Arrays.asList(topic1));
consumer.subscribe(Arrays.asList(topic2));

consumer.subscribe(Pattern.compile("topic-.*"));

    /**
     * Manually assign a list of partitions to this consumer. This interface does not allow for incremental assignment
     * and will replace the previous assignment (if there is one).
     * <p>
     * If the given list of topic partitions is empty, it is treated the same as {@link #unsubscribe()}.
     * <p>
     * Manual topic assignment through this method does not use the consumer's group management
     * functionality. As such, there will be no rebalance operation triggered when group membership or cluster and topic
     * metadata change. Note that it is not possible to use both manual partition assignment with {@link #assign(Collection)}
     * and group assignment with {@link #subscribe(Collection, ConsumerRebalanceListener)}.
     * <p>
     * If auto-commit is enabled, an async commit (based on the old assignment) will be triggered before the new
     * assignment replaces the old one.
     *
     * @param partitions The list of partitions to assign this consumer
     * @throws IllegalArgumentException If partitions is null or contains null or empty topics
     * @throws IllegalStateException If {@code subscribe()} is called previously with topics or pattern
     *                               (without a subsequent call to {@link #unsubscribe()})
     */
    @Override
    public void assign(Collection<TopicPartition> partitions)

consumer.assign(Arrays.asList(new TopicPartition("topic-demo", 0)));

public class PartitionInfo {
  	// 主题
    private final String topic;
  	// 分区编号
    private final int partition;
  	// 分区的leader副本
    private final Node leader;
  	// 分区的AR集合
    private final Node[] replicas;
  	// 分区的ISR集合
    private final Node[] inSyncReplicas;
  	// 分区的OSR集合
    private final Node[] offlineReplicas;
  
  	// ...
}

    /**
     * Unsubscribe from topics currently subscribed with {@link #subscribe(Collection)} or {@link #subscribe(Pattern)}.
     * This also clears any partitions directly assigned through {@link #assign(Collection)}.
     *
     * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. rebalance callback errors)
     */
    public void unsubscribe() {

consumer.unsubscribe();
consumer.subscribe(new ArrayList<>());
consumer.assign(new ArrayList<>());

public ConsumerRecords<K, V> poll(final Duration timeout) {

public class ConsumerRecord<K, V> {
  public static final long NO_TIMESTAMP = RecordBatch.NO_TIMESTAMP;
  public static final int NULL_SIZE = -1;
  public static final int NULL_CHECKSUM = -1;
  /**
   * 所属主题的名称
  **/
  private final String topic;
  
  /**
   * 所在分区的编号
  **/
  private final int partition;
  
  /**
   * 消息在所属分区的偏移量
  **/
  private final long offset;
  
  /**
   * 时间戳
  **/
  private final long timestamp;
  
  /**
   * 时间戳类型
   * timestampType 有两种类型：CreateTime（消息创建的时间戳） 和LogAppendTime（消息追加到日志的时间戳）
  **/
  private final TimestampType timestampType;
  
  /**
   * 消息的键经过序列化之后的大小
   * 如果key为空，则serializedKeySize值为-1
  **/
  private final int serializedKeySize;
  
  /**
   * 消息的值经过序列化之后的大小
   * 如果value为空，则serializedKeySize值为-1
  **/
  private final int serializedValueSize;
  
  /**
   * 消息的头部内容
  **/
  private final Headers headers;
  
  /**
   * 消息的键
  **/
  private final K key;
  
  /**
   * 消息的值
  **/
  private final V value;
  private final Optional<Integer> leaderEpoch;
  /**
   * CRC32的校验值
   */
  private volatile Long checksum;
  
  // ...
}

/**
 * Get just the records for the given topic
 */
public Iterable<ConsumerRecord<K, V>> records(String topic) {
  
/**
 * Get just the records for the given partition
 * 
 * @param partition The partition to get records for
 */
public List<ConsumerRecord<K, V>> records(TopicPartition partition) {

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
for (TopicPartition tp : records.partitions()) {
	for (ConsumerRecord<String, String> record : records.records(tp)) {
		System.out.println(record.partition() + ": " + record.value());
	}
}

// 修改参数 enable.auto.commit 为 false
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

/**
 * @see KafkaConsumer#commitSync(Map)
 */
void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

for (TopicPartition tp : records.partitions()) {
  List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
  for (ConsumerRecord<String, String> record : partitionRecords) {
    System.out.println(record.partition() + ": " + record.value());
  }
  lastConsumeOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
  // 按分区粒度提交位移
  consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastConsumeOffset + 1)));
}

    /**
     * @see KafkaConsumer#commitAsync()
     */
    void commitAsync();
    /**
     * @see KafkaConsumer#commitAsync(OffsetCommitCallback)
     */
    void commitAsync(OffsetCommitCallback callback);
    /**
     * @see KafkaConsumer#commitAsync(Map, OffsetCommitCallback)
     */
    void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);

try {
  while(isRunning.get()) {
    // poll records.
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
    // do some logical processing.
    // ...
    consumer.commitAsync();
  }
} finally {
  try {
    consumer.commitSync();
  } finally {
    consumer.close();
  }
}

/**
 * @see KafkaConsumer#pause(Collection)
 */
void pause(Collection<TopicPartition> partitions);
/**
 * @see KafkaConsumer#resume(Collection)
 */
void resume(Collection<TopicPartition> partitions);

/**
 * @see KafkaConsumer#paused()
 */
Set<TopicPartition> paused();

try {
  while (IS_RUNNING.get()) {
    // consumer.poll(***)
    // process the records.
    // commit offset
    consumer.commitAsync();
  }
} catch (WakeupException e) {
  // ignore ...
} catch (Exception e) {
  // resolve the exception ...
} finally {
  consumer.commitSync();
  consumer.close();
}

/**
 * @see KafkaConsumer#seek(TopicPartition, long)
 */
void seek(TopicPartition partition, long offset);

Properties props = initConfig();
KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(TOPIC));
Set<TopicPartition> assignment = new HashSet<>();
while (assignment.size() == 0) {
  consumer.poll(Duration.ofMillis(100));
  assignment = consumer.assignment();
}
for (TopicPartition tp : assignment) {
  consumer.seek(tp, 10);
}
while (IS_RUNNING.get()) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
  // consume the records...
}

/**
 * @see KafkaConsumer#seekToBeginning(Collection)
 */
void seekToBeginning(Collection<TopicPartition> partitions);
/**
 * @see KafkaConsumer#seekToEnd(Collection)
 */
void seekToEnd(Collection<TopicPartition> partitions);

/**
 * @see KafkaConsumer#offsetsForTimes(Map)
 */
Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch);
/**
 * @see KafkaConsumer#offsetsForTimes(Map, Duration)
 */
Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch, Duration timeout);

Map<TopicPartition, OffsetAndMetadata> currentOffset = new HashMap<>();
consumer.subscribe(Arrays.asList(TOPIC), new ConsumerRebalanceListener() {
  @Override
  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    consumer.commitSync(currentOffset);
    currentOffset.clear();
  }
  @Override
  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    // do noting.
  }
});
try {
  while (IS_RUNNING.get()) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000L));
    for (ConsumerRecord<String, String> record : records) {
      // process the record
      currentOffset.put(new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1));
    }
    consumer.commitAsync();
  }
} finally {
  consumer.close();
}

  public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
  public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
  public void close();

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaConsumerThread consumerThread = new KafkaConsumerThread(props, TOPIC,
                Runtime.getRuntime().availableProcessors());
        consumerThread.start();
    }
    public static Properties initConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.client.id.demo");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
        return props;
    }
    public static class KafkaConsumerThread extends Thread {
        private KafkaConsumer<String, String> kafkaConsumer;
        private ExecutorService executorService;
        public KafkaConsumerThread(Properties props, String topic, int threadNum) {
            this.kafkaConsumer = new KafkaConsumer<>(props);
            this.kafkaConsumer.subscribe(Collections.singletonList(topic));
            this.executorService = new ThreadPoolExecutor(threadNum, threadNum, 0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy());
        }
        @Override
        public void run() {
            try {
                while (true) {
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                    if (!records.isEmpty()) {
                        executorService.submit(new RecordsHandler(records));
                    }
                    // 提交位移
                    synchronized (offsets) {
                        if (!offsets.isEmpty()) {
                            kafkaConsumer.commitSync(offsets);
                            offsets.clear();
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                kafkaConsumer.close();
            }
        }
    }
    public static class RecordsHandler extends Thread {
        public final ConsumerRecords<String, String> records;
        public RecordsHandler(ConsumerRecords<String, String> records) {
            this.records = records;
        }
        @Override
        public void run() {
            for (TopicPartition tp : records.partitions()) {
                List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
                // 处理 tpRecords ...
                // 保存位移变量
                long lastConsumedOffset = tpRecords.get(tpRecords.size() - 1).offset();
                synchronized (offsets) {
                    if (!offsets.containsKey(tp)) {
                        offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1));
                    } else {
                        long position = offsets.get(tp).offset();
                        if (position < lastConsumedOffset + 1) {
                            offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1));
                        }
                    }
                }
            }
        }
    }