1、代码逻辑实现

package day03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @program: Flink_learn
 * @description: 偶数子任务索引
 * @author: Mr.逗
 * @create: 2021-09-17 14:32
 **/
public class RichSourceDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> source = env.addSource(new RichParallelSourceFunction<Integer>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("生命周期开始，子任务索引是：" + getRuntimeContext().getIndexOfThisSubtask());
            }

            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                for (int i = 0; i < 10; i++) {
                    if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                        ctx.collect(i);
                    }
                }
            }

            @Override
            public void cancel() {
            }
        });
        source.setParallelism(1).print().setParallelism(2);
        String name = RichSourceDemo.class.getName();
        env.execute(name);
    }
}

复制

2、结果之展示

生命周期开始，子任务索引是：0
1> 2
2> 0
1> 6
2> 4
2> 8


复制