flink怎么读取kafka多个topic

要在Flink中读取多个Kafka topic，可以使用Flink Kafka Consumer来实现。以下是一个示例代码，演示如何读取多个Kafka topic：

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.Arrays;
import java.util.List;
import java.util.Properties;

public class ReadMultipleKafkaTopics {

    public static void main(String[] args) throws Exception {
        
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 设置Kafka相关配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        
        // 定义要读取的Kafka topic列表
        List<String> topics = Arrays.asList("topic1", "topic2", "topic3");
        
        // 创建Flink Kafka Consumer
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
        
        // 从Kafka读取数据
        DataStream<String> kafkaDataStream = env.addSource(kafkaConsumer);
        
        // 对从Kafka读取的数据进行处理
        kafkaDataStream.print();
        
        // 执行作业
        env.execute("ReadMultipleKafkaTopics");
    }
}

在上面的代码中，我们首先创建了一个Flink的执行环境（StreamExecutionEnvironment），然后设置了Kafka的相关配置，包括Kafka的地址和要读取的Kafka topic列表。接着创建了一个Flink Kafka Consumer，并指定要读取的topic列表、序列化方式（这里使用SimpleStringSchema）和Kafka的配置。最后通过env.addSource()方法将Kafka Consumer添加到Flink的执行环境中，并对从Kafka读取的数据进行处理。最后调用env.execute()方法执行作业。

通过这种方式，我们可以轻松地在Flink中读取多个Kafka topic，并对数据进行处理。