kafka 消费模式 AckMode 以及手动消费
自动消费
Spring-Boot-kafka 的配置中有一个参数，提供了自动消费的功能。使用自动消费我们在收到消息的时候会自动向 kafka 确认消费，是否消费成功，并不能保证。在一些特殊场景中我们可能需要使用一些其他的消费模式。
enable-auto-commit: true
kafka 的消费模式
需要自己配置 AckMode 时候的配置
spring:
application:
name: base.kafka
kafka:
bootstrap-servers: kafka服务地址1:端口,kafka服务地址2:端口,kafka服务地址3:端口
producer:
# 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
# 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
retries: 0
#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
#可以设置的值为：all, -1, 0, 1
acks: 1
consumer:
group-id: testGroup
# smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
auto-offset-reset: earliest
# 设置自动提交offset
enable-auto-commit: false
max-poll-records: 2
server:
port: 8060
kafka 支持的消费模式，设置在 AbstractMessageListenerContainer.AckMode 的枚举中，
更高版本的在 ContainerProperties.AckMode 的枚举中
下面就介绍下各个模式的区别
/**
* The offset commit behavior enumeration.
*/
public enum AckMode {
/**
* Commit after each record is processed by the listener.
*/
RECORD,
/**
* Commit whatever has already been processed before the next poll.
*/
BATCH,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
*/
TIME,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded.
*/
COUNT,
/**
* Commit pending updates after
* {@link ContainerProperties#setAckCount(int) ackCount} has been
* exceeded or after {@link ContainerProperties#setAckTime(long)
* ackTime} has elapsed.
*/
COUNT_TIME,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}.
*/
MANUAL,
/**
* User takes responsibility for acks using an
* {@link AcknowledgingMessageListener}. The consumer
* immediately processes the commit.
*/
MANUAL_IMMEDIATE,
}
AckMode 模式 作用
MANUAL 当每一批 poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用 Acknowledgment.acknowledge() 后提交
MANUAL_IMMEDIATE 手动调用 Acknowledgment.acknowledge() 后立即提交
RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
BATCH 当每一批 poll() 的数据被消费者监听器（ListenerConsumer）处理之后提交
TIME 当每一批 poll() 的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于 TIME 时提交
COUNT 当每一批 poll() 的数据被消费者监听器（ListenerConsumer）处理之后，被处理 record 数量大于等于 COUNT 时提交
COUNT_TIME TIME 或 COUNT 有一个条件满足时提交
MANUAL 和 MANUAL_IMMEDIATE
MANUAL
当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
监听器工厂的配置
/**
* MANUAL 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
* @param consumerFactory
* @return
*/
@Bean("manualListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setPollTimeout(1500);
factory.setBatchListener(true);
//配置手动提交offset
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
return factory;
}
指定监听器使用的配置
/**
* MANUAL 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
* @param message
* @param ack
*/
@KafkaListener(containerFactory = "manualListenerContainerFactory" , topics = "kafka-manual")
public void onMessageManual(List<Object> message, Acknowledgment ack){
log.info("manualListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("manualListenerContainerFactory 处理数据内容：{}",item));
ack.acknowledge();//直接提交offset
}
MANUAL_IMMEDIATE
当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
监听器工厂的配置
/**
* MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
* @param consumerFactory
* @return
*/
@Bean("manualImmediateListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualImmediateListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setPollTimeout(1500);
factory.setBatchListener(true);
//配置手动提交offset
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
指定监听器使用的配置
/**
* MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
* @param message
*/
@KafkaListener(containerFactory = "manualImmediateListenerContainerFactory" , topics = "kafka-manualImmediate")
public void onMessageManualImmediate(List<Object> message, Acknowledgment ack){
log.info("manualImmediateListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("manualImmediateListenerContainerFactory 处理数据内容：{}",item));
ack.acknowledge();//直接提交offset
}
MANUAL 和 MANUAL_IMMEDIATE 两者的相同和区别
相同之处
这两种模式都是需要进行手动确认 ack.acknowledge(); 才能完成消息的消费，否则在重启消费端实例的时候数据会再次被消费端接收到。
两者的区别
MANUAL: 在处理完最后一次轮询的所有结果后，将队列排队，并在一次操作中提交偏移量。可以认为是在批处理结束时提交偏移量
MANUAL_IMMEDIATE：只要在侦听器线程上执行确认，就立即提交偏移。会在批量执行的时候逐一提交它们。
关于两者的解释可以看看这个提问 what-is-the-difference-between-manual-and-manual-immediate-in-spring-kafka-ackmo
RECORD
监听器工厂的配置
/**
* RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
* @param consumerFactory
* @return
*/
@Bean("recordListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> recordListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setPollTimeout(1500);
//配置手动提交offset
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
return factory;
}
指定监听器使用的配置
/**
* RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
* @param message
*/
@KafkaListener(containerFactory = "recordListenerContainerFactory" , topics = "kafka-record")
public void onMessageRecord(List<Object> message){
log.info("recordListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("recordListenerContainerFactory 处理数据内容：{}",item));
}
关于 AckMode.RECORD 模式
使用 RECORD 模式的时候，当监听器处理完消息后会自动处理，使用此模式不需要手动消费。
TIME
监听器工厂的配置
/**
* TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
* @param consumerFactory
* @return
*/
@Bean("timeListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> timeListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setPollTimeout(2000);
//配置手动提交offset
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.TIME);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.TIME);
return factory;
}
指定监听器使用的配置
/**
* TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
* @param message
*/
@KafkaListener(containerFactory = "timeListenerContainerFactory" , topics = "kafka-time")
public void onMessageTime(List<Object> message){
log.info("timeListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("timeListenerContainerFactory 处理数据内容：{}",item));
}
关于 AckMode.TIME 模式
此模式会在监听器监听到消息后的 PollTimeout 的时间段内提交消费请求。上面配置中将 PollTimeout 设置的超时时间为 2 秒。使用请求http://localhost:8060//ack/sendStr/kafka-time/3 发送三条消息，在消费实例接收到消息后在 PollTimeout 到达之前关闭消费端实例，再次启动消费实例会发现之前消费的消息会被再次读取到。而接收到消息 PollTimeout 之后关闭消费实例，则会发现消息已经被成功消费。
COUNT
监听器工厂的配置
/**
* COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
* @param consumerFactory
* @return
*/
@Bean("countListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> countListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setPollTimeout(1500);
factory.getContainerProperties().setAckCount(5);
//配置手动提交offset
//factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.COUNT);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.COUNT);
return factory;
}
指定监听器使用的配置
/**
* COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
* @param message
*/
@KafkaListener(containerFactory = "countListenerContainerFactory" , topics = "kafka-count")
public void onMessageCount(List<Object> message){
log.info("countListenerContainerFactory 处理数据量：{}",message.size());
message.forEach(item -> log.info("countListenerContainerFactory 处理数据内容：{}",item));
}
关于 AckMode.COUNT 模式
此模式会在监听器监听到消息等于 AckCount 的数量提交消费请求。上面配置中将 `AckCount`` 设置为 5 条消息。使用请求http://localhost:8060//ack/sendStr/kafka-count/9 向队列发送 9 条数据，然后关闭消费端实例，然后再次启动时发现最后 4 条消息再次被读取到。此时前 5 条数据被消费，最后四条消息需要凑齐 5 条数据后才能消费。