How to manage Java jobs - SchedulerX - Centro de documentación de Alibaba Cloud

Es posible ejecutar trabajos Java en los procesos de su aplicación. Este tema describe cómo gestionar trabajos Java.

Modos de ejecución

Los trabajos Java admiten los siguientes modos de ejecución:

Standalone: El trabajo se ejecuta en una máquina aleatoria dentro del mismo groupId.
Broadcast: El trabajo se ejecuta de forma concurrente en todas las máquinas del mismo groupId.
Visual MapReduce: Modelo de tarea MapReduce que requiere la Professional Edition. Admite hasta 1,000 subtareas. Permite consultar registros detallados de ejecución, registros operativos y pilas de subtareas mediante una palabra clave de negocio.
MapReduce: Modelo estándar de tarea MapReduce que admite el procesamiento paralelo de múltiples subtareas. Solo es posible consultar información resumida sobre la ejecución de las subtareas. Seleccione este modo cuando el número de subtareas sea menor a un millón.
Sharded execution: Este modo utiliza fragmentación estática y procesamiento por lotes dinámico para procesar grandes volúmenes de datos.

Implemente JavaProcessor para los modos Standalone y Broadcast. Implemente MapJobProcessor para los modos Visual MapReduce, MapReduce y Sharded execution.

La ruta de clase del procesador corresponde al nombre completo de la clase de implementación. Por ejemplo, com.apache.armon.test.schedulerx.processor.MySimpleJob.

Si no carga un paquete JAR, SchedulerX busca la clase de implementación del procesador en la ruta de clases de su aplicación. Por lo tanto, debe recompilar y publicar la aplicación después de cada modificación.
Si carga un paquete JAR, SchedulerX recarga en caliente tanto el paquete JAR como el procesador. No es necesario volver a publicar la aplicación.

Modelos de programación

Los trabajos Java admiten dos modelos de programación: JavaProcessor y MapJobProcessor.

JavaProcessor
- Opcional: public void preProcess(JobContext context) throws Exception
- Obligatorio: public ProcessResult process(JobContext context) throws Exception
- Opcional: public void postProcess(JobContext context)
- Opcional: public void kill(JobContext context)
MapJobProcessor
- Obligatorio: public ProcessResult process(JobContext context) throws Exception
- Opcional: public void postProcess(JobContext context)
- Opcional: public void kill(JobContext context)
- Obligatorio: public ProcessResult map(List<? extends Object> taskList, String taskName)

ProcessResult

Cada proceso debe devolver un objeto ProcessResult. Este objeto indica el estado de ejecución de la tarea, el resultado y los mensajes de error.

La tarea se ejecuta correctamente y devuelve new ProcessResult(true).
Una tarea falla si devuelve new ProcessResult(false, ErrorMsg) o si lanza una excepción.
Una ejecución exitosa de la tarea devuelve return new ProcessResult(true, result), donde result es una cadena que no puede exceder los 1000 bytes.

Ejemplo de trabajo HelloSchedulerx2.0

@Component
public class MyProcessor1 extends JavaProcessor {

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

Ejemplo de trabajo compatible con la función finalizar

@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) {
        stop = false;  // If the job is launched using Spring and the bean is a singleton, use preProcess to reset the flag.
    }
}

Ejemplo de trabajo para procesamiento por lotes con el modelo Map

/**
 * Distributed batch processing for a non-partitioned table.
 * 1. The root task queries a table to get minId and maxId.
 * 2. Construct PageTask and distribute it using the map method.
 * 3. If the next level receives a PageTask, process the data.
 *
 */
@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 subtasks.
        } 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 are finished.");
    }

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

}