import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void afterJob() throws Exception {
    List<Long> activeJobs = BatchRuntime.getJobOperator().getRunningExecutions(BatchConstants.MATRIX_JOB);
    log.info("Finished matrix job. Remaining: {}", activeJobs);

    if (activeJobs.size() == 0) {
        // final job execution: save current time stamp of creation to db
        try {
            logRepository.insert(new Date());
            coreBootstrapper.computationFinished();

            log.info("Finished all matrix computations. All done!");
        } catch (DatabaseException e) {
            log.error("FATAL: Error inserting new computation log! {}", e.getMessage());
        }
    }
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void startCartesianMatrixJob(List<EsriPointFeature> sourcePoints, LegType legType) {
    if (checkNullOrEmpty(sourcePoints)) {
        log.error("Will not start Cartesian Matrix job with invalid sourcePoints for leg type {}", legType);
        return;
    }
    JobOperator jo = BatchRuntime.getJobOperator();

    Properties properties = new Properties();
    properties.setProperty(BatchConstants.PATH_TYPE, legType.toString());
    // assume here that the size of both sets is roughly equivalent!
    properties.setProperty(BatchConstants.BATCH_SIZE, String.valueOf(sourcePoints.size()));
    properties.setProperty(BatchConstants.PARTITIONING, String.valueOf(false));
    properties.setProperty(BatchConstants.CARTESIAN, String.valueOf(true));

    Long jobId = jo.start(BatchConstants.MATRIX_JOB, properties);
    log.info("Dispatched Cartesian Matrix job with id {}", jobId);
    sourcePointStash.put(jobId, sourcePoints);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void startIVMatrixJob(IVRequestTuple request, LegType legType) {
    if (checkNullOrEmpty(request.getTargets())) {
        log.error("Will not start IV Matrix job with invalid targets for leg type {}", legType);
        return;
    }
    JobOperator jo = BatchRuntime.getJobOperator();

    Properties properties = new Properties();
    properties.setProperty(BatchConstants.PATH_TYPE, legType.toString());
    // assume here that the size of both sets is roughly equivalent!
    properties.setProperty(BatchConstants.BATCH_SIZE, String.valueOf(request.getTargets().size()));
    properties.setProperty(BatchConstants.PARTITIONING, String.valueOf(true));
    properties.setProperty(BatchConstants.IVROUTE, String.valueOf(true));

    Long jobId = jo.start(BatchConstants.MATRIX_JOB, properties);
    List<EsriPointFeature> source = new ArrayList<>();
    source.add(request.getSource());
    log.info("Dispatched IV Matrix job with id {}", jobId);
    sourcePointStash.put(jobId, source);
    carTargetStash.put(jobId, new ArrayList<>(request.getTargets()));
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void startMatrixJob(List<IVRequestTuple> points, LegType legType) {
    if (checkNullOrEmpty(points)) {
        log.error("Will not start Matrix job with invalid points for leg type {}", legType);
        return;
    }
    JobOperator jo = BatchRuntime.getJobOperator();

    Properties properties = new Properties();
    properties.setProperty(BatchConstants.PATH_TYPE, legType.toString());
    properties.setProperty(BatchConstants.BATCH_SIZE, String.valueOf(points.size()));
    properties.setProperty(BatchConstants.PARTITIONING, String.valueOf(true));

    Long jobId = jo.start(BatchConstants.MATRIX_JOB, properties);
    log.info("Dispatched Matrix job with id {}", jobId);
    matrixStash.put(jobId, points);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void startNeighborsJob(List<RasterPoint> rasterPoints) {
    if (checkNullOrEmpty(rasterPoints)) {
        log.error("Will not start Neighbors job with invalid rasterPoints.");
        return;
    }
    JobOperator jo = BatchRuntime.getJobOperator();

    Properties properties = new Properties();
    properties.setProperty(BatchConstants.BATCH_SIZE, String.valueOf(rasterPoints.size()));
    properties.setProperty(BatchConstants.MAX_WALK, String.valueOf(maxWalk));

    Long jobId = jo.start(BatchConstants.NEIGHBORS_JOB, properties);
    log.info("Dispatched Neighbors job with id {}", jobId);
    rasterStash.put(jobId, rasterPoints);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
private void waitForCompletion(final long executionId, final int timeout, final TimeUnit unit) {
    final JobExecution jobExecution = BatchRuntime.getJobOperator().getJobExecution(executionId);
    long time = unit.toMillis(timeout);
    long sleep = 100;
    boolean complete = false;
    while (!complete && time > 0) {
        switch (jobExecution.getBatchStatus()) {
            case STARTED:
            case STARTING:
            case STOPPING:
                try {
                    TimeUnit.MILLISECONDS.sleep(sleep);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                time -= sleep;
                sleep = Math.max(sleep / 2, 100L);
                break;
            default:
                complete = true;
                break;
        }
    }
    Assert.assertTrue("Batch job did not complete withing allotted time.", complete);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
public <T> T doWorkWithJobOperator(JobOperatorWork<T> w, String applicationName) throws NamingException {
    if (operationMode == OperationMode.DIRECT) {
        return w.doWorkWithJobOperator(BatchRuntime.getJobOperator());
    }

    final String lookupName = getLookupName(applicationName);
    log.log(Level.FINE, "lookupName={0}, applicationName={1}", new Object[]{lookupName, applicationName});
    InitialContext ic = null;
    try {
        ic = new InitialContext();
        JobOperator jobOperator = (JobOperator) ic.lookup(lookupName);
        return w.doWorkWithJobOperator(jobOperator);
    } finally {
        if (ic != null) {
            ic.close();
        }
    }
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Override
public void open(Serializable checkpoint) throws Exception {
    String resourceName = BatchRuntime
            .getJobOperator()
            .getParameters(this.jobCtx.getExecutionId())
            .getProperty("resourceName");

    if (null == resourceName) {
        resourceName = this.resourceNameDefault;
    }

    if (null == resourceName) {
        throw new NullPointerException("'resourceName' is not set!");
    }

    LOGGER.info("Opening " + resourceName);

    this.reader = new BufferedReader(
            new InputStreamReader(
                    currentThread()
                            .getContextClassLoader()
                            .getResourceAsStream("META-INF/" + resourceName)));
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Test
public void testEndNormallyExecuteBoth() throws Exception {
	JobOperator jo = BatchRuntime.getJobOperator();
	Properties props = new Properties();
	props.put("forceFailure", "true");
	long exec1Id = jo.start("nextOnStep1Failure", props);
	Thread.sleep(normalSleepTime);
	JobExecution je = jo.getJobExecution(exec1Id);
	Map<String, StepExecution> steps = new HashMap<String,StepExecution>();
	for (StepExecution se : jo.getStepExecutions(exec1Id)) { steps.put(se.getStepName(), se); }
	
	assertEquals("Job batch status", BatchStatus.COMPLETED, je.getBatchStatus());
	assertEquals("exit status", "COMPLETED", je.getExitStatus());
	assertEquals("Steps executed", 2, steps.size());
	assertEquals("Step failed", BatchStatus.FAILED, steps.get("step1").getBatchStatus());
	assertEquals("Step failed", BatchStatus.COMPLETED, steps.get("step2").getBatchStatus());
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Test
public void testBatchDecision() throws Exception {
    JobOperator jobOperator = BatchRuntime.getJobOperator();
    Long executionId = jobOperator.start("myJob", new Properties());
    JobExecution jobExecution = jobOperator.getJobExecution(executionId);

    BatchTestHelper.keepTestAlive(jobExecution);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    List<String> executedSteps = new ArrayList<>();
    for (StepExecution stepExecution : stepExecutions) {
        executedSteps.add(stepExecution.getStepName());
    }

    assertEquals(2, stepExecutions.size());
    assertArrayEquals(new String[] {"step1", "step3"}, executedSteps.toArray());
    assertFalse(executedSteps.contains("step2"));
    assertEquals(BatchStatus.COMPLETED, jobExecution.getBatchStatus());
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Test
public void testBatchSplit() throws Exception {
    JobOperator jobOperator = BatchRuntime.getJobOperator();
    Long executionId = jobOperator.start("myJob", new Properties());
    JobExecution jobExecution = jobOperator.getJobExecution(executionId);

    BatchTestHelper.keepTestAlive(jobExecution);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    List<String> executedSteps = new ArrayList<>();
    for (StepExecution stepExecution : stepExecutions) {
        executedSteps.add(stepExecution.getStepName());
    }

    assertEquals(3, stepExecutions.size());
    assertTrue(executedSteps.contains("step1"));
    assertTrue(executedSteps.contains("step2"));
    assertTrue(executedSteps.contains("step3"));
    assertTrue(executedSteps.get(0).equals("step1") || executedSteps.get(0).equals("step2"));
    assertTrue(executedSteps.get(1).equals("step1") || executedSteps.get(1).equals("step2"));
    assertTrue(executedSteps.get(2).equals("step3"));
    assertEquals(BatchStatus.COMPLETED, jobExecution.getBatchStatus());
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
public void startBatches() throws JobSecurityException, JobStartException, NoSuchJobExecutionException {
      JobOperator jobOperator = BatchRuntime.getJobOperator();

      logger.log(Level.INFO, "starting simple");
      Long executionId = jobOperator.start("simple", new Properties());
      JobExecution jobExecution = jobOperator.getJobExecution(executionId);

      logger.log(Level.INFO, "Started simple job with id {0}", jobExecution.getExecutionId());
      logger.log(Level.INFO, "Status simple {0}", jobExecution.getBatchStatus());

//      logger.log(Level.INFO, "starting chunk");
//      Long chunkExecutionId = jobOperator.start("chunk", new Properties());
//      JobExecution chunkJobExecution = jobOperator.getJobExecution(chunkExecutionId);
//
//      logger.log(Level.INFO, "Status chunk {0}", chunkJobExecution.getBatchStatus());
//      logger.log(Level.INFO, "Started chunk job with id {0}", chunkJobExecution.getExecutionId());
   }
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@PostCreated
@PostRecovery
@PostConfig
public void setup()
{
    _batchDataConsumer = new BatchDataConsumer(this);
    _batchDataProvider = new BatchDataProvider(this);

    BatchDataConsumerMap.getBatchDataConsumerMap().add(_batchDataConsumer);
    BatchDataProviderMap.getBatchDataProviderMap().add(_batchDataProvider);

    JobOperator jobOperator = BatchRuntime.getJobOperator();

    Properties jobParameters = new Properties();
    jobParameters.setProperty(BatchDataConsumerMap.ID_PROPERTYNAME, _batchDataConsumer.getId());
    jobParameters.setProperty(BatchDataProviderMap.ID_PROPERTYNAME, _batchDataProvider.getId());

    long execId = jobOperator.start(_properties.get(JOBID_PROPERTYNAME), jobParameters);

    logger.log(Level.FINE, "BatchDataService: " + execId);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@PostCreated
@PostRecovery
@PostConfig
public void setup()
{
    _batchDataConsumer = new BatchDataConsumer(this);

    BatchDataConsumerMap.getBatchDataConsumerMap().add(_batchDataConsumer);

    JobOperator jobOperator = BatchRuntime.getJobOperator();

    Properties jobParameters = new Properties();
    jobParameters.setProperty(BatchDataConsumerMap.ID_PROPERTYNAME, _batchDataConsumer.getId());

    long execId = jobOperator.start(_properties.get(JOBID_PROPERTYNAME), jobParameters);

    logger.log(Level.FINE, "BatchDataService: " + execId);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@PostCreated
@PostRecovery
@PostConfig
public void setup()
{
    _batchDataConsumer = new BatchDataConsumer(this);
    _batchDataProvider = new BatchDataProvider(this);

    BatchDataConsumerMap.getBatchDataConsumerMap().add(_batchDataConsumer);
    BatchDataProviderMap.getBatchDataProviderMap().add(_batchDataProvider);

    JobOperator jobOperator = BatchRuntime.getJobOperator();

    Properties jobParameters = new Properties();
    jobParameters.setProperty(BatchDataConsumerMap.ID_PROPERTYNAME, _batchDataConsumer.getId());
    jobParameters.setProperty(BatchDataProviderMap.ID_PROPERTYNAME, _batchDataProvider.getId());

    long execId = jobOperator.start(_properties.get(JOBID_PROPERTYNAME), jobParameters);

    logger.log(Level.FINE, "BatchDataProcessor: " + execId);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@PostCreated
@PostRecovery
@PostConfig
public void setup()
{
    _batchDataProvider = new BatchDataProvider(this);

    BatchDataProviderMap.getBatchDataProviderMap().add(_batchDataProvider);

    JobOperator jobOperator = BatchRuntime.getJobOperator();

    Properties jobParameters = new Properties();
    jobParameters.setProperty(BatchDataProviderMap.ID_PROPERTYNAME, _batchDataProvider.getId());

    long execId = jobOperator.start(_properties.get(JOBID_PROPERTYNAME), jobParameters);

    logger.log(Level.FINE, "BatchDataSource: " + execId);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
public Object processItem(Object obj) throws Exception {
    Properties jobParameters = BatchRuntime.getJobOperator().getParameters(jobContext.getExecutionId());

    PayrollInputRecord inputRecord = (PayrollInputRecord) obj;
    PayrollRecord payrollRecord = new PayrollRecord();
    payrollRecord.setMonthYear((String) jobParameters.get("monthYear"));

    int base = inputRecord.getBaseSalary();
    float tax = base * 27 / 100.0f;
    float bonus = base * 15 / 100.0f;

    payrollRecord.setEmpID(inputRecord.getId());
    payrollRecord.setBase(base);
    payrollRecord.setTax(tax);
    payrollRecord.setBonus(bonus);
    payrollRecord.setNet(base + bonus - tax);
    
    return payrollRecord;
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
public Object processItem(Object obj) throws Exception {
    Properties jobParameters = BatchRuntime.getJobOperator().getParameters(jobContext.getExecutionId());

    PayrollInputRecord inputRecord = (PayrollInputRecord) obj;
    PayrollRecord payrollRecord = new PayrollRecord();
    payrollRecord.setMonthYear((String) jobParameters.get("Month-Year"));

    int base = inputRecord.getBaseSalary();
    float tax = base * 27 / 100.0f;
    float bonus = base * 15 / 100.0f;

    payrollRecord.setEmpID(inputRecord.getId());
    payrollRecord.setBase(base);
    payrollRecord.setTax(tax);
    payrollRecord.setBonus(bonus);
    payrollRecord.setNet(base + bonus - tax);
    
    return payrollRecord;
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Test
public void testRolledBackDuringWork() {
    final JobOperator jobOperator = BatchRuntime.getJobOperator();
    long executionId = jobOperator.start("txtest1", null);
    BatchStatus batchStatus = Batches.waitFor(jobOperator, executionId);
    Assert.assertEquals(batchStatus, BatchStatus.FAILED);
    Assert.assertEquals(TxErrorWriter1.written.intValue(), 3);

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    Assert.assertEquals(stepExecutions.size(), 1);
    StepExecution stepExecution = stepExecutions.get(0);
    Metric[] metrics = stepExecution.getMetrics();
    assertMetric(Metric.MetricType.READ_COUNT, 2, metrics);
    assertMetric(Metric.MetricType.WRITE_COUNT, 2, metrics);
    assertMetric(Metric.MetricType.ROLLBACK_COUNT, 1, metrics);
}
 

import javax.batch.runtime.BatchRuntime; //导入依赖的package包/类
@Test
public void testDeciderRestart() {

    JobOperator jobOperator = BatchRuntime.getJobOperator();
    long executionId = jobOperator.start("decider-test", new Properties());

    BatchStatus batchStatus = Batches.waitFor(jobOperator, executionId);
    assertEquals(batchStatus, BatchStatus.STOPPED);
    assertEquals(jobOperator.getJobExecution(executionId).getExitStatus(), "decider-stop");

    List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
    assertEquals(stepExecutions.size(), 1);

    long restartExecutionId = jobOperator.restart(executionId, new Properties());

    BatchStatus restartStatus = Batches.waitFor(jobOperator, restartExecutionId);
    assertEquals(restartStatus, BatchStatus.COMPLETED);

    String exitStatus = jobOperator.getJobExecution(restartExecutionId).getExitStatus();
    assertEquals("COMPLETED", exitStatus);

    List<StepExecution> restartExecutions = jobOperator.getStepExecutions(restartExecutionId);
    assertEquals(restartExecutions.size(), 1);
    assertEquals(restartExecutions.get(0).getStepName(), "executeOnRestart");
}