基于SpringBoot 使用 Flink 收发Kafka消息的示例详解

基于 SpringBoot 使用 Flink 收发 Kafka 消息主要包含以下步骤：

第一步：创建 SpringBoot 项目

首先我们需要创建一个 SpringBoot 项目。在 pom.xml 文件中添加 flink 和 kafka 相关依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>${spring-kafka.version}</version>
</dependency>

第二步：编写 Flink 代码

在 SpringBoot 项目中，我们可以通过 @Configuration 注解创建一个 Flink Stream Execution Environment，并使用 Flink Connector for Kafka 连接到 Kafka Cluster。

@Configuration
public class FlinkConfig {
    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public StreamExecutionEnvironment streamExecutionEnvironment() {
        // 创建 Stream Execution Environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置 checkpoint 的时间间隔为 10 秒钟
        env.enableCheckpointing(10000);

        // 创建一个 KafkaSource，从指定的主题中读取消息
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", bootstrapServers);
        props.setProperty("group.id", "test");
        props.setProperty("auto.offset.reset", "latest");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), props);

        // 将 KafkaSource 加入 ExecutionEnvironment
        DataStream<String> stream = env.addSource(kafkaConsumer);

        // 转换操作
        DataStream<String> result = stream
            .flatMap( new Splitter() )
            .keyBy(0)
            .timeWindow(Time.seconds(5))
            .reduce( new WordCountReducer() );

        // 输出结果
        result.print();

        // 启动 ExecutionEnvironment
        env.execute("Flink Streaming Kafka Example");

        return env;
    }
}

这段代码中，我们首先创建了一个 StreamExecutionEnvironment，并设置了 checkpoint 的时间间隔为 10 秒钟。然后创建了一个 Kafka 的 Consumer，从指定的 topic 中读取数据，并使用 addSource 将其加入 StreamExecutionEnvironment。接着进行了一些数据的转换操作，并最终将结果打印出来。最后启动 ExecutionEnvironment。

第三步：编写 Kafka Producer 代码

在 SpringBoot 项目中，我们也可以通过 KafkaTemplate 来实现数据的发送操作。以下代码演示了如何使用 KafkaTemplate 发送数据到 Kafka。

@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String sendMessage() {
        // 构造一条要发送的消息
        String message = "Hello, Kafka!";

        // 使用 KafkaTemplate 将消息发送到指定的 Kafka Topic
        kafkaTemplate.send("test", message);

        return "Message sent to Kafka: " + message;
    }
}

这里我们通过 KafkaTemplate 发送一条名为 "Hello, Kafka!" 的消息。使用 kafkaTemplate.send("test", message) 将其发送到名为 test 的 Kafka topic 中。

这里还有一些 Kafka Producer 的高级操作，比如异步发送和发送带有回调函数的消息，这里就不一一赘述了。

以上就是基于 SpringBoot 使用 Flink 收发 Kafka 消息的示例详解。