设置并发限制后如何立即取回作业ID

Question

问题是，我设置并发限制后（例如限制为2），当第3个请求进来时，作业启动器不会返回作业执行（作业启动器正在阻塞新的作业执行）。换句话说，我想通过运行“runFun()”方法在收到请求后立即获取作业 ID。

我尝试了 SimpleAsyncTaskExecutor 和 SimpleThreadPoolTask​​Executor，它们都不起作用。并且由于挂起的“jobLauncher.run()”方法，之前的作业监听器也没有作业ID信息。我浏览了源代码（SimpleAsyncTaskExecutor.execute()、ConcurrencyThrottleSupport.beforeAccess()），没有运气解决这个问题。

在这种情况下，当作业启动器运行执行时，如何立即取回作业 ID？

我将主要代码发布如下：

**//Config:**

<bean id="jobLauncher"
      class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository"/>
    <property name="taskExecutor">
        <bean class="org.springframework.core.task.SimpleAsyncTaskExecutor">
            <property name="concurrencyLimit" value="2"/>
        </bean>
    </property>
</bean>

<bean id="jobExplorer" class="org.springframework.batch.core.explore.support.JobExplorerFactoryBean">
    <property name="dataSource" ref="sampleservice.persist.datasource"/>
</bean>

<bean id="jobOperator" class="org.springframework.batch.core.launch.support.SimpleJobOperator">
    <property name="jobExplorer" ref="jobExplorer"/>
    <property name="jobRepository" ref="jobRepository"/>
    <property name="jobLauncher" ref="jobLauncher"/>
    <property name="jobRegistry" ref="jobRegistry"/>
</bean>

<bean id="jobRegistry" class="org.springframework.batch.core.configuration.support.MapJobRegistry"></bean>


**//Code:**

    Private Long runFun() {
    new JobParametersBuilder()
                    .addLong("pId", parameterDO.getPId())
                    .toJobParameters();

                try {
                    JobExecution jobExecution = jobLauncher.run(bulkDownloadJob, params);

    //the job execution id will be blocked with status "STARTING" in the batch metadata table until the job is actually started with status "STARTED"
                    return jobExecution.getId();  
    } catch ...

}

经过思考，我正在尝试创建一个新的作业启动器，它会创建一个新线程来专门执行作业并立即返回 jobExecution：

//NewLauncherClass
public JobExecution run(final Job job, final JobParameters jobParameters) {

    ... 

    jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);

        try {
            return jobExecution;
        } finally {
            if (jobExecution != null) {
                new Thread(new Runnable() {
                    public void run() {
                        executeJob(jobExecution, job, jobParameters);
                    }
                }).start();
            } else {
                logger.error("*****");
            }
        }

    }

    //added this method for executing the job
    private void executeJob(final JobExecution jobExecution, final Job job, final JobParameters jobParameters) {
        try {
            logger.info("***** test bulk *****  Job " + jobExecution.getId() + " is going to be executed");

            taskExecutor.execute(new Runnable() {

                ...

Answer 1

您尝试执行的操作不适用于SimpleJobLauncher。问题是TaskExecutor 在使用限制时将阻塞，直到有线程可用于执行任务。问题是您必须有一种方法可以在线程可用于执行它之前返回 JobExecution。

你必须做你正在尝试的消息传递。当前基于消息的组件不会执行此操作，但您可以创建一个 JobLauncher 来创建 JobExecution 并将其发送到 MessageHandler 以执行它。这将允许调用者返回 JobExecution 并且仍然等待执行作业。