SpringBoot集成kafka

SpringBoot集成kafka

集成kafka

pom引入

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4.     <version>2.8.11</version>
5. </dependency>

复制代码

设置kafka

1. spring:
2.   kafka:
3.     bootstrap-servers: 127.0.0.1:9092
4.     producer:
5.       # 发生错误后，消息重发的次数。
6.       retries: 0
7.       #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
8.       batch-size: 16384
9.       # 设置生产者内存缓冲区的大小。
10.       buffer-memory: 33554432
11.       # 键的序列化方式
12.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
13.       # 值的序列化方式
14.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
15.       # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
16.       # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
17.       # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
18.       acks: 1
19.     consumer:
20.       # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
21.       auto-commit-interval: 1S
22.       # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
23.       # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
24.       # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
25.       auto-offset-reset: latest
26.       # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
27.       enable-auto-commit: false
28.       # 键的反序列化方式
29.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
30.       # 值的反序列化方式
31.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
32.       # 设置默认消费者分组id
33.       group-id: dev
34.     listener:
35.       # 在侦听器容器中运行的线程数。
36.       concurrency: 5
37.       #listner负责ack，每调用一次，就立即commit
38.       ack-mode: manual_immediate
39.       missing-topics-fatal: false

复制代码

kafka生产者示例

1. @Component
2. @Slf4j
3. public class KafkaProducer {
5.     @Autowired
6.     private KafkaTemplate<String, Object> kafkaTemplate;
8.     // 发送kafka消息
9.     public void send(String topic, Object data) {
10.         if (StringUtils.isEmpty(topic)) {
11.             log.error("Kafka topic不能为空");
12.             throw new IllegalArgumentException("Kafka topic不能为空");
13.         } else if (ObjectUtil.isEmpty(data)) {
14.             log.error("Kafka消息不能为空");
15.             throw new IllegalArgumentException("Kafka消息不能为空");
16.         } else {
17.             // 将数据转换为JSON字符串
18.             String jsonStr = JSONUtil.toJsonStr(data);
19.             // 向kafka发送消息
20.             ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, jsonStr);
21.             future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
22.                 @Override
23.                 public void onFailure(Throwable throwable) {
24.                     //发送失败的处理
25.                     log.info(topic + " - 生产者 发送消息失败：" + throwable.getMessage());
26.                 }
27.                 @Override
28.                 public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
29.                     //成功的处理
30.                     log.info(topic + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
31.                 }
32.             });
33.         }
35.     }
36. }

复制代码

kafka消费者示例

1. @Component
2. @Slf4j
3. public class KafkaConsumer {
5.     @KafkaListener(topics = "testTopic")
6.     public void topicMessage(String message){
7.         log.info("Received Message in topic: {}", message);
8.         // 处理接收到的消息
9.     }
11.     @KafkaListener(topics = "testTopic2")
12.     public void topicMessage2(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
13.         try {
14.             Optional message = Optional.ofNullable(record.value());
15.             if (message.isPresent()) {
16.                 log.info("Received Message in topic: {}", message.get());
17.             }
18.             // 处理接收到的消息
19.         } catch (Exception e) {
20.             log.error("Error processing message", e);
21.         } finally {
22.             ack.acknowledge();
23.         }
25.     }
27. }

复制代码

kafka设置批量并发消费

设置批量消费

1. spring:
2.   kafka:
3.     consumer:
4.       # 一次 poll 最多返回的记录数
5.       max-poll-records: 3100
6.       max-partition-fetch-bytes: 15728640                 #设置拉取数据的大小,15M
7.     listener:
8.       type: batch # 开启批量消费

复制代码

设置并发消费

为了加速消费，我们可以提高并发数，好比下面设置我们将并发设置为 3。注意：并发量根据现实分区数决定，必须小于等于分区数，否则会有线程不停处于空闲状态

1. spring:
2.   kafka:
3.     consumer:
4.       # 一次 poll 最多返回的记录数
5.       max-poll-records: 3100
6.       max-partition-fetch-bytes: 15728640                 #设置拉取数据的大小,15M
7.     listener:
8.       type: batch # 开启批量消费
9.       concurrency: 3 # 设置并发数

复制代码

批量消费示例

1. @KafkaListener(topics = "testTopic")
2. public void batchListener(List<ConsumerRecord<?,?>> records, Acknowledgment ack){
3.     try {
4.         records.forEach(record -> {
5.             // TODO 业务代码
6.         });
7.     } catch (Exception e) {
8.         log.error("Kafka监听异常"+e.getMessage(),e);
9.     } finally {
10.         ack.acknowledge();//手动提交偏移量
11.     }
12. }

复制代码

除了通过设置文件设置外，还可以通过自界说设置类的方式实现批量消费，但是相对yml设置来说还是有点贫苦的：

1. /**
2. * 消费者配置
3. */
4. @Configuration
5. public class KafkaConsumerConfig {
7.     /**
8.      * 消费者配置
9.      * @return
10.      */
11.     public Map<String,Object> consumerConfigs(){
12.         Map<String,Object> props = new HashMap<>();
13.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
14.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
15.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
16.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
17.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
18.         return props;
19.     }
21.     @Bean
22.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> batchFactory() {
23.         ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
24.         factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
25.         //并发数量
26.         factory.setConcurrency(3);
27.         //开启批量监听
28.         factory.setBatchListener(true);
29.         return factory;
30.     }
31. }

复制代码

同时监听器通过@KafkaListener注解的containerFactory 设置指定批量消费的工厂即可，如下：

1. @KafkaListener(topics = {"testTopic"},containerFactory = "batchFactory")
2. public void batchListener(List<ConsumerRecord<?,?>> records, Acknowledgment ack){
3.     try {
4.         records.forEach(record -> {
5.             // TODO 业务代码
6.         });
7.     } catch (Exception e) {
8.         log.error("Kafka监听异常"+e.getMessage(),e);
9.     } finally {
10.         ack.acknowledge();//手动提交偏移量
11.     }
12. }

复制代码