当前位置:   article > 正文

SpringBoot集成Kafka

springboot集成kafka

SpringBoot集成Kafka

知识索引

  • SpringBoot集成Kafka工程搭建
  • SpringBoot集成Kafka配置
  • SpringBoot集成Kafka生产消息
  • SpringBoot集成Kafka消费消息

1 生产者

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

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml
  • 1
  • 2
  • 3
  • 4
  • 5

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
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

1.2 启动类

创建启动类com.itmentu.KafkaProducerApplication,代码如下:

@SpringBootApplication
public class KafkaProducerApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaProducerApplication.class,args);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

参数说明:

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

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";
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

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

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

image-20211218165744696

2 消费者

SpringBoot集成Kafka实现消费者消费消息,流程如下:

1:pom.xml引入依赖包

2:创建启动类

3:配置核心配置文件application.yml
  • 1
  • 2
  • 3
  • 4
  • 5

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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32

2.2 启动类

创建启动类com.itmentu.KafkaConsumerApplication,代码如下:

@SpringBootApplication
public class KafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class,args);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

参数说明:

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

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);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

测试效果如下:

image-20211218171245637

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/不正经/article/detail/471654
推荐阅读
相关标签
  

闽ICP备14008679号