Flink操练（二十一）之自定义数据源

　　1、代码实现逻辑package day02;  import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.source.SourceFunction;  import java.sql.Timestamp; import java.util.Calendar; import java.util.Random;  /**  * @program: Flink_learn  * @description: 自定义数据源  * @author: Mr.逗  * @create: 2021-09-14 15:58  **/ public class SelfSource {     public static void main(String[] args) {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         DataStreamSource streamSource = env.addSource(new MySource());         streamSource.print();         try {             String name = SelfSource.class.getName();             env.execute(name);         }catch (Exception e)         {             e.printStackTrace();         }     }     /**     * @Description: 自定义数据源     * @Param:      * @return:      * @Author: Mr.逗     * @Date: 2021/9/17     */     public static class MySource implements SourceFunction     {         private boolean running = true;         private String[] userArr = {＂Mary＂, ＂Bob＂, ＂Alice＂, ＂Liz＂};         private String[] urlArr = {＂./home＂, ＂./cart＂, ＂./fav＂, ＂./prod?id=1＂, ＂./prod?id=2＂};         private Random random = new Random();         @Override         public void run(SourceContext out) throws Exception {             while (running)             {// collect方法，向下游发送数据                 out.collect(                         new Event(                                 userArr[random.nextInt(userArr.length)],                                 urlArr[random.nextInt(urlArr.length)],                                 Calendar.getInstance().getTimeInMillis()                         )                 );                 Thread.sleep(1000);             }         }         @Override         public void cancel() {             running=false;         }     }     public static class Event {         public String user;         public String url;         public Long timestamp;          public Event() {         }          public Event(String user, String url, Long timestamp) {             this.user = user;             this.url = url;             this.timestamp = timestamp;         }          @Override         public String toString() {             return ＂Event{＂ +                     ＂user=＂＂ + user + ＂＂＂ +                     ＂, url=＂＂ + url + ＂＂＂ +                     ＂, timestamp=＂ + new Timestamp(timestamp) +                     ＂}＂;         }     } } 2、结果展示Event{user=＂Alice＂, url=＂./fav＂, timestamp=2021-09-14 16:08:30.346} Event{user=＂Bob＂, url=＂./cart＂, timestamp=2021-09-14 16:08:31.362} Event{user=＂Bob＂, url=＂./home＂, timestamp=2021-09-14 16:08:32.373} Event{user=＂Mary＂, url=＂./prod?id=1＂, timestamp=2021-09-14 16:08:33.384} Event{user=＂Liz＂, url=＂./cart＂, timestamp=2021-09-14 16:08:34.395} Event{user=＂Bob＂, url=＂./prod?id=1＂, timestamp=2021-09-14 16:08:35.412} Event{user=＂Alice＂, url=＂./prod?id=1＂, timestamp=2021-09-14 16:08:36.424}