kafka 集成SpringBoot_kafka集成spring boot

目录

一、Kraft模式

配置

集群启动停止脚本

二、集成SpringBoot

资源配置

生产者

消费者

三、API

依赖

yml格式配置文件

properties格式配置文件

---

 

一、Kraft模式

        kafka 2.8.0后新特性

        2.8.0以前，kafka需要搭配zookper，这样会产生大量的网络通信，影响效率，也不便于后续的维护和升级。

 

配置

 

集群启动停止脚本

二、集成SpringBoot

 

资源配置

生产者

 

 

消费者

 加上消费者组id和反序列化

三、API

依赖

        <!-- Springboot整合 Kafka使用。注意：版本一致   -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.8.1</version>
        </dependency>

yml格式配置文件

spring:
  application:
    name: kafka-springboot
  # kafka配置信息
  kafka:
    bootstrap-servers: 192.168.xxx.xxx:9092 # 集群用逗号分隔
    producer: # 生产者
      retries: 3 # 失败重试次数
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer: # 消费者
      group-id: default-group # 消费组
      enable-auto-commit: false
      # auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      ack-mode: MANUAL_IMMEDIATE

 

 

properties格式配置文件

server.port=8080
###########【Kafka集群】###########
spring.kafka.bootstrap-servers=localhost:9092
#==================================【初始化生产者配置】==================================#
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 自定义分区器
#spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner

 

整理的不错的链接：

Spring Boot整合Kafka_Charge8的博客-CSDN博客_springboot整合kafka

kafka客户端操作之Admin API_codingXT的博客-CSDN博客_kafkaadmin

Springboot整合kafka_fFee-ops的博客-CSDN博客_springboot整合kafka