springboot 整合 kafka 使用手动提交消费

- 日理万妓 2023-02-10 07:40 61阅读 0赞

## kafka消费者配置文件 ##

import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.apache.kafka.common.serialization.StringSerializer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.annotation.EnableKafka;
    import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
    import org.springframework.kafka.config.KafkaListenerContainerFactory;
    import org.springframework.kafka.core.ConsumerFactory;
    import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
    import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
    import org.springframework.kafka.listener.ContainerProperties;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    @EnableKafka
    public class KafkaConsumerConfig { 
        @Bean(name = "simpleFactory")
        KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() { 
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.setConcurrency(3);
            factory.getContainerProperties().setPollTimeout(3000);
            factory.getContainerProperties().setGroupId("test");
            //当使用手动提交时必须设置ackMode为MANUAL,否则会报错No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
            factory.getContainerProperties().setAckCount(10);
            factory.getContainerProperties().setAckTime(10000);
            return factory;
        }
    
        //@Bean
        public ConsumerFactory<Integer, String> consumerFactory() { 
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }
    
        //@Bean
        public Map<String, Object> consumerConfigs() { 
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.1.24.220:9082");
            //键的序列化方式
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            //值的序列化方式
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            props.put("group.id", "test");
            return props;
        }
    }

## kafka消费者监听文件 ##

@KafkaListener(containerFactory = "simpleFactory", topics = "test")
    public void test(ConsumerRecord<String, String> record, Acknowledgment ack) { 
    	//提交offset
        ack.acknowledge();
        System.out.println(record.value());
    }