springboot学习(十五)：Kafka的使用

梦里梦外; 2024-04-18 13:54 54阅读 0赞

# 说明 #

本篇博文主要介绍在springboot中使用kafka，以及如何配置kafka。

# 正文 #

## 简单配置 ##

spring kafka提供了spring对kafka的支持，通过简单的配置我们可以快速使用KafkaTemplate进行消息的发送，使用@KafkaListener注解来进行消息消费。以下内容，通过简单的配置，实现在springboot中快速使用kafka.

### 配置文件 ###

server:
      port: 8080
    
    spring:
      kafka:
        bootstrap-servers: // kafka集群地址 ip:port, ip:port
        producer:
          acks: all // 消息确认模式 等同于-1  值还可以是0 (表示发后即忘) 1 (只要master写入就确认)
          retries: 0 // 消息发送失败时的重试次数
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
    
        consumer:
          enable-auto-commit: false // 表示消息消费后是否自动提交消费位移
          auto-commit-interval: 1000 // 设置自动提交的间隔
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          group-id: grouptest3 // 设置消费者的groupid  这个可以在@KafkaListener中设置 也可以不设置使用默认值
        listener:
          concurrency: 2 // 设置消费者的数量
          poll-timeout: 1000 // 拉取消息时的阻塞时间
          ack-mode: manual_immediate // 设置消息确认模式

--------------------

### 生产者 ###

spring kafka提供了KafkaTemplate类来操作kafka，通过该类实现kafka Producer的功能

@RestController
    public class ProducerController {
    
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        @RequestMapping("/send")
        public String send(@RequestParam("msg") String msg) {
            kafkaTemplate.send("test3", msg);
            return "success";
        }
    }

通过查看源码，在发送消息时，每次都会新建一个Producer对象。在kafka官方文档中，已说明该对象是线程安全的，可以线程共享来得到更好的性能（The producer is thread safe and should generally be shared among all threads for best performance.）。

同时，Producer 采用异步批量的方式发送消息的。在进行发送时，先将消息累积在内存，直到消息数超过设置的大小或延迟查过设定的时间，默认为64k,10ms。参数值可以在配置Producer时通过batch.size和linger.ms指定。这种发送方式是吞吐量和延迟之间的妥协。

--------------------

### 消费者 ###

spring kafka提供@KafkaListener注解来自动创建消费者，我们可以在注解中设置消费者的topics，groupid，以及该消费者组对应消费者的数量，消费的分区等。

@Service
    public class ConsumerService {
    
        @KafkaListener(topics = "test3")
        public void listen (ConsumerRecord<?, ?> record) {
            System.out.printf("topic = %s, offset = %d, value = %s, key = %s%n", record.topic(), record.offset(), record.value(), record.key());
        }
    }

至此，kafka简单配置使用已经完成，接下介绍针对不同的消费者，我们如何配置。

--------------------

## 多配置 ##

在实际使用中，我们可能会要求消费者从不同的集群进行消费，或者采用不同的反序列化方式（该配置要和生产者的序列化方式一样）。对此，我们可以根据配置创建不同ConsumerFactory对象和ConcurrentkafkaListenerConfigFactory对象。

### 配置文件 ###

这里我们希望可以从两个不同的集群中进行消费，通过在配置文件中指定不同的集群地址：

spring:
      kafka:
        consumer:
          bootstrap-servers1: // 第一个集群地址
          bootstrap-servers2: // 第二个集群地址

--------------------

### 配置类 ###

在指定了不同的集群地址后，我们需要为消费者配置对应的factory：

@Configuration
    public class KafkaConfig {
    
        @Value("${spring.kafka.consumer.bootstrap-servers1}")
        private String servers1;
        @Value("${spring.kafka.consumer.bootstrap-servers2}")
        private String servers2;
    
        @Autowired
        private KafkaProperties kafkaProperties;
    
        // server1 consumer config
        @Bean
        public ConsumerFactory<String, String> consumerFactory() {
            KafkaProperties.Consumer consumer = kafkaProperties.getConsumer();
            consumer.setBootstrapServers(Arrays.asList(servers1.split(",")));
            return new DefaultKafkaConsumerFactory<>(consumer.buildProperties());
        }
    
        @Bean
        public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory containerFactory = new ConcurrentKafkaListenerContainerFactory();
            containerFactory.setConsumerFactory(consumerFactory());
            return containerFactory;
        }
    
        // server2 consumer config
        @Bean
        public ConsumerFactory<String, String> server2ConsumerFactory() {
            KafkaProperties.Consumer consumer = kafkaProperties.getConsumer();
            consumer.setBootstrapServers(Arrays.asList(servers2.split(",")));
            return new DefaultKafkaConsumerFactory<>(consumer.buildProperties());
        }
    
        @Bean
        public ConcurrentKafkaListenerContainerFactory server2ListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory containerFactory = new ConcurrentKafkaListenerContainerFactory();
            containerFactory.setConsumerFactory(server2ConsumerFactory());
            return containerFactory;
        }
    }

--------------------

### 使用@KakaListener创建不同的消费者 ###

对于不同的集群，我们配置了不同的topic和groupid

@Service
    public class ConsumerService {
    
        @KafkaListener(containerFactory = "kafkaListenerContainerFactory", topics = "test4", groupId = "grouptest4")
        public void consumeServer1(ConsumerRecord<String, String> record) {
            System.out.printf("topic = %s, offset = %d, value = %s, key = %s%n", record.topic(), record.offset(), record.value(), record.key());
        }
    
        @KafkaListener(containerFactory = "server2ListenerContainerFactory", topics = "test666", groupId = "groups2")
        public void consumeServer2(ConsumerRecord<String, String> record) {
            System.out.printf("topic = %s, offset = %d, value = %s, key = %s%n", record.topic(), record.offset(), record.value(), record.key());
        }
    }

同样地，针对生产者，我们也可以配置生成不同的KafkaTemplate对象进行消息发送。

--------------------

源码地址：https://github.com/Edenwds/springboot\_study/tree/master/kafka\_examples

参考资料：https://thepracticaldeveloper.com/2018/11/24/spring-boot-kafka-config/