Spring Kafka 批量消费

Kafka 作为一个分布式发布订阅的消息系统，是目前最流行的消息队列之一，批量消费在现实业务场景中可以提高 kafka 消费吞吐量。Spring 框架可以使用 @KafkaListener 注解来实现消费端批量消费的功能。

Spring for Apache Kafka

以下 Demo 基于 SpringBoot 2.1.1，spring-kafka-2.3

引入依赖

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

配置 yml 文件

spring:
  kafka:
    bootstrap-servers: 192.168.10.1:9093,192.168.10.2:9093
    consumer:
      enable-auto-commit: false
      max-poll-records: 500
      auto-offset-reset: latest
      group-id: group-dev
    listener:
      concurrency: 1

kafka:
  leaface:
    topic: test-topic

Kafka 配置类

package com.leaface.test.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
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 org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * kafka 消费端配置
 *
 * @author leaface
 */
@Configuration
@EnableKafka
@Slf4j
public class KafkaConsumerConfig {

    /**
     * 以逗号分隔的主机：端口对列表，用于建立与Kafka群集的初始连接
     */
    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    /**
     * 一次调用poll()操作时返回的最大记录数，默认值为500
     */
    @Value("${spring.kafka.consumer.max-poll-records}")
    private int maxPollRecords;

    /**
     * 监听器容器中运行的线程数
     */
    @Value("${spring.kafka.listener.concurrency}")
    private int concurrency;

    /**
     * Consumer.poll() 超时时间, 默认5000
     */
//    @Value("${spring.kafka.listener.poll-timeout}")
//    private int pollTimeout;

    /**
     * 如果为true，则消费者的偏移量将在后台定期提交，默认值为true
     */
    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private boolean enableAutoCommit;

    /**
     * 心跳与消费者协调员之间的预期时间（以毫秒为单位），默认值为3000
     */
//    @Value("${spring.kafka.consumer.auto-commit-interval}")
//    private String autoCommitInterval;

    /**
     * 当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量 可选的值为latest, earliest, none, exception
     */
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    /**
     * 用于标识此使用者所属的使用者组的唯一字符串
     */
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    /**
     * 消费者批量工厂
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 并发创建的消费者数量
        factory.setConcurrency(concurrency);
        // 设置为批量消费，每个批次数量在Kafka配置参数中设置 ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        // factory.getContainerProperties().setPollTimeout(pollTimeout);
        // 手动提交ackMode
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // factory.getContainerProperties().setSyncCommits(false);
        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, servers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        // props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 10485760);
        // props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 10485760);
        return props;
    }

}

Consumer 端

/**
 * 单条消费
 */
// @KafkaListener(topics = "#{'${kafka.leaface.topic}'.split(',')}")
// @KafkaListener(topics = "#{'${kafka.leaface.topic}'}", groupId = "id")
// @KafkaListener(topicPartitions = {@TopicPartition(topic = "#{'${kafka.leaface.topic}'}", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0", relativeToCurrent = "true"))})
public void onMessage(@Payload ConsumerRecord<String, String> record) {
    try {
      //  processMsg(record);
    } catch (Exception e) {
        log.error("process msg error！", e);
    }
}

/**
 * 批量消费
 */
@KafkaListener(topics = "#{'${kafka.leaface.topic}'}")
public void onMessage(@Payload List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
    processMsg(recordList, ack);
}

private void processMsg(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
    try {
        for (ConsumerRecord<String, String> record : records) {
           // do something
        }
    } catch (Exception e) {
        log.error("process msg error!", e);
    } finally {
        // 手动提交(默认同步提交)
        ack.acknowledge();
    }
}