代码实现

public class SubmitFlinkJob {
    public static void main(String[] args) throws Exception {
        // jar包地址
        String jarFilePath = "C:\\Users\\Desktop\\Flink_jars\\Job.jar";
        // 集群 host
        String clusterHost = "cdh100";
        // 集群 post
        int clusterPort = 21414;
        // job 类路径
        List<String> names = new ArrayList<>();
        names.add("com.realtime.CheckJob");
        // 任务启动列表
        List<String> success = new ArrayList<>();
        List<String> failed = new ArrayList<>();
        for (String name : names) {
            boolean status = submit(jarFilePath, clusterHost, clusterPort, name);
            // 统计任务情况
            if (status) {
                success.add(name);
            } else {
                failed.add(name);
            }
        }
        System.out.println("启动成功的任务列表！！");
        success.forEach(System.out::println);
        System.out.println();
        System.out.println("启动失败的任务列表！！");
        failed.forEach(System.out::println);
    }
    private static boolean submit(String jarFilePath, String clusterHost, int clusterPort, String jobClassName) {
        RestClusterClient<StandaloneClusterId> client = null;
        try {
            // 集群信息
            Configuration configuration = new Configuration();
            configuration.setString(JobManagerOptions.ADDRESS, clusterHost);
            configuration.setInteger(JobManagerOptions.PORT, 6123);
            configuration.setInteger(RestOptions.PORT, clusterPort);
            client = new RestClusterClient<>(configuration, StandaloneClusterId.getInstance());
            int parallelism = 1;
            File jarFile = new File(jarFilePath);
            SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.none();
            PackagedProgram program = PackagedProgram.newBuilder()
                    .setConfiguration(configuration)
                    .setEntryPointClassName(jobClassName)
                    .setJarFile(jarFile)
                    .setSavepointRestoreSettings(savepointRestoreSettings).build();
            JobGraph jobGraph = PackagedProgramUtils.createJobGraph(program, configuration, parallelism, false);
            CompletableFuture<JobID> result = client.submitJob(jobGraph);
            result.get();
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}