任务流之间的调度依赖

检查任务

import java.sql.SQLException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 每一个有依赖其他flow的flow，都要先进行依赖的检查
 * 这是一个切面
 */
public class FlowCheck {


    private static String flow_id = "order";

    public static void main(String[] args) {

        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);

        final MysqlDao azkabanDao = new MysqlDao();

        if (System.getProperties().getProperty("os.name").contains("Windows")) {

        } else {
            flow_id = args[0];
        }

        /**
         * 坑，会吞掉异常
         */
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {


                try {
                    MysqlDao.AzkabanExecInfo info = azkabanDao.getAzkabanJobExecInfo(flow_id);

                    /**
                     * {"ok":false,"reason":"submitDay is 2022-09-29 and today is 2022-09-30","status":"50"}
                     */
                    String output = JsonUtil.getJsonStrFromObj(info);

                    if (info.getOk()) {

                        System.out.println(output);

                        service.shutdown();
                    } else {

                        if ("30".equals(info.getStatus())) {
                            //wait
                        } else {
                            System.out.println(output);
                            service.shutdown();
                        }

                    }


                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }


            }
        }, 0, 5, TimeUnit.SECONDS);

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

jobA

#!/bin/bash
echo "do JobA"



r=`java -cp ./scheduler-1.0-SNAPSHOT-jar-with-dependencies.jar com.mz.scheduler.FlowCheck common_process`



echo 'result is:'$r

echo ${r} >> "${JOB_OUTPUT_PROP_FILE}"
1
2
3
4
5
6
7
8
9
10
11
12

jobB

新的flow开启的第一个job

#!/bin/bash
echo "do JobB"

1
2
3
4

定义flow

jobA和jobB任务依赖文件

nodes:
  - name: jobA
    type: command
    config:
      command: sh jobA.sh


  - name: jobB
    type: command
    dependsOn:
      - jobA
    config:
      command: sh jobB.sh
    condition: ${jobA:ok} == "true"
1
2
3
4
5
6
7
8
9
10
11
12
13
14

打包

这种情况下的打包需要将jar包一同打入zip
因为JobA引用了java处理的逻辑