05、Flink实战：DataStream之DataSourceAPI(四）addSource用法-程序员快看-教程，程序员编程资料站

addSource可以实现读取第三方数据源的数据。

1、此处以读取kafka数据为例（kafka集群安装参考https://blog.csdn.net/zhuzuwei/article/details/107136796）；

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class KafkaSouceReview {
    public static void main(String[] args) throws Exception{
        // 1.获取flink流计算的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //Kafka props
        Properties properties = new Properties();
        //指定Kafka的Broker地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.***.***:9092,192.168.***.***:9092,192.168.***.***:9092");
        //指定组ID
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka_test_group1");
        //如果没有记录偏移量，第一次从最开始消费
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 2.从kafka读取数据
       FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("kafka_test1", new SimpleStringSchema(), properties);

        DataStreamSource<String> stringDataStreamSource = env.addSource(kafkaSource);

        //3.调用Sink
        stringDataStreamSource.print();

        //4.启动流计算
        env.execute("KafkaSouceReview");
    }
}

FlinkKafkaConsumer方法源码如下，第二个参数要传入DeSerializationSchema的实现类的实例。

public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props) {
    this(Collections.singletonList(topic), valueDeserializer, props);
}

DeSerializationSchema实现类如下，最常用的就是SimpleStringSchema.

2、启动zookeeper集群：依次在不同节点执行./zkServer.shstart；

3、启动kafka集群：依次在不同节点执行./kafka-server-start.sh-daemon/usr/local/kafka_2.11-2.1.1/config/server.properties&；

在后台启动kafka

4、创建topickafka_test1；

5、启动并写数据到kafkatopic；

6、启动flink的java程序，成功读取到数据；