只要自定一个Source类实现SourceFunction接口，一个Sink类实现SinkFunction接口，就能正常使用自定义的Source和Sink，或者直接extends继承RichSourceFunction和RichSinkFunction,RichSinkFunction:多个open和close方法

1、自定义Source

import org.apache.flink.streaming.api.datastream.DataStream;
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;public class Demo3SourceFunction {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//使用自定义sourceDataStream<Integer> myDS = env.addSource(new MySource());myDS.print();env.execute();}
}//自定义source
//实现SourceFunction接口
class MySource implements SourceFunction<Integer> {//在run方法中读取外部的数据，使用原生java代码@Overridepublic void run(SourceContext<Integer> ctx) throws Exception {while (true) {ctx.collect(1);Thread.sleep(1000);}}//cancel方法是任务被取消是执行的，用于回收资源@Overridepublic void cancel() {}
}

2、自定义Sink

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;public class Demo2MySink {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> linesDS = env.socketTextStream("master", 8888);linesDS.addSink(new MySink());env.execute();}
}//自定义Sink
class MySink implements SinkFunction<String> {//每一条数据执行一次@Overridepublic void invoke(String value, Context context) throws Exception {System.out.println("mySink:" + value);}
}