Flink自定义source function用线程池产生数据流

可以使用 Java 中的线程池来实现 Flink 的自定义 source function 产生数据流。具体步骤如下：

1. 实现一个实现了 Runnable 接口的数据生成类，该类可以在 run 方法中产生数据，并将数据发送到 Flink 的 source context 中。

public class DataSource implements Runnable {
    private final SourceContext<String> context;
    private final int delay;

    public DataSource(SourceContext<String> context, int delay) {
        this.context = context;
        this.delay = delay;
    }

    @Override
    public void run() {
        while (true) {
            // 产生数据
            String data = "data-" + UUID.randomUUID().toString();
            // 发送数据到 Flink 的 source context 中
            context.collect(data);
            // 休眠指定时间
            try {
                Thread.sleep(delay);
            } catch (InterruptedException e) {
                // ignore
            }
        }
    }
}

2. 在自定义 source function 中创建线程池，并在 open 方法中启动线程池，每个线程都运行数据生成类的实例。

public class ThreadPoolSourceFunction extends RichSourceFunction<String> {
    private transient ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 创建线程池
        executorService = Executors.newFixedThreadPool(5);
        // 启动线程池
        for (int i = 0; i < 5; i++) {
            executorService.submit(new DataSource(getSourceContext(), 1000));
        }
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        // do nothing
    }

    @Override
    public void cancel() {
        // do nothing
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭线程池
        executorService.shutdownNow();
    }
}

在上面的例子中，我们创建了一个固定大小为 5 的线程池，并将数据生成类的实例提交到线程池中运行。每个数据生成类实例都会在 run 方法中产生数据，并将数据发送到 Flink 的 source context 中。由于每个数据生成类实例都是在一个独立的线程中运行，因此可以实现并行产生数据的效果。