问Spring batch和Spring Boot中的DataSource问题

EN

spring.batch.initialize-schema=always

batch.jdbc.driver=oracle.jdbc.OracleDriver
batch.jdbc.url=jdbc:oracle:thin:@oracle:1521:xe
batch.jdbc.user=spring
batch.jdbc.password=spring
batch.jdbc.testWhileIdle=false
batch.jdbc.validationQuery=
batch.drop.script=classpath:/org/springframework/batch/core/schema-drop-oracle10g.sql
batch.schema.script=classpath:/org/springframework/batch/core/schema-oracle10g.sql
batch.business.schema.script=business-schema-oracle10g.sql
batch.database.incrementer.class=org.springframework.jdbc.support.incrementer.OracleSequenceMaxValueIncrementer
batch.database.incrementer.parent=sequenceIncrementerParent
batch.lob.handler.class=org.springframework.jdbc.support.lob.OracleLobHandler
batch.grid.size=2
batch.jdbc.pool.size=6
batch.verify.cursor.position=true
batch.isolationlevel=ISOLATION_SERIALIZABLE
batch.table.prefix=BATCH_

package com.barley.batch.config;
    
import java.sql.ResultSet;

import javax.sql.DataSource;

import com.barley.batch.listener.JobCompletionNotificationListener;
import com.barley.batch.model.RecordSO;
import com.barley.batch.model.WriterSO;
import com.barley.batch.processor.RecordProcessor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.ItemPreparedStatementSetter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

   

 private static final Logger LOGGER = LoggerFactory.getLogger(BatchConfiguration.class);

            @Autowired
            private JobBuilderFactory jobBuilderFactory;
        
            @Autowired
            private StepBuilderFactory stepBuilderFactory;
        
            @Autowired
            private DataSource dataSource;
        
            @Bean
            public ItemReader<RecordSO> reader() {
                return new JdbcCursorItemReaderBuilder<RecordSO>().name("the-reader")
                        .sql("select id, firstName, lastname, random_num from reader").dataSource(dataSource)
                        .rowMapper((ResultSet resultSet, int rowNum) -> {
                            if (!(resultSet.isAfterLast()) && !(resultSet.isBeforeFirst())) {
                                RecordSO recordSO = new RecordSO();
                                recordSO.setFirstName(resultSet.getString("firstName"));
                                recordSO.setLastName(resultSet.getString("lastname"));
                                recordSO.setId(resultSet.getInt("Id"));
                                recordSO.setRandomNum(resultSet.getString("random_num"));
        
                                LOGGER.info("RowMapper record : {}", recordSO);
                                return recordSO;
                            } else {
                                LOGGER.info("Returning null from rowMapper");
                                return null;
                            }
                        }).build();
            }
        
            @Bean
            public ItemProcessor<RecordSO, WriterSO> processor() {
                return new RecordProcessor();
            }
        
            @Bean
            public JdbcBatchItemWriter<WriterSO> writer(DataSource dataSource, ItemPreparedStatementSetter<WriterSO> setter) {
                return new JdbcBatchItemWriterBuilder<WriterSO>()
                        .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
                        .itemPreparedStatementSetter(setter)
                        .sql("insert into writer (id, full_name, random_num) values (?,?,?)").dataSource(dataSource).build();
            }
        
            @Bean
            public ItemPreparedStatementSetter<WriterSO> setter() {
                return (item, ps) -> {
                    ps.setLong(1, item.getId());
                    ps.setString(2, item.getFullName());
                    ps.setString(3, item.getRandomNum());
                };
            }
        
            @Bean
            public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
                return jobBuilderFactory.get("importUserJob").incrementer(new RunIdIncrementer()).listener(listener).flow(step1)
                        .end().build();
            }
        
            @Bean
            public Step step1(JdbcBatchItemWriter<WriterSO> writer, ItemReader<RecordSO> reader) {
                return stepBuilderFactory.get("step1").<RecordSO, WriterSO>chunk(5).reader(reader).processor(processor())
                        .writer(writer).build();
            }
        }
    
    -------
    
    package com.barley.batch.listener;
    
    import java.util.List;
    
    import com.barley.batch.model.WriterSO;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.batch.core.BatchStatus;
    import org.springframework.batch.core.JobExecution;
    import org.springframework.batch.core.listener.JobExecutionListenerSupport;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
    
        private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);
        private final JdbcTemplate jdbcTemplate;
    
        public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {
            this.jdbcTemplate = jdbcTemplate;
        }
    
        @Override
        public void afterJob(JobExecution jobExecution) {
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                log.info("!!! JOB FINISHED! Time to verify the results");
    
                List<WriterSO> results = jdbcTemplate.query("SELECT id, full_name, random_num FROM writer", (rs, row) -> {
                    WriterSO writerSO = new WriterSO();
                    writerSO.setId(rs.getLong("id"));
                    writerSO.setFullName(rs.getString("full_name"));
                    writerSO.setRandomNum(rs.getString("random_num"));
                    return writerSO;
                });
    
                for (WriterSO writerSO : results) {
                    log.info("Found <" + writerSO + "> in the database.");
                }
            }
        }
    }

----------------

package com.barley.batch.model;

public class RecordSO {

    private long id;
    private String firstName;
    private String lastName;
    private String randomNum;

    public long getId() {
        return id;
    }

    public void setId(long id) {
        this.id = id;
    }

    public String getFirstName() {
        return firstName;
    }

    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }

    public String getLastName() {
        return lastName;
    }

    public void setLastName(String lastName) {
        this.lastName = lastName;
    }

    public String getRandomNum() {
        return randomNum;
    }

    public void setRandomNum(String randomNum) {
        this.randomNum = randomNum;
    }

    @Override
    public String toString() {
        return "RecordSO{" +
                "id=" + id +
                ", firstName='" + firstName + '\'' +
                ", lastName='" + lastName + '\'' +
                ", randomNum='" + randomNum + '\'' +
                '}';
    }
}

--------------

package com.barley.batch.model;

public class WriterSO {

    private long id;
    private String fullName;
    private String randomNum;

    public long getId() {
        return id;
    }

    public void setId(long id) {
        this.id = id;
    }

    public String getFullName() {
        return fullName;
    }

    public void setFullName(String fullName) {
        this.fullName = fullName;
    }

    public String getRandomNum() {
        return randomNum;
    }

    public void setRandomNum(String randomNum) {
        this.randomNum = randomNum;
    }

    @Override
    public String toString() {
        return "WriterSO{" +
                "id=" + id +
                ", fullName='" + fullName + '\'' +
                ", randomNum='" + randomNum + '\'' +
                '}';
    }
}

-----------

package com.barley.batch.processor;

import com.barley.batch.model.RecordSO;
import com.barley.batch.model.WriterSO;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;

public class RecordProcessor implements ItemProcessor<RecordSO, WriterSO> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RecordProcessor.class);

    @Override
    public WriterSO process(final RecordSO item) throws Exception {
        LOGGER.info("Processing Record: {}", item);
        WriterSO writerSo = new WriterSO();
        writerSo.setId(item.getId());
        writerSo.setFullName(item.getFirstName() + " " + item.getLastName());
        writerSo.setRandomNum(String.valueOf(Math.random()).substring(3, 8));
        LOGGER.info("Processed Writer: {}", writerSo);
        return writerSo;
    }
}

---------

package com.barley.batch;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {
    public static void main(String[] args) throws Exception {
        SpringApplication.run(Application.class, args);
    }
}

-------------------

spring.batch.initialize-schema=ALWAYS
spring.datasource.driverClassName=com.mysql.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/batch
spring.datasource.username=root
spring.datasource.password=root
spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect
spring.jpa.show-sql=false
spring.jpa.hibernate.ddl-auto=create-drop

---------------------------

DROP TABLE IF EXISTS reader;
DROP TABLE IF EXISTS writer;

CREATE TABLE `reader` (`id` INT  NOT NULL AUTO_INCREMENT,`firstName`  VARCHAR(20) NULL,`lastName`   VARCHAR(20) NULL,`random_num` VARCHAR(20) NULL, PRIMARY KEY (`id`));

CREATE TABLE `writer` (`id` INT  NOT NULL AUTO_INCREMENT,`full_name`  VARCHAR(40) NULL,`random_num` VARCHAR(20) NULL, PRIMARY KEY (`id`));

INSERT INTO `reader` (`firstName`, `lastName`, `random_num`) VALUES ('abc', 'def', '1');
INSERT INTO `reader` (`firstName`, `lastName`, `random_num`) VALUES ('def', 'zhu', '2');
INSERT INTO `reader` (`firstName`, `lastName`, `random_num`) VALUES ('dummy', 'name', '3');
INSERT INTO `reader` (`firstName`, `lastName`, `random_num`) VALUES ('non', 'pay', '4');
INSERT INTO `reader` (`firstName`, `lastName`, `random_num`) VALUES ('spring', 'batch', '5');