Spring Cloud Stream 和 Kafka实战教程

一时失言乱红尘 2024-04-06 13:31 64阅读 0赞

本文通过实战项目基于事件驱动架构实现实时流处理应用，使用到Spring Boot、Spring mvc、Spring Cloud Stream、Apache Kafka、Lombok。读完基本能够理解基于流的微服务应用。

### 项目需求 ###

1.  从Restful API获取消息
2.  给kafka主题发送消息
3.  从kafka主题读取消息
4.  把消息输出到控制台

让我们开始吧。

### 搭建环境 ###

Spring Cloud Stream是基于Spring Boot构建消息驱动微服务框架。Kafka是高性能的、可伸缩的分布式消息代理平台。Lombok让我们简化java Beandinginess。下面简单介绍下安装Kafka，并新建SpringBoot项目。

1.  下载kafka

在 [官网][Link 1] 下载kafka，注意和你本机的scala版本一致。解压后启动kafka服务。

1.  启动kafka

启动zookeeper， 如果windows环境，bin/window/对应的.bat文件：

$ bin/zookeeper-server-start.sh config/zookeeper.properties

启动kafka， 如果windows环境，bin/window/对应的.bat文件：

$ bin/kafka-server-start.sh config/server.properties

#### Spring Boot项目 ####

使用Idea新建Spring Boot项目，并增加下列依赖：

spring-cloud-stream、spring-kafka、spring-cloud-stream-binder-kafka、spring-boot-starter-web、spring-boot-starter-actuator、lombok、spring-boot-devtools。完整代码如下：

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <groupId>com.dataz</groupId>
        <artifactId>streamkafka</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>streamkafka</name>
        <description>streamkafka</description>
    
        <properties>
            <java.version>1.8</java.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <spring-boot.version>2.3.7.RELEASE</spring-boot.version>
            <spring-cloud.version>Hoxton.SR9</spring-cloud.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-actuator</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-binder-kafka</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-devtools</artifactId>
                <scope>runtime</scope>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-stream-test-support</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>org.springframework.cloud</groupId>
                    <artifactId>spring-cloud-dependencies</artifactId>
                    <version>${spring-cloud.version}</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
                <dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-dependencies</artifactId>
                    <version>${spring-boot.version}</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
            </dependencies>
        </dependencyManagement>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.1</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <version>2.3.7.RELEASE</version>
                    <configuration>
                        <mainClass>com.dataz.streamkafka.StreamkafkaApplication</mainClass>
                    </configuration>
                    <executions>
                        <execution>
                            <id>repackage</id>
                            <goals>
                                <goal>repackage</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    
    </project>

### 定义Kafka Stream ###

我们的项目需要于kafka进行通讯，因此需要定义输出流（往kafka主题写消息）、输入流（从kafka主题中读消息）。Spring Cloud 提供了便捷方式实现上述功能，仅需要简单创建接口，给每个流向指定相应方法。代码如下：

import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.SubscribableChannel;
    
    public interface GreetingsStreams {
        String INPUT = "greetings-in";
        String OUTPUT = "greetings-out";
    
        /**
         * input
         * @return SubscribableChannel
         */
        @Input(INPUT)
        SubscribableChannel inboundGreetings();
    
        /**
         * output
         * @return MessageChannel
         */
        @Output(OUTPUT)
        MessageChannel outboundGreetings();
    }

inboundGreetings() 方法定义输入流从 Kafka 读取消息，outboundGreetings() 方法定义输出流往kafka写消息。应用运行时Spring会创建java代理实现GreetingsStreams 接口，它可作为Spring Bean注入至任何需要访问两个流向的业务。

### 配置Spring Cloud Stream ###

接下来配置Spring Cloud Stream，绑定至流接口GreetingsStreams 。通过@Configuration注解实现：

import org.springframework.cloud.stream.annotation.EnableBinding;
    
    @EnableBinding(GreetingsStreams.class)
    public class StreamsConfig {
    }

@EnableBinding注解指定GreetingsStreams接口实现绑定。

### 配置kafka属性 ###

Spring Boot项目默认生成了application.properties,但我习惯使用yaml文件，新建application.yaml文件，配置下列内容：

spring:
      cloud:
        stream:
          kafka:
            binder:
              brokers: localhost:9092
          bindings:
            greetings-in:
              destination: greetings
              contentType: application/json
            greetings-out:
              destination: greetings
              contentType: application/json

上面配置了kafka代理地址，kafka主题，这里输入流和输出流必须使用相同主题。contentType熟悉告诉Spring Cloud Stream发送接收消息对象作为json String.

### 创建消息对象 ###

创建Greetings类，表示消息对象，用于读取或写入消息：

import lombok.*;
    
    @Getter
    @Setter
    @ToString
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public class Greetings {
        private long timestamp;
        private String message;
    }

上面注解都是lombok的注解，比较直观，不再解释。

### 创建服务写kafka ###

创建GreetingsService 类，sendGreeting方法实现往kafka主题写Greetings消息：

import lombok.extern.slf4j.Slf4j;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHeaders;
    import org.springframework.stereotype.Service;
    import org.springframework.util.MimeTypeUtils;
    
    @Service
    @Slf4j
    public class GreetingsService {
        private final GreetingsStreams greetingsStreams;
    
        public GreetingsService(GreetingsStreams greetingsStreams) {
            this.greetingsStreams = greetingsStreams;
        }
    
        public void sendGreeting(final Greetings greetings) {
            log.info("Sending greetings {}", greetings);
    
            MessageChannel messageChannel = greetingsStreams.outboundGreetings();
            messageChannel.send(MessageBuilder
                    .withPayload(greetings)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                    .build());
        }
    }

@Service 注解配置该类作为Spring Bean，通过构造函数注入GreetingsStreams。

@Slf4j 注解生成slf4j日志，用于使用日志。

### 创建REST API ###

现在创建REST API，用于触发发送消息给kafka，通过注入GreetingsService Bean实现：

import org.springframework.http.HttpStatus;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.ResponseStatus;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class GreetingsController {
        private final GreetingsService greetingsService;
    
        public GreetingsController(GreetingsService greetingsService) {
            this.greetingsService = greetingsService;
        }
    
        @GetMapping("/greetings")
        @ResponseStatus(HttpStatus.ACCEPTED)
        public void greetings(@RequestParam("message") String message) {
            Greetings greetings = Greetings.builder()
                    .message(message)
                    .timestamp(System.currentTimeMillis())
                    .build();
    
            greetingsService.sendGreeting(greetings);
        }
    }

@RestController注解告诉Spring这是Controllerbean。greetings方法定义入口点 `HTTP GET /greetings` 接收消息，最终通过sendGreeting发送消息。

### 监听kafka主题greetings ###

创建GreetingsListener类，监听greetings主题并在控制台打印日志：

import lombok.extern.slf4j.Slf4j;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    @Component
    @Slf4j
    public class GreetingsListener {
        @StreamListener(GreetingsStreams.INPUT)
        public void handleGreetings(@Payload Greetings greetings) {
            log.info("Received greetings: {}", greetings);
        }
    }

@Component 注解与@Service、@RestController类似，定义Spring Bean。GreetingsListener 只有一个方法handleGreetings，每次从kafka主题greetings读取新消息时会执行该方法。这是因为@StreamListener注解配置在该方法上。

### 运行程序 ###

该类是Spring Boot项目生成的：

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

不需要任何修改，直接运行下列命令：

mvn spring-boot:run

启动成功后，在浏览器中输入下列内容，然后检查控制台：

localhost:8080/greetings?message=hello

本文介绍了Spring Cloud Stream 搭建与Kafka交互的事件流处理微服务应用。

[Link 1]: https://archive.apache.org/dist/kafka/3.2.1/kafka_2.12-3.2.1.tgz