欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Spring Boot 整合——Spring Boot整合kafka整合

程序员文章站 2022-05-01 16:36:02
...

Spring Boot 整合之前的内容

项目名称 描述 地址
base-data-mybatis 整合mybatis-plus(实际上官方教程已经很多,只做了自定义插件) 未完成
base-jpa JPA基础使用 JPA 数据模型定义
base-jpa-query JPA多表关联使用 JPA 数据模型关联操作
base-log 日志配置 SpringBoot日志配置
base-rabbit rabbitMQ简单使用 RabbitMQ基础使用
base-rabbit3 rabbitMQ一些自定义配置 消息确认回调、消息转换以及消息异常处理
base-rabbit-delay rabbitMQ延时队列 延时队列和消息重试
base-redis redis简单使用 RedisTemplate基础使用;Redis实现简单的发布订阅以及配置序列化方式
base-redis-lock redis分布式锁 Redis分布式锁的简单实现
base-redis-delay 基于有赞的延时消息方案的简单实现 延时队列的简单实现
base-swagger swagger使用 wagger2使用
base-mongodb mongodb简单使用 MongoDB安装以及Spring Boot整合,MongoDB实体创建以及简单CRUD,MongoDB聚合操作,MongoDB分组去重以及MongoDB联表查询

关于版本

依赖 版本
springboot 2.0.8.RELEASE
mongodb 4.0.14

项目地址

因为涉及的代码较多,所以并不会贴出所有代码。本篇文章涉及的源码下载地址:springboot-samples

kafka消息请求的原理

这里简单的介绍下kafka发送消息的流程。(更详细的原理在网络上已经有足够的文章,这里不再赘述)。另外关于如何安装kafka目前也存在相当多的文章。

kafka相关名词

  • broker
  • producer
  • consumer
  • topic
  • partition

producer

Producer 作为消息的生产者,负责生产数据推入broker中。

consumer

Producer 作为消息的消费者,从broker中的某个topic中获取数据。

broker

broker 作为消息服务载体,一般又多个kafka server组成。

topic

kafka对消息进行拆分的方式,不同的数据被保存在不同的topic中,通过将消息推入某个topic或者从某个topic中获取消息来实现消息业务的进行。

partition

每个topic拆分成多个partition,每个partition又由一个一个消息组成。每个消息都被标识了一个递增***代表其进来的先后顺序,并按顺序存储在partition中。

consumer group

关于用户组可以简单理解为。假如有十个consumer订阅同一个topci其groupid不同,则这个topic中的任一条消息被消费10次。如果其groupid相同,此消息只能被其中一个消费。

整个消息生产流程

topic1
推入数据到topic1
推入数据到topic1
数据保存到partition0
数据保存到partition1
数据同步
数据同步
数据同步
数据同步
topic1
partition0-leader
partition1-leader
partition0-following1
partition0-following2
partition1-following1
partition1-following2
producer1
producer2

消息的消费流程

groupid2 groupid1
消息被用户组groupid1消费
消息被用户组groupid2消费
此时只能被一个此组中的一个消费
此时只能被一个此组中的一个消费
consumer21
groupid2
consumer22
consumer23
groupid1
consumer11
consumer12
consumer13
topic1
consumer

关于Spring-Boot整合kafka

关于依赖

  • 主项目依赖
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.8.RELEASE</version>
        <relativePath/> 
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

  • 子项目依赖
    <dependencies>
        <dependency>
            <artifactId>base-core</artifactId>
            <groupId>daifyutils</groupId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
    </dependencies>

spring-kafka配置

spring:
  application:
    name: base.kafka
  kafka:
    bootstrap-servers: kafka服务地址1:端口,kafka服务地址2:端口,kafka服务地址3:端口
    producer:
      # 写入失败时,重试次数。当leader节点失效,一个repli节点会替代成为leader节点,此时可能出现写入失败,
      # 当retris为0时,produce不会重复。retirs重发,此时repli节点完全成为leader节点,不会产生消息丢失。
      retries: 0
      #procedure要求leader在考虑完成请求之前收到的确认数,用于控制发送记录在服务端的持久化,其值可以为如下:
      #acks = 0 如果设置为零,则生产者将不会等待来自服务器的任何确认,该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下,无法保证服务器已收到记录,并且重试配置将不会生效(因为客户端通常不会知道任何故障),为每条记录返回的偏移量始终设置为-1。
      #acks = 1 这意味着leader会将记录写入其本地日志,但无需等待所有副本服务器的完全确认即可做出回应,在这种情况下,如果leader在确认记录后立即失败,但在将数据复制到所有的副本服务器之前,则记录将会丢失。
      #acks = all 这意味着leader将等待完整的同步副本集以确认记录,这保证了只要至少一个同步副本服务器仍然存活,记录就不会丢失,这是最强有力的保证,这相当于acks = -1的设置。
      #可以设置的值为:all, -1, 0, 1
      acks: 1
    consumer:
      group-id: testGroup
      # smallest和largest才有效,如果smallest重新0开始读取,如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
      auto-offset-reset: earliest
      # 设置自动提交offset
      enable-auto-commit: true
      # 如果'enable.auto.commit'为true,则消费者偏移自动提交给Kafka的频率(以毫秒为单位),默认值为5000。
      auto-commit-interval: 100
      max-poll-records: 5

server:
  port: 8060

对topic的操作

基于topic的操作需要我们在项目中实例化下面的bean。SpringBoot对topic的操作主要是通过KafkaAdmin进行操作

/**
 * 主要是初始化对kafka进行操作的admin对象
 * @author daify
 */
@Configuration
@ConditionalOnClass(KafkaAdmin.class)
@EnableConfigurationProperties(KafkaProperties.class)
public class KafkaBaseConfiguration {

    private final KafkaProperties properties;


    public KafkaBaseConfiguration(KafkaProperties properties) {
        this.properties = properties;
    }

    /**
     * 初始化对kafka执行操作的对象
     * @return
     */
    @Bean
    public KafkaAdmin kafkaAdmin() {
        KafkaAdmin admin = new KafkaAdmin(this.properties.buildProducerProperties());
        return admin;
    }

    /**
     * 初始化操作连接
     * @return
     */
    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }
}

新增topic

假如希望操作kafka中的topic可以使用下面的代码

  • 新增
    /**
     * 创建topic
     * @param topicName
     * @return
     */
    public String createTopic(String topicName) {
        NewTopic topic = new NewTopic(topicName, 2, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));
        return topicName;
    }

查看topic

需要查询指定topic的信息可以使用下面的方式

    /**
     * 查询topic
     * @param topicName
     * @return
     */
    public String queryTopic(String topicName) {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topicName));
        StringBuffer sb = new StringBuffer("topic信息:");
        try {
            result.all().get().forEach((k,v)->sb.append("key").append(k).append(";v:").append(v));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return sb.toString();
    }

比如上面我们创建的topic用上面代码查询可以得到下面内容

name=createTopic3, 
internal=false, 
partitions=(
	partition=0, 
	leader=localhost:9093 (id: 1 rack: null),
	replicas=localhost:9093 (id: 1 rack: null), 
	localhost:9094 (id: 2 rack: null), isr=localhost:9093 (id: 1 rack: null), localhost:9094 (id: 2 rack: null)
	),
(
	partition=1, 
	leader=localhost:9094 (id: 2 rack: null), 
	replicas=localhost:9094 (id: 2 rack: null), 
	localhost:9095 (id: 3 rack: null), isr=localhost:9094 (id: 2 rack: null), localhost:9095 (id: 3 rack: null)
)

删除topic

需要删除指定topic的信息可以使用下面的方式

    /**
     * 删除topic
     * @param topicName
     * @return
     */
    public String deleteTopic(String topicName) {
        adminClient.deleteTopics(Arrays.asList(topicName));
        return topicName;
    }

消息的生产和消费

消息的生产

需要向kafka发送消息可以使用下面的语句。

/**
 * kafka消息发送者
 */
@Component
public class KafkaSender {

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    /**
     *  发送文字消息
     * @param message
     * @return
     */
    public String sendStr(String message){
        kafkaTemplate.send(KafkaConfig.TOPIC1,message);
        return message;
    }

    /**
     *  发送对象消息
     * @param obj
     * @return
     */
    public String sendObj(Object obj){
        String message = JSON.toJSONString(obj);
        kafkaTemplate.send(KafkaConfig.TOPIC2,message);
        return message;
    }
}

消息的消费

Spring-Boot封装了大量kafka的方法。所以仅仅是需要获得某些topic的数据可以使用这种方法创建消息监听器


@Component
@Log4j2
public class KafkaConsumerListener {

    @KafkaListener(topics = "kafka-topic1")
    public void onMessage1(String message){
        System.out.println(message);
        log.info("kafka-topic1接收结果:{}",message);
    }

    @KafkaListener(topics = "kafka-topic2")
    public void onMessage2(String message){
        System.out.println(message);
        log.info("kafka-topic2接收结果:{}",message);
    }
}

上面的代码订阅量两个不同的topic。而我们在初始化配置的时候有一项group-id: testGroup这个时候项目启动时,使用上面的方法消费者实例会被归入到testGroup用户组。

当我们向某个topic发送消息的时候控制台会输入下面内容

{"id":100,"name":"message对象","type":2}
2020-05-02 00:09:01.400  INFO 20296 --- [ntainer#0-0-C-1] d.s.k.b.consumer.KafkaConsumerListener   : kafka-topic2接收结果:{"id":100,"name":"message对象","type":2}

到目前,springboot关于kafka的简单操作就截止了。后面我会介绍一些其他的内容。


个人水平有限,上面的内容可能存在没有描述清楚或者错误的地方,假如开发同学发现了,请及时告知,我会第一时间修改相关内容。假如我的这篇内容对你有任何帮助的话,麻烦给我点一个赞。你的点赞就是我前进的动力。