需要掌握：kafka教程-Offset偏移量（位移）和自动提交、手动同步提交、手动异步提交
介绍
自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握offset 提交的时机。
因此 Kafka 还提供了 手动提交 offset 的 API。
手动提交 offset 的方法有两种：
- commitSync（同步提交）
- commitAsync（异步提交）（吞吐量较大，推荐）
- ack方式。spring kafka对 commitSync、commitAsync 进行封装，默认用 commitSync 同步提交
相同点是：都会将本次 poll 的一批数据最高的偏移量提交；
不同点是：
- commitSync 阻塞当前线程，一直到提交成功，吞吐量会受到很大的影响。失败会自动重试（可能会提交失败）
- commitAsync 没有失败重试机制，故有可能提交失败。不会阻塞当前线程，吞吐量较大
注意：依赖2.4.7以上（含）可能报错
依赖2.4.7以上（含）可能报错，所以此处 尽量用 2.4.6 以下
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.4.1.RELEASE</version>
</dependency>
修改application.properties
# 不自动提交
spring.kafka.consumer.enable-auto-commit=false
# ack`manual` 或 `manual_immediate`
spring.kafka.listener.ack-mode=manual
kafka教程-springboot消费者-spring.kafka.listener.ack-mode
完整配置
server.port=8083
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=user-group
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=false
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
spring.kafka.listener.ack-mode=manual
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=earliest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
#spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
#spring.kafka.consumer.max-poll-records=50
JAVA
ack方式
spring kafka提供的方式，使用简单，默认使用 commitSync 同步提交
@KafkaListener(topics = "userlog")
public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", record.topic(),record.partition(),record.offset(),record.value()));
ack.acknowledge();
}
commitAsync 异步方式
@KafkaListener(topics = "userlog")
public void onMessage(ConsumerRecord<?, ?> record,
Consumer consumer) {
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", record.topic(),record.partition(),record.offset(),record.value()));
consumer.commitAsync();
}
commitSync同步方式
形参与上面方法一样，只是调用方法不同
@KafkaListener(topics = "userlog")
public void onMessage(ConsumerRecord<?, ?> record,
Consumer consumer) {
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", record.topic(),record.partition(),record.offset(),record.value()));
consumer.commitSync();
}