并行的Spring Batch作业导致数据混乱

后端开发 2026-07-12

我想为每个文件独立运行一个Spring Batch作业实例，最多同时20个。不知道为什么，作业在同一组参数下被多次启动（同一文件名+ 同一UUID），尽管这些参数是唯一的。可能是什么原因导致UUID和文件名被重复使用？

我已经简化了一个用于重现问题的作业实现。下面是我通过主类调用作业的方式：

public class Main implements CommandLineRunner {
    private final JobLauncher jobLauncher;
    private final Job jobConfig;
    private int parallelism = 20;

    public static void main(String[] args) {SpringApplication.run(Main.class, args);}

    @Override
    public void run(String... args) throws Exception {
        File[] files = getFiles();

        ExecutorService pool = Executors.newFixedThreadPool(parallelism);
        List<Future<?>> futures = new ArrayList<>(files.length);

        for (int i = 0; i < files.length; i++) {
            final File file = files[i];

            futures.add(pool.submit(() -> {
                try {
                    JobParameters params = new JobParametersBuilder()
                            .addString("fileName", file.getCanonicalPath(), true)
                            .addString("guid", UUID.randomUUID().toString(), true)
                            .toJobParameters();
                    JobExecution exec = jobLauncher.run(jobConfig, params);
                    return exec.getExitStatus();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }));
        }
        for (Future<?> f : futures) {
            f.get();
        }
        pool.shutdown();
    }

    public static File[]  getFiles() {
        File currentDir = new File("C:\\Code\\spring-batch\\input");
        File[] files = currentDir.listFiles((dir, name) -> name.endsWith(".csv"));
        return files;
    }

这是作业。它所做的只是根据inputFile名称创建一个输出文件。

@Configuration
@RequiredArgsConstructor
public class JobConfig {

    @Bean
    public PlatformTransactionManager transactionManager() {
        return new ResourcelessTransactionManager();
    }

    @Bean
    public JobLauncher jobLauncher(JobRepository jobRepository) throws Exception {
        TaskExecutorJobLauncher launcher = new TaskExecutorJobLauncher();
        launcher.setJobRepository(jobRepository);
        launcher.setTaskExecutor(new org.springframework.core.task.SyncTaskExecutor());
        launcher.afterPropertiesSet();
        return launcher;
    }
    @Bean
    public Job simpleJob(JobRepository jobRepository, Step simpleStep) {
        return new JobBuilder("simpleJob", jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(simpleStep)
                .build();
    }

    @Bean
    public Step simpleStep(JobRepository jobRepository,
                           PlatformTransactionManager transactionManager,
                           ItemReader<String> reader,
                           ItemProcessor<String, String> processor,
                           FlatFileItemWriter<String> writer) {

        return new StepBuilder("simpleStep", jobRepository)
                .<String, String>chunk(100, transactionManager)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .stream(writer)
                .build();
    }

    @Bean
    @StepScope
    public FlatFileItemReader<String> reader(
            @Value("#{jobParameters['fileName']}") String fileName,
            @Value("#{jobParameters['guid']}") String guid) {

        return new FlatFileItemReaderBuilder<String>()
                .name("reader-" + guid)
                .resource(new FileSystemResource(fileName))
                .lineMapper((line, lineNumber) -> line)
                .build();
    }

    @Bean
    public ItemProcessor<String, String> processor() {
        return item -> item.toUpperCase();
    }


    @Bean
    @StepScope
    public FlatFileItemWriter<String> writer(
            @Value("#{jobParameters['fileName']}") String fileName,
            @Value("#{jobParameters['guid']}") String guid) {

        String outputName = "OUTPUT_" + getFileNameWithoutExtension(fileName) + "_" + ".out";

        return new FlatFileItemWriterBuilder<String>()
                .name("writer-" + guid)
                .resource(new FileSystemResource(outputName))
                .lineAggregator(new PassThroughLineAggregator<>())
                .build();
    }

    public static String getFileNameWithoutExtension(String fileName) {
      String filename = java.nio.file.Paths.get(fileName).getFileName().toString();
      int dotIndex = filename.lastIndexOf('.');
      if (dotIndex > 0) {
        return filename.substring(0, dotIndex);
      }
      return fileName;
    }
}

我可以确认这会返回每个CSV文件中的一个，没有重复，File[] 数组是唯一的。问题在于，某种方式joblauncher启动的实例竟然使用完全相同的文件名和GUID。下面是我的输出，你可以看到有几个文件名被多次启动且使用相同的UUID（ex_13.csv、ex_19.csv）。

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-20\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_18.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=78696daa-4df3-473f-87cf-42b9a7dc1169, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-4\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_12.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=94a99043-91e6-4466-87e1-b70683e0ab5b, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-5\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_13.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=f9c2ecae-f68a-4c79-a5ef-4af7e0f3930c, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-15\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_13.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=f9c2ecae-f68a-4c79-a5ef-4af7e0f3930c, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-1\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_1.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=1c21c4a5-4956-4fcf-8cce-868ab1bf78cb, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-8\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_16.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=27042f9d-f963-4d96-ad6e-b5dabe18ad2e, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-7\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_15.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=ef576a4e-d225-4eb2-91a4-b3fb1d8b562d, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-3\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_11.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=1b596d93-4311-44e7-9251-84f8eaa9ba21, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-13\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_20.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=18bde259-67ea-4c0d-a20d-3c82b654019b, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-18\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_7.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=e8885719-be1e-4190-996b-011d77cc015b, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-10\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_18.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=78696daa-4df3-473f-87cf-42b9a7dc1169, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-9\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_19.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=f818a9c9-65c2-4fae-b4d4-0f39c0b79a3f, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-17\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_10.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=c47e6895-3ab0-483e-bc2a-adbd334bd33e, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-11\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_19.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=f818a9c9-65c2-4fae-b4d4-0f39c0b79a3f, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-2\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_10.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=c47e6895-3ab0-483e-bc2a-adbd334bd33e, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-19\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_8.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=1bc84fd7-4cf3-4163-a406-2edbcd297045, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-16\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_5.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=5166ebc1-5e03-4d25-a332-9a04cbc2530b, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-12\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_10.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=c47e6895-3ab0-483e-bc2a-adbd334bd33e, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[pool-2-thread-6\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_8.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=1bc84fd7-4cf3-4163-a406-2edbcd297045, type=class java.lang.String, identifying=true}}\]

2026-02-28T23:15:53.606-05:00  INFO 16316 --- \[ool-2-thread-14\] o.s.b.c.l.s.TaskExecutorJobLauncher      : Job: \[SimpleJob: \[name=simpleJob\]\] launched with the following parameters: \[{JobParameter{name='fileName', value=C:\\Code\\spring-batch\\input\\ex_3.csv, type=class java.lang.String, identifying=true},JobParameter{name='guid', value=1bb5fea1-333f-4bba-b6c7-ece83149bd18, type=class java.lang.String, identifying=true}}\]

解决方案

问题是在你的循环中存在一个竞争条件。当你使用 files[i] 时，线程池的快速迭代导致多个线程在下一轮完成之前获取相同的索引值。

// Change your loop to this:
for (File file : files) {
    futures.add(pool.submit(() -> {
        JobParameters params = new JobParametersBuilder()
                .addString("fileName", file.getCanonicalPath(), true)
                .addString("guid", UUID.randomUUID().toString(), true)
                .toJobParameters();
        return jobLauncher.run(jobConfig, params).getExitStatus();
    }));
}

解决思路：
在一个标准的 for(int i...) 循环中，变量 i 在线程仍在尝试读取 files[i] 时就被更新。增强的for循环为每次迭代创建一个新的、独立的变量，因此每个线程都能保证获得自己的唯一文件并生成唯一的UUID。

站内所有文章版权归属LeftHeroAI导航站，无授权禁止任何主体转载、抄袭、复制内容，亦不得私自架设镜像站点。一经侵权，本站将通过法律途径追责。

相关文章

在同一运行环境中，为什么一个算法能比另一个算法更快地把相同的分数加起来？
3
在PostgreSQL 16 / PolarDB-PG 16中，如何在分区表的父表上并发创建索引？
0
在被 @TestConfiguration注解的类中，标注了 @DynamicPropertySource的方法不起作用
0
Rspack太慢了
0
在Spring Boot的测试中，Testcontainers会在无需显式的启动代码或注解的情况下自动启动
1
如何在静态初始化之前调用Spring的 @Service方法（在Spring上下文尚未完全加载时）？
1
汇集全球数千款 AI 工具，每日持续更新，配套 AI 开发学习资源，轻松用 AI 提升工作效率。快捷键 Ctrl+D/⌘+D 一键收藏本站。

超人AI导航关于我们免责声明

Copyright © 超人AI导航  辽ICP备2024029316号-4