介绍

自动提交 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 以下

org.springframework.kafka

spring-kafka

2.4.1.RELEASE

修改application.prpertise# 不自动提交

spring.kafka.consumer.enable-auto-commit=false

# ack`manual` 或 `manual_immediate`

spring.kafka.listener.ack-mode=manual

完整配置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(id = "consumer1",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(id = "consumer1",topics = {"userlog"})

public void onMessage(@Payload String message,

@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,

@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,

@Header(KafkaHeaders.OFFSET) String offset,

Consumer consumer) {

// 消费的哪个topic、partition的消息,打印出消息内容

System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", topic,partition,offset,message));

consumer.commitAsync();

}

commitSync同步方式

形参与上面方法一样，只是调用方法不同

@KafkaListener(id = "consumer1",topics = {"userlog"})

public void onMessage(@Payload String message,

@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,

@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,

@Header(KafkaHeaders.OFFSET) String offset,

Consumer consumer) {

// 消费的哪个topic、partition的消息,打印出消息内容

System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", topic,partition,offset,message));

consumer.commitSync();

}