当前位置:   article > 正文

Spring Boot 2.x实战94 - 事件驱动3 - Kafka与Kafka Streams_boot kafkasteam

boot kafkasteam

3.Kafka

Apache Kafka以Topic为导向,提供消息中间件的功能。一个类型的数据称之为一个Topic

3.1 安装Kafka

使用docker compose安装Apache Kafka。

stack.yml

version: '3.1'

services:
  zookeeper:
    image: wurstmeister/zookeeper
    restart: always

  kafka:
    image:  wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: localhost
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

执行命令:

$ docker-compose -f stack.yml up -d
3.2 新建应用

新建应用,信息如下:

Group:top.wisely

Artifact:learning-kafka

Dependencies:Spring for KafkaLombok

build.gradle文件中的依赖如下:

dependencies {
   
	implementation 'org.springframework.boot:spring-boot-starter'
	implementation 'org.springframework.kafka:spring-kafka'
	implementation 'com.fasterxml.jackson.core:jackson-databind' //消息序列化、反序列化使用
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
  //...
}
3.3 Spring Boot的自动配置

Spring Boot给我们提供的自动配置为:KafkaAutoConfiguration,它所做的自动配置有:

  • 注册Kafka消息操作的BeankafkaTemplate
  • 注解kafkaAdmin的Bean用来在应用中新建Topic;
  • 使用@EnableKafka注解开启Kafka的支持;
  • 对Kafka Streams的注解驱动配置的支持。

通过KafkaProperties使用spring.kafka.*来配置Kafka。

3.4 示例
  • 连接Kafka并配置

    spring:
      kafka:
        bootstrap-servers: #1
          - localhost:9092
        producer: # 2
    #      key-serializer: org.springframework.kafka.support.serializer.JsonSerializer #3
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer #4
        consumer: #5
    #      key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer #6
          value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer #7
          group-id: consumers #8
          properties:
            spring:
              json:
                trusted:
                  packages: top.wisely.learningkafka.messaging #9
    
    1. 配置连接Kafka服务器,若我们是Kafka集群可连接多个服务器;
    2. 配置Kafka消息提供者;
    3. 消息提供者对Key序列化的方式,默认为StringSerializer,本例的Key为String类型,可省略;
    4. 消息提供者对消息序列化的方式;
    5. 配置Kafka消息消费者;
    6. 消息消费者对Key反序列化的方式,默认为StringDeserializer,符合默认,可省略;
    7. 消息消费者对消息反序列化的方式;
    8. 配置消费者的group-id,若多个应用的消费者的group-id相同可负载消费消息;
    9. top.wisely.learningkafka.messaging加入消费者信任的发序列化包中。
  • 消息定义

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

闽ICP备14008679号