SpringBoot 集成 Kafka 高级实现

1、简介

之前博客中记录了直接使用Kafka客户端实现生产者和消费者之间的交互，这种方式通过设置各种参数编码繁琐，因此通过SpringBoot集成Kafka成为一种常用的实现，下面就详细介绍 SpringBoot 是如何和Kafka进行集成的，本文主要参考官网进行学习（Messaging）。

2、引入依赖

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

3、生产者

3.1、生产者配置文件

spring:
  kafka:
    bootstrap-servers: node-1:9092,node-2:9092
    producer:
      # 指定key-value 序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 指定缓冲区一批大小，默认16k
      batch-size: *
      # 指定缓冲区总大小，默认32m
      buffer-memory: *
      # 指定消息确认方式
      acks: -1
      # 指定消息发送压缩方式
      compression-type: snappy
      # 配置额外参数
      properties:
        # 自定义分区
        partitioner.class:
        # 指定发送延迟时间，默认0ms
        linger.ms:

3.2、java代码实现生产者发送消息

public class KafkaProducerController {
    @Autowired
    private KafkaTemplate<String, String> template;

    public String test(){
// 发送消息，send方法有不同的重载
        template.send("topic1", "hello long!");
        return "ok!";
    }
}

3.3、自定义 KafkaTemplate

通过自定义的 KafkaTemplate 可以快速指定需要的生产者参数，能够做到高度可控，灵活编码。

@SpringBootConfiguration
@ConfigurationProperties(prefix="kafka")
public class KafkaProducerConfig {
    private String bootstrap_servers_config;
    private String retries_config;
    private String batch_size_config;
    private String linger_ms_config;
    private String buffer_memory_config;
    private String topic;
    @Bean
    public KafkaTemplate kafkaTemplate(){
        HashMap<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrap_servers_config);
        configs.put(ProducerConfig.RETRIES_CONFIG,retries_config);
        configs.put(ProducerConfig.BATCH_SIZE_CONFIG,batch_size_config);
        configs.put(ProducerConfig.LINGER_MS_CONFIG,linger_ms_config);
        configs.put(ProducerConfig.BUFFER_MEMORY_CONFIG,buffer_memory_config);
        configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,RoundRobinPartitioner.class);
        //设置序列化
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
        //设置自定义分区
        DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory(configs);
        return new KafkaTemplate(producerFactory);
    }
}

4、消费者

4.1、消费者配置文件

spring:
  kafka:
    bootstrap-servers: node-1:9092,node-2:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: test  # 指定消费者组

4.2、消费者java代码实现

@SpringBootConfiguration
public class KafkaConsumerConfig {
    @KafkaListener(topics = {"topic1"})
    public void consumer(String msg){
        System.out.println("consumer massage from kafka: " + msg);
    }
}

4.3、@KafkaListener参数详解

注：topicPartitions和topics、topicPattern不能同时使用

示例如下：

@KafkaListener(id = "test1", // 监听器ID(唯一)
               groupId = "test", // 设置消费者组
               topicPartitions = { // 配置topic和分区：有两个topic，分别为topic1、topic2，topic1只接收分区0，2的消息;
                    // topic2接收分区0和分区1的消息，但是分区1的消费者初始位置为10
                    @TopicPartition(topic = "topic1", partitions = { "0", "2" }),                      
                    @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = 
                                 @PartitionOffset(partition = "1", initialOffset = "9"))},
                    properties = {"enable.auto.commit:false","max.poll.interval.ms:6000"})

参数详解：

参数	描述
topic	指定要监听哪些topic(与topicPattern、topicPartitions 三选一)
topicPattern	匹配Topic进行监听(与topics、topicPartitions 三选一)
topicPartitions	显式分区分配
errorHandler	异常处理,填写 beanName
properties	配置其他属性

注意：

1）、消费者组使用优先级：groupId > id > 配置文件中指定的消费者组。

2）、如果 groupId 不存在，id 存在，但是在注解中将 idIsGroup 设置为 false，则使用配置文件中的消费者组。

3）、@KafkaListener 注解中和配置文件中的相同配置优先级高于配置文件。

4.4、自定义消费者异常

@Component
public class KafkaDefaultListenerErrorHandler implements KafkaListenerErrorHandler {
    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
        return null;
    }

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
    	//do someting
        return null;
    }
}