Apache Kafka是一个分布式流处理平台，被广泛应用于构建实时数据管道和流式应用。在Kafka中，消费者是负责从主题（Topic）读取消息的组件。随着Kafka的发展，消费者API也经历了重要的变更，理解新旧消费者之间的区别对于开发和维护Kafka应用至关重要。
最初的消费者API被称为“低级API”，后来又推出了“高级API”，而旧的消费者API则被称为“简单消费者”。2017年，Kafka引入了全新的消费者API，通常被称为“新消费者”，以替代之前的“简单消费者”和“高级消费者”API。以下是新旧消费者的主要区别：
设计理念与API结构：
旧消费者API（简单消费者和高级消费者）主要关注点在于提供灵活的消费者实现方式，允许开发者控制如何读取数据、提交偏移量以及处理消息。然而，这种灵活性同时带来了复杂性和对消息顺序保证的困难。
相反，新消费者API采用了不同的设计哲学，它基于更为简洁和高效的抽象，将消息获取和偏移量管理的细节隐藏起来，提供了更简单的接口，使得开发者可以更容易地构建健壮的消费者应用。
偏移量管理：
旧消费者要求开发者自己管理偏移量，即决定何时以及如何提交偏移量。如果处理不当，可能导致重复读取或数据丢失。
新消费者则自动管理偏移量，只需要通过调用commitSync（同步提交）或设置自动提交策略，即可轻松保证消息的确切一次处理（exactly-once processing）。
消费组支持：
高级消费者支持消费组，但配置复杂且功能受限。新消费者不仅支持消费组，还提供了更加丰富的特性，如动态的成员增加与移除、自动的再平衡等。
示例代码比较：
这是一个简单的旧消费者（高级API）代码示例：
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerConnector;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
// 初始化配置
Properties props = new Properties();
props.put("zookeeper.connect", "localhost:2181");
props.put("group.id", "test-group");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset", "smallest");
// 创建连接
ConsumerConfig config = new ConsumerConfig(props);
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);
// 订阅主题
Map<String, Integer> topicCount = new HashMap<>();
topicCount.put("my-topic", 1);
Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer.createMessageStreams(topicCount);
List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get("my-topic");
// 消费消息
ConsumerIterator<byte[], byte[]> iterator = streams.get(0).iterator();
while (iterator.hasNext()) {
MessageAndMetadata<byte[], byte[]> messageAndMetadata = (MessageAndMetadata<byte[], byte[]>) iterator.next();
// 处理消息
}
而新消费者的代码则简洁许多：
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;
public class SimpleConsumer {
public static void main(String[] args) {
// 配置设置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset", "earliest");
// 创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 订阅主题
consumer.subscribe(Collections.singletonList("my-topic"));
// 拉取并处理消息
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
consumer.commitAsync(); // 异步提交偏移量
}
}
}
通过以上示例可以看出，新消费者API的设计使得集成和使用变得更加简单明了，极大降低了开发者处理消息和偏移量时的复杂性，从而提升了开发效率和系统的可维护性。
