如何在 Spring Batch 中排除作业参数的唯一性？

Question

我正在尝试在 Spring Batch 2 中启动作业，并且需要在作业参数中传递一些信息，但我不希望它计入作业实例的唯一性。例如，我希望这两组参数被认为是唯一的：

file=/my/file/path,session=1234
file=/my/file/path,session=5678

这个想法是，将有两个不同的服务器尝试启动相同的作业，但附加不同的会话。在这两种情况下我都需要该会话号。有任何想法吗？

谢谢！

Answer 1

因此，如果“文件”是唯一应该唯一的属性，并且下游代码使用“会话”，那么您的问题几乎与我遇到的问题完全匹配。我有一个 JMSCorrelationId，需要将其存储在执行上下文中以供以后使用，但我不希望它影响作业参数的唯一性。根据戴夫·赛尔（Dave Syer）的说法，这确实是不可能的，所以我采取了使用参数（不是您的情况下的“会话”）创建作业的路线，然后在任何实际运行之前将“会话”属性添加到执行上下文中。

这使我可以访问下游的“会话”，但它不在作业参数中，因此不会影响唯一性。

参考

https://jira.springsource.org/browse/BATCH-1412

http://forum.springsource.org/showthread.php?104440-Non-Identity-Job-Parameters&highlight=

您将从这个论坛中看到，没有好的方法可以做到这一点（根据 Dave Syer），但我基于 SimpleJobLauncher 编写了自己的启动器（事实上，如果调用非重载方法，我会委托给 SimpleLauncher），该启动器具有用于启动作业的重载方法，该方法采用回调接口，允许将参数贡献给执行上下文，但不是“真实”作业参数。你可以做一些非常类似的事情。

我认为适合您的 LOC 就在这里： jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);

    if (contributor != null) {
        if (contributor.contributeTo(jobExecution.getExecutionContext())) {
            jobRepository.updateExecutionContext(jobExecution);
        }
    }

这是在创建执行上下文之后添加执行上下文的地方。希望这对您的实施有所帮助。

public class ControlMJobLauncher implements JobLauncher, InitializingBean {

    private JobRepository jobRepository;
    private TaskExecutor taskExecutor;
    private SimpleJobLauncher simpleLauncher;
    private JobFilter jobFilter;

    public void setJobRepository(JobRepository jobRepository) {
        this.jobRepository = jobRepository;
    }

    public void setTaskExecutor(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    /**
     * Optional filter to prevent job launching based on some specific criteria.
     * Jobs that are filtered out will return success to ControlM, but will not run
     */
    public void setJobFilter(JobFilter jobFilter) {
        this.jobFilter = jobFilter;
    }

    public JobExecution run(final Job job, final JobParameters jobParameters, ExecutionContextContributor contributor)
            throws JobExecutionAlreadyRunningException, JobRestartException,
            JobInstanceAlreadyCompleteException, JobParametersInvalidException, JobFilteredException {

        Assert.notNull(job, "The Job must not be null.");
        Assert.notNull(jobParameters, "The JobParameters must not be null.");

        //See if job is filtered
        if(this.jobFilter != null && !jobFilter.launchJob(job, jobParameters)) {
            throw new JobFilteredException(String.format("Job has been filtered by the filter: %s", jobFilter.getFilterName()));
        }

        final JobExecution jobExecution;

        JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
        if (lastExecution != null) {
            if (!job.isRestartable()) {
                throw new JobRestartException("JobInstance already exists and is not restartable");
            }
            logger.info(String.format("Restarting job %s instance %d", job.getName(), lastExecution.getId()));
        }

        // Check the validity of the parameters before doing creating anything
        // in the repository...
        job.getJobParametersValidator().validate(jobParameters);

        /*
         * There is a very small probability that a non-restartable job can be
         * restarted, but only if another process or thread manages to launch
         * <i>and</i> fail a job execution for this instance between the last
         * assertion and the next method returning successfully.
         */
        jobExecution = jobRepository.createJobExecution(job.getName(),
                jobParameters);

        if (contributor != null) {
            if (contributor.contributeTo(jobExecution.getExecutionContext())) {
                jobRepository.updateExecutionContext(jobExecution);
            }
        }

        try {
            taskExecutor.execute(new Runnable() {

                public void run() {
                    try {
                        logger.info("Job: [" + job
                                + "] launched with the following parameters: ["
                                + jobParameters + "]");
                        job.execute(jobExecution);
                        logger.info("Job: ["
                                + job
                                + "] completed with the following parameters: ["
                                + jobParameters
                                + "] and the following status: ["
                                + jobExecution.getStatus() + "]");
                    } catch (Throwable t) {
                        logger.warn(
                                "Job: ["
                                        + job
                                        + "] failed unexpectedly and fatally with the following parameters: ["
                                        + jobParameters + "]", t);
                        rethrow(t);
                    }
                }

                private void rethrow(Throwable t) {
                    if (t instanceof RuntimeException) {
                        throw (RuntimeException) t;
                    } else if (t instanceof Error) {
                        throw (Error) t;
                    }
                    throw new IllegalStateException(t);
                }
            });
        } catch (TaskRejectedException e) {
            jobExecution.upgradeStatus(BatchStatus.FAILED);
            if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
                jobExecution.setExitStatus(ExitStatus.FAILED
                        .addExitDescription(e));
            }
            jobRepository.update(jobExecution);
        }

        return jobExecution;
    }

    static interface ExecutionContextContributor {
        boolean CONTRIBUTED_SOMETHING = true;
        boolean CONTRIBUTED_NOTHING = false;
        /**
         * 
         * @param executionContext
         * @return true if the exeuctioncontext was contributed to
         */
        public boolean contributeTo(ExecutionContext executionContext);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.state(jobRepository != null, "A JobRepository has not been set.");
        if (taskExecutor == null) {
            logger.info("No TaskExecutor has been set, defaulting to synchronous executor.");
            taskExecutor = new SyncTaskExecutor();
        }
        this.simpleLauncher = new SimpleJobLauncher();
        this.simpleLauncher.setJobRepository(jobRepository);
        this.simpleLauncher.setTaskExecutor(taskExecutor);
        this.simpleLauncher.afterPropertiesSet();
    }


    @Override
    public JobExecution run(Job job, JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException,
            JobInstanceAlreadyCompleteException, JobParametersInvalidException {
        return simpleLauncher.run(job, jobParameters);
    }

}