说明

通过 KafkaConsumer 类对象，可以操作消费者

创建

//初始化配置信息
Properties config = new Properties();
//定义连接的主机信息，相当于kafka脚本命令的--bootstrap-server
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,server);
//定义分组信息，相当于kafka脚本命令的-group
config.put(ConsumerConfig.GROUP_ID_CONFIG,userGroup);
//定义数据偏移量配置，配置信息有：earliest、latest、none和anything else四种配置
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,mode);
//定义自动提交时间，时间单位为ms
config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,500);
//定义是否开启自动提交
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
//定义消费者的键的反序列化的配置
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringDeserializer");
//定义消费者的值的反序列化配置
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringDeserializer");
//初始化存放消费者的队列
KafkaConsumer<String,String> consumer consumer=new KafkaConsumer<>(config);

关闭

consumer.close();

查看主题

consumer.listTopics();

订阅主题

可以订阅多个主题：

consumer.subscribe(Arrays.asList(topic));

订阅主题并指定某个分区

可以订阅多个：

TopicPartition tp = new TopicPartition(topic,partition);
consumer.assign(Arrays.asList((tp)));

订阅主题、指定某个分区、偏移量

// 订阅主题
List<TopicPartition> list = new ArrayList();
TopicPartition tp = new TopicPartition("主题名", 分区编号);
list.add(tp);
consumer.assign(list);

// 指定分区的偏移量
consumer.seek(tp,偏移量);

获取某主题、某分区的最后一个偏移量

// 封装主题、分区编号
List<TopicPartition> list = new ArrayList();
TopicPartition tp = new TopicPartition("主题名", 分区编号);
list.add(tp);
// 获取指定主题、分区的最后一个偏移量
consumer.endOffsets(list);

原文出处：/show_1IX4bYH3Jyig.html