SpringBoot集成Kafka

1 生产者

SpringBoot集成Kafka，无非就是生产者和消费者，但首先得实现SpringBoot集成，流程如下：

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml

1.1 pom.xml

在pom.xml中引入spring-kafka，代码如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.itmentu</groupId>
    <artifactId>kafka-producer-01</artifactId>
    <version>1.0-SNAPSHOT</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.8.RELEASE</version>
    </parent>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>
    </dependencies>
</project>

1.2 启动类

创建启动类com.itmentu.KafkaProducerApplication，代码如下：

@SpringBootApplication
public class KafkaProducerApplication {

    public static void main(String[] args){
        SpringApplication.run(KafkaProducerApplication.class,args);
    }
}

1.3 核心配置文件

创建application.yml核心配置文件，代码如下：

server:
  port: 18081
spring:
  kafka:
    bootstrap-servers: 192.168.211.130:9092,192.168.211.130:9093,192.168.211.130:9094
    producer: # producer 生产者
      retries: 0 # 重试次数
      acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      batch-size: 16384 # 批量大小
      buffer-memory: 33554432 # 生产端缓冲区大小
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

参数说明：

bootstrap-servers:Kafka服务地址
retries:发送失败后，重试次数，0表示不重试
key-serializer:向Kafka发送数据，key采用的序列化方式
value-serializer:向Kafka发送数据，数据采用的序列化方式

1.4 生产者消息发送

消息发送使用KafkaTemplate实现消息发送，创建com.itmentu.controller.SendController，实现消息发送，代码如下：

@RestController
@RequestMapping(value = "/producer")
public class SendController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /***
     * 发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @GetMapping(value = "/send/{topic}/{msg}")
    public String send(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg){
        //消息发送
        kafkaTemplate.send(topic,msg);
        return "SUCCESS";
    }
}

发送消息的时候，使用的是kafkaTemplate.send(topic,msg)，第1个参数是队列名字或者叫主题名字，第2个参数是发送的消息。

我们可以请求​​http://localhost:18081/producer/send/itmentu/hello​​测试，eagle打开，可以看到效果如下：

2 消费者

SpringBoot集成Kafka实现消费者消费消息，流程如下：

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml

2.1 pom.xml

和生产者一样，都需要引入spring-kafka依赖，配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.itmentu</groupId>
    <artifactId>kafka-consumer-01</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.8.RELEASE</version>
    </parent>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>
    </dependencies>
</project>

2.2 启动类

创建启动类com.itmentu.KafkaConsumerApplication，代码如下：

@SpringBootApplication
public class KafkaConsumerApplication {

    public static void main(String[] args){
        SpringApplication.run(KafkaConsumerApplication.class,args);
    }
}

2.3 核心配置文件

创建application.yml，配置如下：

server:
  port: 18082
spring:
  kafka:
    bootstrap-servers: 192.168.211.130:9092,192.168.211.130:9093,192.168.211.130:9094
    consumer: # consumer消费者
      group-id: mentugroup # 默认的消费组ID
      enable-auto-commit: true # 是否自动提交offset
      auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

参数说明：

bootstrap-servers:Kafka服务地址
group-id:消费者组默认名字
enable-auto-commit:每次消费数据后，需要提交数据的偏移量，这里true表示自动提交，false表示手动提交
key-deserializer:读Kafka数据，key采用的反序列化方式
value-deserializer:读Kafka数据，数据采用的反序列化方式

2.4 消费消息

Kafka消费消息采用@KafkaListener(topics="{}",groupId="")方式消费数据，我们可以创建一个消息监听类实现消息消费，创建com.itmentu.listener.MessageListener，代码如下：

@Component
public class MessageListener {

    @KafkaListener(topics = {"itmentu"},groupId = "itmentuGroup")
    public void listener(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        System.out.println("读取的消息："+message+"\n当前偏移量："+offset);
    }
}

测试效果如下：