关闭自动消费
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
的枚举中,下面就介绍下各个模式的区别
/** * 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, }
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); 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
标签:提交,配置,MANUAL,factory,kafka,详解,监听器,message From: https://www.cnblogs.com/brandon-i/p/17106850.html当每一条记录被消费者监听器(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); 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-ackmoRECORD
监听器工厂的配置/** * 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); 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); 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); 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条数据后才能消费。