SchedulerX:Java jobs

Basic job

A minimal JavaProcessor that logs a message and returns success:

@Component
public class MyProcessor1 extends JavaProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("Hello, schedulerx2.0!");
        return new ProcessResult(true);
    }
}

Terminable job

A JavaProcessor that supports graceful shutdown through the kill method. When SchedulerX calls kill, the stop flag breaks the processing loop.

@Component
public class MyProcessor2 extends JavaProcessor {
    private volatile boolean stop = false;

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        int N = 10000;
        while (!stop && N >= 0) {
            //TODO
            N--;
        }
        return new ProcessResult(true);
    }

    @Override
    public void kill(JobContext context) {
        stop = true;
    }

    @Override
    public void preProcess(JobContext context) {
        // Reset the flag for Singleton beans -- without this,
        // a previous kill signal would block the next run.
        stop = false;
    }
}

Distributed batch processing with the Map model

The following MapJobProcessor scans a database table in parallel:

1. The root task queries the min and max IDs from the target table.

2. It splits the ID range into pages and calls map(taskList, "Level1Dispatch") to distribute them across workers.

3. Each worker receives a PageTask with a specific ID range to process.

4. After every page task finishes, SchedulerX calls postProcess for final aggregation or cleanup.

@Component
public class ScanSingleTableJobProcessor extends MapJobProcessor {
    private static final int pageSize = 100;

    static class PageTask {
        private int startId;
        private int endId;

        public PageTask(int startId, int endId) {
            this.startId = startId;
            this.endId = endId;
        }

        public int getStartId() {
            return startId;
        }

        public int getEndId() {
            return endId;
        }
    }

    @Override
    public ProcessResult process(JobContext context) {
        String taskName = context.getTaskName();
        Object task = context.getTask();

        if (isRootTask(context)) {
            System.out.println("start root task");
            Pair<Integer, Integer> idPair = queryMinAndMaxId();
            int minId = idPair.getFirst();
            int maxId = idPair.getSecond();

            List<PageTask> taskList = Lists.newArrayList();
            int step = (int) ((maxId - minId) / pageSize); //Calculate the number of pages.
            for (int i = minId; i < maxId; i += step) {
                taskList.add(new PageTask(i, (i + step > maxId ? maxId : i + step)));
            }
            return map(taskList, "Level1Dispatch"); //The process calls the map method to distribute tasks.

        } else if (taskName.equals("Level1Dispatch")) {
            PageTask record = (PageTask) task;
            long startId = record.getStartId();
            long endId = record.getEndId();
            //TODO
            return new ProcessResult(true);
        }

        return new ProcessResult(true);
    }

    @Override
    public void postProcess(JobContext context) {
        //TODO
        System.out.println("all tasks is finished.");
    }

    private Pair<Integer, Integer> queryMinAndMaxId() {
        //TODO select min(id),max(id) from xxx
        return null;
    }
}