Flink自定义source、自定义sink

public class TestSource extends RichSourceFunction {
    private volatile boolean isRunning = true;
    @Override
    public void open(Configuration parameters) throws Exception {
    }
    @Override
    public void run(SourceContext sourceContext) throws Exception {
        Random random = new Random();
        while (isRunning) {
            Thread.sleep((getRuntimeContext().getIndexOfThisSubtask() + 1) * 1000 * 5);
            //String key = "类别" + (char) ('A' + random.nextInt(3));
            int sku = random.nextInt(10);
            int value = random.nextInt(10) + 1;
            String str = sku + "," + value;
            System.out.println(str);
            sourceContext.collect(str);
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }
}