Java并发和循环任务执行

Question

我有一个包含 20 个不同步骤的任务:

public class MyTask implements Runnable {
    @Override
    public void run() {
        // Step #1. Fetch data
        Data data = fetchData();

        // Step #2. Do something with data
        SomeResult result = dataProcessor.process(data);

        // ...

        // Step #19. Generate report
        Report report = reportFactory.newReport(result);

        // Step #20. Clean up
        cleanUp();
    }
}

我想要:

• 在任意时间(应用程序启动)开始运行任务，我们将调用 STARTING_POINT。
• 每分钟从 STARTING_POINT 开始一遍又一遍地运行该任务(Timer？ScheduledExecutorService？)。
• 如果 1 分钟周期结束时任务仍在运行(例如，在步骤 #17)，则中断该任务并阻止其继续执行，然后(再次因为我们正在开始一个新周期)每分钟)，重新开始一个新的周期(因此从头开始再次调用 run())。
• 否则，如果任务在 1 分钟周期结束之前完成，则将 STARTING_POINT 重置为当前时间(System.currentTimeInMillis()？)并开始循环再次分钟，但从新的 STARTING_POINT 开始。
• 一旦这些 1 分钟的周期开始，它们就不会停止，除非被可能从 ShutdownHook 或类似的东西调用的外部“控制”线程中断。

例如，假设应用启动，STARTING_POINT 是美国东部时间 2014 年 2 月 4 日 11:15:36。假设第一个周期需要正好 1 分钟来完成 run() 内的所有 20 个步骤的执行。然后它将在 11:15:36 到 11:16:36 之间执行，然后第二个周期将重新开始并调用 run()。假设这个周期很慢，当我们到达 11:17:36 时，它仅处于步骤#11。它将关闭并停止执行，然后第三个周期将重新开始并调用 run()。如果这个周期很快，并且在 11:17:53 之前通过 run() 全部完成，那么 STARTING_POINT 将变为 2/4/2014 11:17:53，并且第四个周期将开始执行并再次调用 run()。等等

我完全不知道从哪里开始深入研究这个。有任何想法吗？我愿意使用 JDK 附带的任何内容以及除 Apache Camel 之外的任何开源库。

Answer 1

看起来像Spring Batch将是一个值得考虑的好选择。我会推荐它。

您可以使用 Spring 集成 quartz 调度程序来安排作业。一个作业会被分成多个步骤(也可以实现并行性)。工作可以重新开始，最重要的是，一切都在那里，您的工作量最少。

大部分内容都需要正确接线。我提供了一个 bean 配置作为示例。

    <bean id="threadPoolExecutor"
    class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="20" />
    <property name="maxPoolSize" value="25" />
    <property name="queueCapacity" value="500" />
</bean>
<bean id="jobLauncher"
    class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository" />
    <property name="taskExecutor" ref="asyncTaskExecutor" />
</bean>

<batch:job id="batchjob" job-repository="jobRepository">
    <batch:step id="step1" next="step2">
        <batch:tasklet ref="start" />
    </batch:step>
    <batch:step id="step2" next="step3.master">
        <batch:tasklet ref="fetchData" />
    </batch:step>
    <batch:step id="step3.master">
        <batch:partition step="step3.slave"
            partitioner="processingPartitioner">
            <batch:handler grid-size="20" task-executor="threadPoolExecutor" />
        </batch:partition>
    </batch:step>
    <batch:listeners>
        <batch:listener ref="MyListener" />
    </batch:listeners>
</batch:job>

<batch:step id="step3.slave">
    <batch:tasklet ref="processFile" />
</batch:step>

<bean id="myLauncher" class="com.<>.batch.MyLauncher">
    <property name="job" ref="batchJob" />
    <property name="jobLauncher" ref="jobLauncher" />
</bean>

<task:scheduled-tasks>
    <task:scheduled ref="myLauncher" method="launch"
        cron="0 0 0 0 0 1" />
</task:scheduled-tasks>

每个步骤都可以定义并且应该实现Tasklet。 spring本身提供了一些例子。