卡夫卡消费者多个主题

我们可以使用以下API订阅多个主题：Consumer.subscribe（Arrays.asList（topic1，topic2），ConsumerRebalanceListener
obj）

消费者拥有主题信息，我们可以通过如下创建OffsetAndmetadata对象来使用Consumer.commitAsync或Consumer.commitSync（）。

ConsumerRecords<String, String> records = consumer.poll(long value);for (TopicPartition partition : records.partitions()) {    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);    for (ConsumerRecord<String, String> record : partitionRecords) {        System.out.println(record.offset() + ": " + record.value());    }    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();    consumer.commitSync(Collections.singletonMap(partition, new OffsetAndmetadata(lastOffset + 1)));}