Java 实现flink实时处理kafka

你可以使用 Java 编程语言来实现 Flink 实时处理 Kafka。下面是一个简单的示例代码：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class FlinkKafkaExample {
    public static void main(String[] args) throws Exception {
        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置 Kafka 连接参数
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-consumer");

        // 创建 Kafka 数据流
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
        DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

        // 在数据流上进行处理逻辑
        DataStream<String> processedStream = kafkaStream.map(str -> "Processed: " + str);

        // 打印处理后的结果
        processedStream.print();

        // 执行任务
        env.execute("Flink Kafka Example");
    }
}

在上述代码中，我们使用 `FlinkKafkaConsumer` 连接到 Kafka 主题，将 Kafka 中的数据流添加到 Flink 的执行环境中。然后，我们对数据流进行处理，并将处理后的结果打印出来。最后，通过调用 `env.execute()` 来执行任务。

请确保在运行代码之前，您已经正确配置了 Kafka 的连接参数，并将相关的 Flink 和 Kafka 依赖项添加到您的项目中。