Spring Batch - MySQL 到平面文件

在本章中,我们将创建一个使用 MySQL Reader 和 Flatfile Writer (.txt) 的 Spring Batch 应用程序。

Reader − 我们在应用程序中使用的 Reader 是 JdbcCursorItemReader,用于从 MySQL 数据库中读取数据。

假设我们在 MySQL 数据库中创建了一个表,如下所示。

实例


CREATE TABLE details.xml_mysql( 
   person_id int(10) NOT NULL, 
   sales VARCHAR(20), 
   qty int(3), 
   staffName VARCHAR(20), 
   date VARCHAR(20) 
); 

假设我们已将以下记录插入其中。

mysql> select * from tutorialsdata; 
+-------------+-----------------+----------------+-----------------+ 
| tutorial_id | tutorial_author | tutorial_title | submission_date | 
+-------------+-----------------+----------------+-----------------+ 
|         101 | Sanjay          | Learn Java     | 06-05-2007      | 
|         102 | Abdul S         | Learn MySQL    | 19-04-2007      | 
|         103 | Krishna Kasyap  | Learn JavaFX   | 06-07-2017      | 
+-------------+-----------------+----------------+-----------------+ 
3 rows in set (0.00 sec) 

Writer − 我们在应用程序中使用的 Writer 是 FlatFileItemWriter,用于将数据写入 flatfile (.txt)。

Processor − 我们在应用程序中使用的处理器是一个自定义处理器,它只打印从 CSV 文件中读取的记录。


jobConfig.xml

以下是我们的示例 Spring Batch 应用程序的配置文件。 在这个文件中,我们将定义 Job 和 Steps。 除了这些之外,我们还为 ItemReader、ItemProcessor 和 ItemWriter 定义了 bean。 (在这里,我们将它们与各自的类相关联,并传递所需属性的值来配置它们。)

实例


<beans xmlns = "http://www.springframework.org/schema/beans" 
   xmlns:batch = "http://www.springframework.org/schema/batch" 
   xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance" 
   xmlns:util = "http://www.springframework.org/schema/util" 
   xsi:schemaLocation = "http://www.springframework.org/schema/batch 
   
      http://www.springframework.org/schema/batch/spring-batch-2.2.xsd 
      http://www.springframework.org/schema/beans 
      http://www.springframework.org/schema/beans/spring-beans-3.2.xsd">  
   
   <import resource = "../jobs/context.xml" />  
   <bean id = "tutorial" class = "Tutorial" scope = "prototype" /> 
   <bean id = "itemProcessor" class = "CustomItemProcessor" />  
   
   <batch:job id = "helloWorldJob"> 
      <batch:step id = "step1"> 
         <batch:tasklet> 
            <batch:chunk reader = "mysqlItemReader" 
               writer = "flatFileItemWriter" processor = "itemProcessor" 
               commit-interval = "10"> 
            </batch:chunk> 
         </batch:tasklet> 
      </batch:step> 
   </batch:job> 
         
   <bean id = "mysqlItemReader" 
      class = "org.springframework.batch.item.database.JdbcCursorItemReader" > 
      <property name = "dataSource" ref = "dataSource" /> 
      <property name = "sql" value = "select * from details.tutorialsdata" /> 
      <property name = "rowMapper">  
         <bean class = "TutorialRowMapper" /> 
      </property> 
   </bean>
   
   <bean id = "flatFileItemWriter" 
      class = " org.springframework.batch.item.file.FlatFileItemWriter">      
      <property name = "resource" value = "file:target/outputfiles/employee_output.txt"/> 
      <property name = "lineAggregator"> 
         <bean class = " org.springframework.batch.item.file.transform.PassThroughLineAggregator"/> 
      </property> 
   </bean> 
</beans> 

Context.xml

以下是我们 Spring Batch 应用程序的 context.xml。 在这个文件中,我们将定义像作业存储库、作业启动器和事务管理器这样的 bean。

实例


<beans xmlns = "http://www.springframework.org/schema/beans" 
   xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance" 
   xmlns:jdbc = "http://www.springframework.org/schema/jdbc" 
   xsi:schemaLocation = "http://www.springframework.org/schema/beans 
      http://www.springframework.org/schema/beans/spring-beans-3.2.xsd 
      http://www.springframework.org/schema/jdbc 
      http://www.springframework.org/schema/jdbc/spring-jdbc-3.2.xsd "> 
   
   <!-- stored job-meta in database --> 
   <bean id = "jobRepository"  
      class = "org.springframework.batch.core.repository.support.JobRepositoryFactoryBean"> 
      <property name = "dataSource" ref = "dataSource" /> 
      <property name = "transactionManager" ref = "transactionManager" /> 
      <property name = "databaseType" value = "mysql" /> 
   </bean>  
 
   <bean id = "transactionManager"  
      class = "org.springframework.batch.support.transaction.ResourcelessTransactionManager" />  
   
   <bean id = "dataSource" 
      class = "org.springframework.jdbc.datasource.DriverManagerDataSource"> 
      <property name = "driverClassName" value = "com.mysql.jdbc.Driver" /> 
      <property name = "url" value = "jdbc:mysql://localhost:3306/details" /> 
      <property name = "username" value = "myuser" /> 
      <property name = "password" value = "password" /> 
   </bean> 
    
   <bean id = "jobLauncher"  
      class = "org.springframework.batch.core.launch.support.SimpleJobLauncher"> 
      <property name = "jobRepository" ref = "jobRepository" /> 
   </bean> 
  
   <!-- create job-meta tables automatically --> 
   <jdbc:initialize-database data-source = "dataSource">   
      <jdbc:script location = "org/springframework/batch/core/schema-drop-mysql.sql" />   
      <jdbc:script location = "org/springframework/batch/core/schema-mysql.sql" /> 
   </jdbc:initialize-database> 
</beans> 

CustomItemProcessor.java

以下是处理器类。 在这个类中,我们编写应用程序中的处理代码。 在这里,我们打印每条记录的内容。

实例


import org.springframework.batch.item.ItemProcessor;  

// Implementing the ItemProcessor interface 
public class CustomItemProcessor implements ItemProcessor<Tutorial, Tutorial> {  
 
   @Override 
   public Tutorial process(Tutorial item) throws Exception { 
      System.out.println("Processing..." + item); 
      return item; 
   } 
}

TutorialRowMapper.java

以下是 TutorialRowMapper 类,它将数据设置为 Tutorial 类。

实例


public class TutorialRowMapper implements RowMapper<Tutorial> {  
   
   @Override 
   public Tutorial mapRow(ResultSet rs, int rowNum) throws SQLException {  
  
      Tutorial tutorial = new Tutorial();  
  
      tutorial.setTutorial_id(rs.getInt("tutorial_id")); 
      tutorial.setTutorial_title(rs.getString("tutorial_title")); 
      tutorial.setTutorial_author(rs.getString("tutorial_author")); 
      tutorial.setSubmission_date(rs.getString("submission_date"));  
      return tutorial; 
   } 
}

Tutorial.java

以下是 Tutorial 类。 它是一个带有 settergetter 方法的简单 Java 类。 在这个类中,我们使用注解将这个类的方法与 XML 文件的标签相关联。

实例


public class Tutorial { 
   private int tutorial_id; 
   private String tutorial_title; 
   private String tutorial_author; 
   private String submission_date; 
  
   public int getTutorial_id() { 
      return tutorial_id; 
   }  
   
   public void setTutorial_id(int tutorial_id) { 
      this.tutorial_id = tutorial_id; 
   }
   
   public String getTutorial_title() { 
      return tutorial_title; 
   }   
 
   public void setTutorial_title(String tutorial_title) { 
      this.tutorial_title = tutorial_title; 
   }  
   
   public String getTutorial_author() { 
      return tutorial_author; 
   }  
 
   public void setTutorial_author(String tutorial_author) { 
      this.tutorial_author = tutorial_author; 
   }  
 
   public String getSubmission_date() { 
      return submission_date; 
   }  
   public void setSubmission_date(String submission_date) { 
      this.submission_date = submission_date; 
   }  
 
   @Override 
   public String toString() { 
      return " [id=" + tutorial_id + ", title=" + 
      tutorial_title                      + ", 
      author=" + tutorial_author + ", date=" + 
      submission_date + "]"; 
   } 
}    

App.java

以下是启动批处理的代码。 在本课程中,我们将通过运行 JobLauncher 来启动批处理应用程序。

实例


import org.springframework.batch.core.Job; 
import org.springframework.batch.core.JobExecution; 
import org.springframework.batch.core.JobParameters; 
import org.springframework.batch.core.launch.JobLauncher; 
import org.springframework.context.ApplicationContext; 
import org.springframework.context.support.ClassPathXmlApplicationContext;  

public class App {  
   
   public static void main(String[] args) throws Exception { 
     
      String[] springConfig  =  { "jobs/job_hello_world.xml" };  
      
      // Creating the application context object  
      ApplicationContext context = new ClassPathXmlApplicationContext(springConfig);  
      
      // Creating the job launcher 
      JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher"); 
    
      // Creating the job 
      Job job = (Job) context.getBean("helloWorldJob"); 
    
      // Executing the JOB 
      JobExecution execution = jobLauncher.run(job, new JobParameters()); 
      System.out.println("Exit Status : " + execution.getStatus()); 
   } 
}

在执行此应用程序时,它将产生以下输出。

May 09, 2017 5:44:48 PM org.springframework.context.support.ClassPathXmlApplicationContext prepareRefresh 
INFO: Refreshing org.springframework.context.support.ClassPathXml
ApplicationContext@3d646c37: startup date [Tue May 
09 17:44:48 IST 2017]; root of context hierarchy 
May 09, 2017 5:44:48 PM org.springframework.beans.factory.xml.XmlBeanDefinitionReader loadBeanDefinitions  
May 09, 2017 5:44:56 PM org.springframework.batch.core.launch.support.SimpleJobLauncher run 
INFO: Job: [FlowJob: [name=helloWorldJob]] launched 
with the following parameters: [{}] 
May 09, 2017 5:44:56 PM org.springframework.batch.core.job.SimpleStepHandler handleStep 
INFO: Executing step: [step1] 
Processing...Report [id=101, title=Learn Java, author=Sanjay, date=06-05-2007] 
Processing...Report [id=102, title=Learn MySQL, author=Abdul S, date=19-04-2007] 
Processing...Report [id=103, title=Learn JavaFX, author=Krishna Kasyap, date=0607-2017] 
May 09, 2017 5:44:57 PM org.springframework.batch.core.launch.support.SimpleJobLauncher run 
INFO: Job: [FlowJob: [name=helloWorldJob]] completed with the following parameters: 
[{}] and the following status: [COMPLETED] 
Hello 
Exit Status : COMPLETED 

这将生成一个包含以下内容的 .txt 文件。

Report [id=101, title=Learn Java, author=Sanjay, date=06-05-2007] 
Report [id=102, title=Learn MySQL, author=Abdul S, date=19-04-2007] 
Report [id=103, title=Learn JavaFX, author=Krishna Kasyap, date=06-07-2017]