使用多个数据源的Spring批处理慢写

Question

我有一个spring批处理应用程序，可以将5M条记录从一个文件加载到SQL Server数据库中。我有根据国家代码区分的不同数据源。当我使用带有@primary注释的单个数据源时，spring batch writer在5分钟内写入5M条记录。但是，当我使用@bean注释给出多个数据源并使用一个非主数据源将文件数据写入数据库时，perforamnce变得非常慢，对于同样的500万条记录，大约需要15分钟。谁能解释一下Spring Batch与主数据源和其他数据源的行为是否不同。

    @Repository
    @ComponentScan(basePackages = "com.abc.extract")
    @EnableBatchProcessing
    @EnableTransactionManagement
    @Slf4j
    public class DataSourceConfig {
    
        public DataSourceConfig() {
    
        }
    
        @Autowired
        private CCMConfiguration ccmConfig;
    
        @Autowired
        SpotlightConfig spotlight;
    
        private DataSource datasourcesVal = null;
        
        private DataSource datasourcesVal1 = null;
        
        private Map<String,DataSource> datasourcesMap = null;
    
        @Value("${app.country.code}")
        String countryCode;
        
        @Primary
        @Bean("batchprimary")
        public HikariDataSource hikariDataSource(@Value("${app.country.code}")
        String countryCode1) throws IllegalBlockSizeException, NoSuchPaddingException, BadPaddingException,
                NoSuchAlgorithmException, InvalidKeyException {
            System.out.println("Contrycode:"+countryCode1);
            SqlServerConfiguration dbConfiguration = ccmConfig.getSqlServerDatabaseConfig(countryCode1);
            HikariDataSource ds = null;
            try {
                final String password = dbConfiguration.getDataBasePwd();
                final String username = dbConfiguration.getDataBaseUserName();
                ds = (HikariDataSource) DataSourceBuilder.create().type(HikariDataSource.class).build();
                ds.setUsername(username);
                ds.setPassword(password);
                ds.setDriverClassName(dbConfiguration.getDataBaseDriver());
                ds.setJdbcUrl(dbConfiguration.getDataBaseUrl());
                ds.setConnectionTestQuery("SELECT 1");
                ds.setConnectionTimeout(dbConfiguration.getConnectionTimeout());
                ds.setIdleTimeout(dbConfiguration.getIdleTimeout());
                ds.setMaximumPoolSize(dbConfiguration.getHikariPoolSize());
                //ds.setMaximumPoolSize(2);
                ds.setMaxLifetime(dbConfiguration.getMaxLifetime());
                ds.setLeakDetectionThreshold(dbConfiguration.getLeakDetectionThreshold());
                ds.setPoolName(dbConfiguration.getPoolName());
                this.datasourcesVal = ds;
            } catch (Exception ex) {
                spotlight.sendNotification(ex, "Critical: Failed to establish Database connection");
                log.error(ex.getMessage());
            } finally {
                if (!Objects.nonNull(ds) || ds.isClosed()) {
                    ds.close();
                    spotlight.sendNotification(new NullPointerException("Primary data source creation failed"),
                            "Critical: Failed to establish Database connection");
                }
            }
            return ds;
    
        }
    
    
        

    @Bean("batchprimary1")
            public HikariDataSource hikariDataSource1(@Value("${app.country.code}")
            String countryCode1) throws IllegalBlockSizeException, NoSuchPaddingException, BadPaddingException,
                    NoSuchAlgorithmException, InvalidKeyException {
                System.out.pri

ntln("Contrycode:"+countryCode1);
            SqlServerConfiguration dbConfiguration = ccmConfig.getSqlServerDatabaseConfig(countryCode1);
            HikariDataSource ds = null;
            try {
                final String password = dbConfiguration.getDataBasePwd();
                final String username = dbConfiguration.getDataBaseUserName();
                ds = (HikariDataSource) DataSourceBuilder.create().type(HikariDataSource.class).build();
                ds.setUsername(username);
                ds.setPassword(password);
                ds.setDriverClassName(dbConfiguration.getDataBaseDriver());
                ds.setJdbcUrl(dbConfiguration.getDataBaseUrl());
                ds.setConnectionTestQuery("SELECT 1");
                ds.setConnectionTimeout(dbConfiguration.getConnectionTimeout());
                ds.setIdleTimeout(dbConfiguration.getIdleTimeout());
                ds.setMaximumPoolSize(dbConfiguration.getHikariPoolSize());
                //ds.setMaximumPoolSize(2);
                ds.setMaxLifetime(dbConfiguration.getMaxLifetime());
                ds.setLeakDetectionThreshold(dbConfiguration.getLeakDetectionThreshold());
                ds.setPoolName(dbConfiguration.getPoolName());
                //this.datasourcesVal = ds;
            } catch (Exception ex) {
                spotlight.sendNotification(ex, "Critical: Failed to establish Database connection");
                log.error(ex.getMessage());
            } finally {
                if (!Objects.nonNull(ds) || ds.isClosed()) {
                    ds.close();
                    spotlight.sendNotification(new NullPointerException("Primary data source creation failed"),
                            "Critical: Failed to establish Database connection");
                }
            }
            return ds;
    
        }





    @Autowired
        @Qualifier("batchprimary")
        public DataSource datasourcesVal;
        
        @Autowired
        @Qualifier("batchprimary1")
        public DataSource datasourcesVal1;
    @Bean
        @JobScope
        public Step ExtractNLoadItemOnHand(TaskExecutor taskExecutor,@Value("#{jobParameters}") Map<String, JobParameter> jobParameters)
                throws InvalidKeyException, IllegalBlockSizeException,
                NoSuchPaddingException, BadPaddingException,
                NoSuchAlgorithmException, SQLException, ValidationException,
                CalpiDataException {
            int chunkSize = ApplicationConstants.CHUNK_SIZE;
            log.info(" Extract and Load ItemOnHand Job - Started....");
            countryCode = jobParameters.get("CountryCode").toString();
            //dataSourceTemp = datasourcesMap.get(countryCode);
            return stepBuilderFactory
                    .get("READ ITEM ONHAND STEP")
                    .listener(stepListners)
                    .<AccumOnhand, AccumOnhand> chunk(chunkSize)
                    .reader(extractItemOnHand(null))
                    .processor(new ItemProcessor<AccumOnhand, AccumOnhand>() {
                        @Override
                        public AccumOnhand process(AccumOnhand accumOnhand)
                                throws Exception {
                            return accumOnhand;
                        }
                    })
                    .writer(compositeItemWriter.compositeItemWriter(Arrays
                            .asList(loadItemOnHand()))).faultTolerant()
                    .retryLimit(ApplicationConstants.RETRY_SKIP_LIMIT)
                    .retry(Exception.class).skip(Exception.class)
                    .skipLimit(ApplicationConstants.RETRY_SKIP_LIMIT)
                    .taskExecutor(taskExecutor)
                    .throttleLimit(ApplicationConstants.THROTTLE_LIMIT).build();
        }
    
        @Bean
        @StepScope
        public JdbcBatchItemWriter<AccumOnhand> loadItemOnHand()
                throws InvalidKeyException, IllegalBlockSizeException,
                NoSuchPaddingException, BadPaddingException,
                NoSuchAlgorithmException, ValidationException, CalpiDataException {
            try {
                //country = jobParameters.get("CountryCode");
                JdbcBatchItemWriter<AccumOnhand> writer;
                //datasource = dataSourceConfig.hikariDataSource(country.toString());
                if(countryCode.equals("MX")) {
                     writer = stepDBItemWriter.writeDBData(
                            MessageFormat.format(SQLConstants.INS_ACCUM_ONHAND,
                                    countryCode), datasourcesVal);
                } else {
                     writer = stepDBItemWriter.writeDBData(
                                MessageFormat.format(SQLConstants.INS_ACCUM_ONHAND,
                                        countryCode), datasourcesVal1);
                }
        
                
                log.info("Item on hand writer created successfully");
                return writer;
            } catch (Exception ex) {
                log.error("Load loadItemOnHand Failed {} ", ex);
                throw new ValidationException("Load loadItemOnHand Failed", ex);
            }
        }

Answer 1

以下是根据我使用spring Batch的经验提出的建议。

• 确保每个数据库都具有相同的写入性能
• 是否尝试使用commitInterval（）？
• 尽可能使用JdbcItemWriter或任何基于JDBC的策略编写
• 运行时检查cpu利用率。如果超过90%，则速度将降低
• 如果捕获错误，尝试捕获将影响性能