赞
踩
- @KafkaListener(
- id = "${spring.kafka.consumer.notice.id}",
- topicPartitions = {@TopicPartition(topic = "${spring.kafka.consumer.notice.topic}", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "0"))},
- groupId = "dvsfdvasdva",
- clientIdPrefix = "exactness-notice-batch",
- containerFactory = "batchFactory"
- )
- public void batchListener(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment) {
- for (ConsumerRecord<?, ?> record : records) {
- try {
- NoticeModel noticeModels = (NoticeModel)JsonUtil.jsonToBean((String) record.value(), NoticeModel.class);
- iBigdateService.exactnessMessageNoticeOne(noticeModels);
- } catch (Exception e) {
- log.error("编号:{},消费数据data:{},异常:{}",uuid,record.value(),e);
- }
- }
- acknowledgment.acknowledge();
- }
-
-
-
- @Bean
- @ConditionalOnMissingBean(name = "batchFactory")
- KafkaListenerContainerFactory<?> batchFactory() {
- ConcurrentKafkaListenerContainerFactory<String, String> factory = new
- ConcurrentKafkaListenerContainerFactory<>();
- //并发数量
- factory.setConcurrency(concurrency);
- factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
- // 开启批量监听
- factory.setBatchListener(batchListener);
- factory.getContainerProperties().setPollTimeout(pollTimeout);
- if (!enableAutoCommit) {//非自动提交需设置ACK方式
- factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
- }
- //配置kafka监听是否自动启动
- factory.setAutoStartup(kafkaListenerFlag);
- return factory;
- }

| 作用 |
---|---|
id | 消费者的id,当GroupId没有被配置的时候,默认id为GroupId |
groupId | 消费组ID,当前消费组,(小心修改导致重复消费) |
| 配置的是监听容器工厂,也就是ConcurrentKafkaListenerContainerFactory,配置BeanName |
| 需要监听的Topic,可监听多个 将topics 配置为:@KafkaListener(topics = “#{’${topics}’.split(’,’)}”) |
| 可配置更加详细的监听信息,必须监听某个Topic中的指定分区,或者从offset为200的偏移量开始监听 |
| topicPattern 已经可以做到定期检查topic列表,然后将新加入的topic分配至某个消费者。 topicPattern = "test_topic2.*" |
clientIdPrefix | 消费者Id前缀 |
| |
| 监听异常处理器,配置BeanName |
| id是否为GroupId |
| 真实监听容器的BeanName,需要在 BeanName前加 "__" |
| 线程数量,消费者个数(注意,消费者数要小于等于你开的所有topic的分区数总和) |
| 是否自动启动 true to auto start, false to not auto start. |
Spring-Boot-kafka的配置中有一个参数,提供了自动消费的功能。使用自动消费我们在收到消息的时候会自动向kafka确认消费。但是在一些特殊场景中我们可能需要使用一些其他的消费模式。
enable-auto-commit: true
需要自己配置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
的枚举中,下面就介绍下各个模式的区别
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 有一个条件满足时提交 |
当每一条记录被消费者监听器(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
- }
当每一条记录被消费者监听器(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
- }
这两种模式都是需要进行手动确认ack.acknowledge();
才能完成消息的消费,否则在重启消费端实例的时候数据会再次被消费端接收到。
关于两者的解释可以看看这个提问 what-is-the-difference-between-manual-and-manual-immediate-in-spring-kafka-ackmo
- /**
- * 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));
- }
使用RECORD模式的时候,当监听器处理完消息后会自动处理,使用此模式不需要手动消费。
- /**
- * 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));
- }
此模式会在监听器监听到消息后的PollTimeout
的时间段内提交消费请求。上面配置中将PollTimeout
设置的超时时间为2秒。使用请求http://localhost:8060//ack/sendStr/kafka-time/3
发送三条消息,在消费实例接收到消息后在PollTimeout
到达之前关闭消费端实例,再次启动消费实例会发现之前消费的消息会被再次读取到。而接收到消息PollTimeout
之后关闭消费实例,则会发现消息已经被成功消费。
- /**
- * 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));
- }
此模式会在监听器监听到消息等于AckCount
的数量提交消费请求。上面配置中将`AckCount``设置为5条消息。使用请求http://localhost:8060//ack/sendStr/kafka-count/9
向队列发送9条数据,然后关闭消费端实例,然后再次启动时发现最后4条消息再次被读取到。此时前5条数据被消费,最后四条消息需要凑齐5条数据后才能消费。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。