Spring Boot 2.x实战94 - 事件驱动3 - Kafka与Kafka Streams_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 Kafka 、Lombok

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