消息中间件（五）之-SpringBoot集成kafka消息中间件

深藏阁楼爱情的钟 2022-02-05 07:15 568阅读 0赞

springboot集成消息中间件的步骤十分简单、方便，但是针对不同的生产、消费情况需要调整相应的生产消费配置（这是一个繁杂的过程，因具体测试情况而慢慢摸索最适合的配置参数）。

1、引入相应的pom依赖：

<dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka</artifactId>
    </dependency>

2、复写kafka的相关配置类：生产、消费相关配置

@Configuration
    @RefreshScope
    public class KafkaMQConfig {
    
        // 此处配置代替zk
        @Value("${spring.kafka.bootstrap-servers}")
        private String servers;
        // 消费者一次会话时长
        @Value("${spring.kafka.consumer.session-timeout-ms}")
        private String sessionTimeoutMs;
        // 消费组标识
        @Value("${spring.kafka.consumer.group-id}")
        private String groupId;
        // 偏移量的起始点
        @Value("${spring.kafka.consumer.auto-offset-reset}")
        private String autoOffsetReset;
        // 偏移量的提交方式
        @Value("${spring.kafka.consumer.enable-auto-commit}")
        private String enableAutoCommit;
        // 一次从kafka服务拉取的数据量
        @Value("${spring.kafka.consumer.max-poll-records}")
        private String maxPollRecords;
        // 监测消费端心跳时间
        @Value("${spring.kafka.consumer.heartbeat-interval}")
        private String heartbeatInterval;
        // 两次拉取数据的最大时间间隔
        @Value("${spring.kafka.consumer.max-poll-interval-ms}")
        private String maxPollIntervalMs;
    
        
        //生产者相关配置
        @Bean
        public KafkaProducer kafkaProducer() {
            Properties props = new Properties();
            // 设置接入点，请通过控制台获取对应 Topic 的接入点
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            // Kafka 消息的序列化方式
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            // 请求的最长等待时间
            props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
            // 构造 Producer 对象，注意，该对象是线程安全的
            // 一般来说，一个进程内一个 Producer 对象即可
            // 如果想提高性能，可构造多个对象，但最好不要超过 5 个
            return new KafkaProducer<String, String>(props);
        }
    
        // 消费端相关配置
        @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
            props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalMs);
            props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, heartbeatInterval);
            props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return props;
        }
    
        @Bean
        public ConsumerFactory<String, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }
    
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
            return factory;
        }
    
    }

3、配置文件配置：

spring.kafka.consumer.group-id: ******
    spring.kafka.consumer.auto-offset-reset: latest			// 偏移量拉取点
    spring.kafka.consumer.enable-auto-commit: false			// 手动提交
    spring.kafka.consumer.max-poll-records: 400				// 每次拉取数据量
    spring.kafka.consumer.heartbeat-interval: 30000			// 监测消费端服务心跳时间
    spring.kafka.consumer.session-timeout-ms: 100000
    spring.kafka.consumer.max-poll-interval-ms: 300000 		// 两次拉取数据的最大时间间隔
    spring.kafka.listener.concurrency = 4 					// 最大线程数
    spring.kafka.bootstrap-servers = host:port,host0:port0,host1:port1

4、生产、消费的伪代码：

@Autowired
        private KafkaProducer<String, String> kafkaProducer;
    	
    	// 推送消息到kafka 指定topic
    	public void sendMessage(String topic, String data) {
            Future<RecordMetadata> metadataFuture = kafkaProducer.send(new ProducerRecord<>(topic, data));
            RecordMetadata recordMetadata = metadataFuture.get();
    	}
    	
    	// 从指定的topic上拉取消息消费 确认机制ack
    	@KafkaListener(topics = {Topic.CUSTOMER_QRCODETYPE}, containerFactory = "kafkaListenerContainerFactory")
        public void processMessage10And20(ConsumerRecord<String, String> record, Acknowledgment ack) throws IOException {
            ******业务处理
    		ack.acknowledge();
        }

以上为springboot 集成kafka消息中间件的相关配置及开发伪代码，希望对于同路上的小伙伴有所帮助。

**如有披露或问题欢迎留言或者入群探讨**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NDcwNzMz_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NDcwNzMz_size_16_color_FFFFFF_t_70]: /images/20220205/7caade1840c342899d153aadd1458715.png