springboot整合kafka，指定分区发送，批量消费，指定topic分区消费

package net.xdclass.base_project.consumerConfig; import net.xdclass.base_project.consumer.KafkaPartitionLintener; import net.xdclass.base_project.consumer.KafkaTopicLintener; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.beans.factory.annotation.Value; 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 java.util.HashMap; import java.util.Map; /** * kafka消费者配置 * @author sqz * */ @Configuration @EnableKafka public class KafkaConsumerConfig { @Value("${spring.kafka.consumer.servers}") private String servers; @Value("${spring.kafka.consumer.enable.auto.commit}") private boolean enableAutoCommit; @Value("${spring.kafka.consumer.auto.commit.interval}") private String autoCommitInterval; @Value("${spring.kafka.consumer.group-id}") private String groupId; @Value("${spring.kafka.consumer.auto-offset-reset}") private String autoOffsetReset; @Value("${spring.kafka.consumer.concurrency}") private int concurrency; @Value("${spring.kafka.consumer.session.timeout.ms}") private int sessionTimeout; @Bean public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.setConcurrency(concurrency); // 拉取超时时间,只限自动提交 factory.getContainerProperties().setPollTimeout(3000); //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG,不设置这个那么批量消费就会报错 factory.setBatchListener(true); return factory; } public ConsumerFactory<String, String> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } public Map<String, Object> consumerConfigs() { Map<String, Object> propsMap = new HashMap<>(); propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit); propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval); propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout); propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); //最大一次拉取50条 propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50); //手动提交设置与poll的心跳数,如果消息队列中没有消息，等待毫秒后，调用poll()方法。 // 如果队列中有消息，立即消费消息，每次消费的消息的多少可以通过max.poll.records配置。 propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 15000); return propsMap; } /** * kafka多分区监听 * @return */ // @Bean // public KafkaPartitionLintener listener() { // return new KafkaPartitionLintener(); // } /** * kafka多topic监听 * @return */ @Bean public KafkaTopicLintener listenerTopic() { return new KafkaTopicLintener(); } }