提前检测到 AsyncListItemWriter 步骤结束

Question

我已经编写了 ItemWriter 的异步版本来异步写入我的项目：

public class AsyncListItemWriter<T> implements ItemStreamWriter<T>, InitializingBean {

    private ItemWriter<T> delegate;

    private TaskExecutor taskExecutor = new SyncTaskExecutor();

    public void afterPropertiesSet() throws Exception {
        Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
    }

    public void setDelegate(ItemWriter<T> delegate) {
        this.delegate = delegate;
    }

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

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).open(executionContext);
        }
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).update(executionContext);
        }
    }

    @Override
    public void close() throws ItemStreamException {
        if (delegate instanceof ItemStream) {
            ((ItemStream) delegate).close();
        }
    }

    @Override
    public void write(List<? extends T> items)  {
        StepExecution stepExecution = getStepExecution();
        taskExecutor.execute(() -> {
            if (stepExecution != null) {
                StepSynchronizationManager.register(stepExecution);
            }
            try {
                delegate.write(items);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (stepExecution != null) {
                    StepSynchronizationManager.close();
                }
            }
        });
    }

    private StepExecution getStepExecution() {
        StepContext context = StepSynchronizationManager.getContext();
        if (context == null) {
            return null;
        }
        StepExecution stepExecution = context.getStepExecution();
        return stepExecution;
    }
}

配置：

    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(64);
        executor.setMaxPoolSize(64);
        executor.setQueueCapacity(64);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setThreadNamePrefix("MultiThreaded-");
        return executor;
    }

    @Bean
    public ItemWriter<STModel> writer(){
        return items -> {
            Thread.sleep(1000);
            System.out.println("Writing...");
            for(STModel c : items) {
                System.out.println("######### Writer : ------> " + c + " inside size : " + c.relation.size() + ", On : " +Thread.currentThread().getName());
            }
        };
    }

    @Bean
    public AsyncListItemWriter<STModel> asyncWriter() throws Exception {
        AsyncListItemWriter<STModel> asyncItemWriter = new AsyncListItemWriter<>();
        asyncItemWriter.setDelegate(writer());
        asyncItemWriter.setTaskExecutor(taskExecutor());
        asyncItemWriter.afterPropertiesSet();
        return asyncItemWriter;
    }

    @Bean
    public Step sampleStep() throws Exception{
        return stepBuilderFactory.get("processingStep")
                .<STModel, STModel>chunk(10)
                .reader(itpReader())
                .writer(asyncWriter())
                .build();
    }

    @Bean
    public Job job() throws Exception{
        return jobBuilderFactory.get("job")
                .start(sampleStep())
                .build();
    }

在阅读完所有文件后，我收到了这个日志（文件在主线程上读取）：

同时读取 ---->STModel(tripId=109138356-1_459178)
同读时---->STModel(tripId=109138356-1_459178)
虽然相同读取 ---->null
读取---->null，打开：main
2021-03-16 10:38:48.409 INFO 17042 --- [main] os.batch.core.step.AbstractStep：步骤：[processingStep] 在 337 毫秒内执行
2021-03-16 10:38:48.412 INFO 17042 --- [main] o.s.b.c.l.support.SimpleJobLauncher：作业：[SimpleJob：[name=job]] 已完成，参数如下：[{}] 和以下状态：[已完成] 在 349 毫秒内
写作...
######### 编写器：------> STModel(tripId=109138355-1_459164) 内部大小：2，开启：MultiThreaded-1
######### 编写者：------> STModel(tripId=109138355-1_459165) 内部大小：1，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=109138355-1_459166) 内部大小：1，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=109138355-1_459167) 内部大小：1，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=109138355-1_459168) 内部大小：2，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=113507833-1_38959) 内部尺寸：23，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=113507835-1_38960) 内部尺寸：23，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=113507852-1_38961) 内部尺寸：23，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=113507863-1_38962) 内部尺寸：23，开启：MultiThreaded-1
######### 编写器：------> STModel(tripId=113507871-1_38963) 内部尺寸：23，开启：MultiThreaded-1
写作...
######### 编写器：------> STModel(tripId=113507882-1_38964) 内部大小：23，开启：MultiThreaded-2
######### 编写器：------> STModel(tripId=113507890-1_38965) 内部尺寸：23，开启：MultiThreaded-2
######### 编写器：------> STModel(tripId=113507900-1_38966) 内部尺寸：23，开启：MultiThreaded-2
######### Writer : ------> STModel(tripId=113507911-1_38967) inside size : 23, On : MultiThreaded-2

---

如您所见，spring batch 过早检测到 steap 的结束，就在读取操作之后。

如何告诉 spring batch 步骤结束是在所有写入任务完成之后？

Answer 1

如你所见，spring batch 过早检测到 steap 的结束，

这是您应该期待的，因为您的编写器在后台异步写入项目而不等待它们完成。一旦作者的write 方法返回，该步骤将继续读取下一个项目块，如果没有更多项目要读取，则可能会完成。这可能发生在在前一个块由您的异步任务执行器在后台写入之前。

如何告诉 spring batch 步骤结束是在所有写入任务完成之后？

在您的编写器中，您需要submit 任务（而不是execute 他们），获取他们的Futures 的句柄并等待他们完成Future.get（如果需要，超时）。

编辑：添加示例

@Override
public void write(List<? extends T> items) {
    List<Future<?>> futures = new ArrayList<>(items.size());
    // write items in parallel (note the singleton list passed to the delegate)
    for (T item : items) { 
        Future<?> future = taskExecutor.submit(() -> {
            try {
                delegate.write(Collections.singletonList(item));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        futures.add(future);
    }
    // wait for futures to finish
    futures.forEach(future -> {
        try {
            future.get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    });
}

我在这里看到的唯一优点是项目将并行编写。这可以加快您为每个项目发送电子邮件的用例的速度。但是，我从您的 cmets 中看到以下内容：

写入与读取并发，因此可以加快处理速度！

这是不正确的，读取下一个块仍然会等待当前的写入操作完成。所以如上所述，您需要等待写入操作在您的写入器中完成，否则您的工作可能看起来已经完成，而项目仍在后台写入。

如果你真的希望读取和写入都由多个线程同时完成，你需要使用多线程步骤：

@Bean
public Step sampleStep() throws Exception{
    return stepBuilderFactory.get("processingStep")
                .<STModel, STModel>chunk(10)
                .reader(itpReader())
                .writer(writer())
                .taskExecutor(taskExecutor())
                .build();
}

另一种选择是使用并发步骤，如本期所述：https://github.com/spring-projects/spring-batch/issues/2044。如果您有兴趣，我有一个使用 BlockingQueue 作为暂存区 here 的 PoC。