如何在Spring Batch中实现支持多租户的RepositoryItemReader？

场景与问题 #

已实现基于Job参数运行的批处理任务，集成Spring Web+Quartz支持按需调用和定时调度，当前用RepositoryItemReader配合Spring Data运行正常。需为任务添加多租户机制（tenant1/tenant2/tenant3对应独立数据库），要求根据Job参数中的tenantId让任务读取对应租户的数据。

当前困境：使用AsyncTaskScheduler启动任务，且上下文存在其他任务，不清楚如何将租户上下文注入到任务线程中。

现有代码片段：

任务启动代码 #

JobParameters jobParameters = new JobParametersBuilder()
        .addString("tenantId",tenantId)
        .addString("jobName",jobName)
        .addLong("time", System.currentTimeMillis()).toJobParameters();
Job job = jobRegistry.getJob(jobName);
JobExecution jobExecution = asyncJobLauncher.run(job, jobParameters);

ItemReader Bean定义 #

@StepScope
@Bean
public ItemReader<Person> itemReader() {
    return new RepositoryItemReaderBuilder<Person>()
            .name("ItemReader")
            .repository(personRepository)
            .arguments("personName").methodName("findByPersonNameEquals")
            .maxItemCount(30).pageSize(5)
            .sorts(Collections.singletonMap("createTs", Sort.Direction.ASC)).build();
}

解决方案步骤 #

1. 实现线程安全的租户上下文持有者 #

定义线程本地存储，用于在任务线程中传递租户ID：

public class TenantContextHolder {
    private static final ThreadLocal<String> TENANT_ID = new ThreadLocal<>();

    public static void setTenantId(String tenantId) {
        TENANT_ID.set(tenantId);
    }

    public static String getTenantId() {
        return TENANT_ID.get();
    }

    public static void clear() {
        TENANT_ID.remove();
    }
}

2. 自定义Job执行监听器，注入租户上下文 #

通过Spring Batch的JobExecutionListener，在任务执行前从Job参数中提取tenantId并设置到线程上下文，任务结束后清理上下文，避免线程复用导致的租户污染：

public class TenantJobExecutionListener implements JobExecutionListener {
    @Override
    public void beforeJob(JobExecution jobExecution) {
        String tenantId = jobExecution.getJobParameters().getString("tenantId");
        if (tenantId != null) {
            TenantContextHolder.setTenantId(tenantId);
        }
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        TenantContextHolder.clear();
    }
}

将该监听器注册到目标Job中：

@Bean
public Job myBatchJob(Step myStep, TenantJobExecutionListener tenantListener) {
    return jobBuilderFactory.get("myBatchJob")
            .listener(tenantListener)
            .start(myStep)
            .build();
}

3. 实现多租户数据源路由 #

继承Spring的AbstractRoutingDataSource，根据租户上下文动态切换数据源：

public class TenantRoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return TenantContextHolder.getTenantId();
    }
}

在配置类中配置多数据源及路由规则：

@Configuration
public class DataSourceConfig {
    @Bean
    public DataSource tenant1DataSource() {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/tenant1_db");
        dataSource.setUsername("user1");
        dataSource.setPassword("pass1");
        return dataSource;
    }

    @Bean
    public DataSource tenant2DataSource() {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/tenant2_db");
        dataSource.setUsername("user2");
        dataSource.setPassword("pass2");
        return dataSource;
    }

    @Bean
    public DataSource tenant3DataSource() {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/tenant3_db");
        dataSource.setUsername("user3");
        dataSource.setPassword("pass3");
        return dataSource;
    }

    @Bean
    public DataSource routingDataSource() {
        TenantRoutingDataSource routingDataSource = new TenantRoutingDataSource();
        Map<Object, Object> targetDataSources = new HashMap<>();
        targetDataSources.put("tenant1", tenant1DataSource());
        targetDataSources.put("tenant2", tenant2DataSource());
        targetDataSources.put("tenant3", tenant3DataSource());
        routingDataSource.setTargetDataSources(targetDataSources);
        routingDataSource.setDefaultTargetDataSource(tenant1DataSource());
        return routingDataSource;
    }

    @Bean
    public LocalContainerEntityManagerFactoryBean entityManagerFactory(DataSource routingDataSource) {
        LocalContainerEntityManagerFactoryBean em = new LocalContainerEntityManagerFactoryBean();
        em.setDataSource(routingDataSource);
        em.setPackagesToScan("com.yourpackage.model");
        HibernateJpaVendorAdapter vendorAdapter = new HibernateJpaVendorAdapter();
        em.setJpaVendorAdapter(vendorAdapter);
        
        Properties properties = new Properties();
        properties.setProperty("hibernate.hbm2ddl.auto", "none");
        properties.setProperty("hibernate.dialect", "org.hibernate.dialect.MySQL8Dialect");
        em.setJpaProperties(properties);
        return em;
    }
}

4. 验证ItemReader的多租户适配 #

由于你的ItemReader标记了@StepScope，Spring Batch会在任务执行时创建实例，此时租户上下文已由TenantJobExecutionListener设置完成，RepositoryItemReader使用的personRepository会自动通过路由数据源访问对应租户的数据库，无需修改现有ItemReader代码。

5. 异步任务线程隔离保障 #

TenantJobExecutionListener的afterJob方法会在任务结束后清理线程本地的租户上下文，避免AsyncTaskScheduler的线程池复用导致的租户信息串流问题。

内容的提问来源于stack exchange，提问作者abreddy