spring Batch实现数据库大数据量读写

短命女 2022-08-21 12:55 12阅读 0赞

1. data-source-context.xml

Xml代码   ![收藏代码][icon_star.png]

1.  **<?****xml** version="1.0" encoding="UTF-8"**?>**  
2.  **<****beans** xmlns="http://www.springframework.org/schema/beans"  
3.      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  
4.      xmlns:tx="http://www.springframework.org/schema/tx"  
5.      xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
6.      http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd  
7.      http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd"**>**  
8.    
9.        
10.     **<****context:component-scan** base-package="net.etongbao.vasp.ac" **/>**  
11.       
12.       
13.    
14.     **<****bean** id="dataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource"  
15.         destroy-method="close" abstract="false" scope="singleton"**>**  
16.           
17.         **<****property** name="driverClass" value="oracle.jdbc.OracleDriver" **/>**  
18.         **<****property** name="jdbcUrl" value="jdbc:oracle:thin:@192.168.1.23:1521:orcl01" **/>**  
19.         **<****property** name="user" value="USR\_DEV01" **/>**  
20.         **<****property** name="password" value="2AF0829C" **/>**  
21.         **<****property** name="checkoutTimeout" value="30000" **/>**  
22.         **<****property** name="maxIdleTime" value="120" **/>**  
23.         **<****property** name="maxPoolSize" value="100" **/>**  
24.         **<****property** name="minPoolSize" value="2" **/>**  
25.         **<****property** name="initialPoolSize" value="2" **/>**  
26.         **<****property** name="maxStatements" value="0" **/>**  
27.         **<****property** name="maxStatementsPerConnection" value="0" **/>**  
28.         **<****property** name="idleConnectionTestPeriod" value="30" **/>**     
29.     **</****bean****>**  
30.     **<****bean** id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate"**>**  
31.         **<****property** name="dataSource" ref="dataSource" **/>**  
32.     **</****bean****>**  
33.   
34.     **<****bean** id="transactionManager"  
35.         class="org.springframework.jdbc.datasource.DataSourceTransactionManager"**>**  
36.         **<****property** name="dataSource" ref="dataSource" **/>**  
37.     **</****bean****>**  
38.       
39.     **<****tx:annotation-driven** transaction-manager="transactionManager" **/>**  
40. **</****beans****>**

2. quartz-context.xml       commit-interval="10000"每次批量数据的条数，数值越大效率越高，可在此处添加事物处理，

每次回滚数就是commit-interval数

Xml代码   ![收藏代码][icon_star.png]

1.  **<?****xml** version="1.0" encoding="UTF-8"**?>**  
2.  **<****beans** xmlns="http://www.springframework.org/schema/beans"  
3.      xmlns:batch="http://www.springframework.org/schema/batch"  
4.      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  
5.      xmlns:tx="http://www.springframework.org/schema/tx"  
6.      xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
7.      http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd  
8.      http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd  
9.      http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd"**>**  
10.   
11.     **<****import** resource="data-source-context.xml"**/>**  
12.       
13.       
14.     **<****bean** id="jobRepository"  
15.         class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"**>**  
16.         **<****property** name="transactionManager" ref="transactionManager" **/>**  
17.     **</****bean****>**  
18.       
19.       
20.     **<****bean** id="jobLauncher"  
21.         class="org.springframework.batch.core.launch.support.SimpleJobLauncher"**>**  
22.         **<****property** name="jobRepository" ref="jobRepository" **/>**  
23.     **</****bean****>**  
24.   
25.       
26.     **<****batch:job** id="ledgerJob"**>**  
27.         **<****batch:listeners****>**  
28.             **<****batch:listener** ref="appJobExecutionListener" **/>**  
29.         **</****batch:listeners****>**  
30.         **<****batch:step** id="step1"**>**

Xml代码   ![收藏代码][icon_star.png]

1.  **<****span** style="white-space: pre;"**>**    **</****span****>**   **<****batch:tasklet** transaction-manager="transactionManager"**>**  
2.          **<****batch:tasklet****>**  
3.              **<****batch:listeners****>**  
4.                  **<****batch:listener** ref="itemFailureLoggerListener" **/>**  
5.              **</****batch:listeners****>**  
6.              **<****batch:chunk** reader="ledgerReader" writer="ledgerWriter"  
7.                  commit-interval="10000" **/>**   
8.          **</****batch:tasklet****>**

Xml代码   ![收藏代码][icon_star.png]

1.                    **</****batch:tasklet****>**  
2.      **</****batch:step****>**  
3.  **</****batch:job****>**  
4.     
5.    
6.  **<****bean** id="ledgerReader"  
7.      class="org.springframework.batch.item.database.JdbcCursorItemReader"**>**  
8.      **<****property** name="dataSource" ref="dataSource" **/>**  
9.      **<****property** name="sql" value="select \* from ledger" **/>**   
10.     **<****property** name="rowMapper" ref="ledgerRowMapper" **/>**  
11. **</****bean****>**  
12.    
13.     
18. **<****bean** id="ledgerJobDetail" class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean"**>**    
19.     **<****property** name="targetObject"**>**    
20.             
21.         **<****ref** bean="quartzLedgerJob" **/>**    
22.     **</****property****>**    
23.     **<****property** name="targetMethod"**>**    
24.             
25.         **<****value****>**execute**</****value****>**    
26.     **</****property****>**    
27. **</****bean****>**    
28.    
29. **<****bean** id="ledgerCronTrigger" class="org.springframework.scheduling.quartz.CronTriggerBean" **>**    
30.         
31.     **<****property** name="jobDetail" **>**    
32.         **<****ref** bean="ledgerJobDetail" **/>**    
33.     **</****property****>**    
34.         
35.         
36.     **<****property** name="cronExpression" **>**    
37.           
38.         **<****value****>**0 45 10 \* \* ? \* **</****value****>**   
39.     **</****property****>**    
40.          
41. **</****bean****>**    
42.        
43.     
44. **<****bean** class="org.springframework.scheduling.quartz.SchedulerFactoryBean"**>**    
45.         
46.     **<****property** name="triggers"**>**    
47.         **<****list****>**    
48.                 
49.             **<****ref** local="ledgerCronTrigger" **/>**    
50.         **</****list****>**    
51.     **</****property****>**    
52. **</****bean****>**    
53.     
54. lt;/beans**>**

3.定时调度job类 QuartzLedgerJob.java

package net.etongbao.vasp.ac.quartz;

Java代码   ![收藏代码][icon_star.png]

1.  **import** java.util.Date;  
2.  **import** org.slf4j.Logger;  
3.  **import** org.slf4j.LoggerFactory;  
4.  **import** org.springframework.batch.core.Job;  
5.  **import** org.springframework.batch.core.JobParametersBuilder;  
6.  **import** org.springframework.batch.core.launch.JobLauncher;  
7.  **import** org.springframework.beans.factory.annotation.Autowired;  
8.  **import** org.springframework.stereotype.Component;  
9.  **import** org.springframework.util.StopWatch;  
10.   
11. /\*\* 
12.  \* 定时调度类 
13.  \* @author Fu Wei 
14.  \* 
15.  \*/  
16.   
17. @Component("quartzLedgerJob")  
18. **public** **class** QuartzLedgerJob \{  
19.   
20.     **private** **static** **final** Logger LOG = LoggerFactory.getLogger(QuartzLedgerJob.**class**);  
21.   
22.     @Autowired  
23.     **private** JobLauncher jobLauncher;  
24.   
25.     @Autowired  
26.     **private** Job ledgerJob;  
27.   
28.     @Autowired  
29.     JobParametersBuilder jobParameterBulider;  
30.   
31.     **private** **static** **long** counter = 0l;  
32.       
33.     /\*\* 
34.      \* 执行业务方法 
35.      \* @throws Exception 
36.      \*/  
37.     **public** **void** execute() **throws** Exception \{  
38.         LOG.debug("start...");  
39.         StopWatch sw = **new** StopWatch();  
40.         sw.start();  
41.         /\* 
42.          \* Spring Batch Job同一个job instance，成功执行后是不允许重新执行的【失败后是否允许重跑， 
43.          \* 可通过配置Job的restartable参数来控制，默认是true】，如果需要重新执行，可以变通处理， 
44.          \* 添加一个JobParameters构建类,以当前时间作为参数，保证其他参数相同的情况下却是不同的job instance 
45.          \*/  
46.         jobParameterBulider.addDate("date", **new** Date());  
47.         jobLauncher.run(ledgerJob, jobParameterBulider.toJobParameters());  
48.         sw.stop();  
49.         LOG.debug("Time elapsed:\{\},Execute quartz ledgerJob:\{\}", sw.prettyPrint(), ++counter);  
50.     \}  
51. \}

4.程序启动类 StartQuartz.java

package net.etongbao.vasp.ac.quartz;

Java代码   ![收藏代码][icon_star.png]

1.  **import** java.io.FileNotFoundException;  
2.    
3.  **import** org.springframework.context.support.ClassPathXmlApplicationContext;  
4.    
5.  /\*\* 
6.   \* 启动定时调度 
7.   \* @author Fu Wei 
8.   \* 
9.   \*/  
10. **public** **class** StartQuartz \{  
11.   
12.     **public** **static** **void** main(String\[\] args) **throws** FileNotFoundException \{  
13.   
14.         **new** ClassPathXmlApplicationContext("/net/etongbao/vasp/ac/resources/quartz-context.xml");  
15.     \}  
16. \}

5.pojo类 Ledger.java

Java代码   ![收藏代码][icon_star.png]

1.  **package** net.etongbao.vasp.ac.pojo;  
2.    
3.  **import** java.io.Serializable;  
4.  **import** java.util.Date;  
5.    
6.  **public** **class** Ledger **implements** Serializable \{  
7.    
8.      **private** **int** id;  
9.      **private** Date receiptDate;  
10.     **private** String memberName;  
11.     **private** String checkNumber;  
12.     **private** Date checkDate;  
13.     **private** String paymentType;  
14.     **private** **double** depositAmount;  
15.     **private** **double** paymentAmount;  
16.     **private** String comments;  
17.   
18.     **public** Ledger() \{  
19.         **super**();  
20.     \}  
21.   
22.     **public** Ledger(**int** id, Date receiptDate, String memberName, String checkNumber, Date checkDate, String paymentType,  
23.             **double** depositAmount, **double** paymentAmount, String comments) \{  
24.         **super**();  
25.         **this**.id = id;  
26.         **this**.receiptDate = receiptDate;  
27.         **this**.memberName = memberName;  
28.         **this**.checkNumber = checkNumber;  
29.         **this**.checkDate = checkDate;  
30.         **this**.paymentType = paymentType;  
31.         **this**.depositAmount = depositAmount;  
32.         **this**.paymentAmount = paymentAmount;  
33.         **this**.comments = comments;  
34.     \}  
35.   
36.     **public** **int** getId() \{  
37.         **return** id;  
38.     \}  
39.   
40.     **public** **void** setId(**int** id) \{  
41.         **this**.id = id;  
42.     \}  
43.   
44.     **public** Date getReceiptDate() \{  
45.         **return** receiptDate;  
46.     \}  
47.   
48.     **public** **void** setReceiptDate(Date receiptDate) \{  
49.         **this**.receiptDate = receiptDate;  
50.     \}  
51.   
52.     **public** String getMemberName() \{  
53.         **return** memberName;  
54.     \}  
55.   
56.     **public** **void** setMemberName(String memberName) \{  
57.         **this**.memberName = memberName;  
58.     \}  
59.   
60.     **public** String getCheckNumber() \{  
61.         **return** checkNumber;  
62.     \}  
63.   
64.     **public** **void** setCheckNumber(String checkNumber) \{  
65.         **this**.checkNumber = checkNumber;  
66.     \}  
67.   
68.     **public** Date getCheckDate() \{  
69.         **return** checkDate;  
70.     \}  
71.   
72.     **public** **void** setCheckDate(Date checkDate) \{  
73.         **this**.checkDate = checkDate;  
74.     \}  
75.   
76.     **public** String getPaymentType() \{  
77.         **return** paymentType;  
78.     \}  
79.   
80.     **public** **void** setPaymentType(String paymentType) \{  
81.         **this**.paymentType = paymentType;  
82.     \}  
83.   
84.     **public** **double** getDepositAmount() \{  
85.         **return** depositAmount;  
86.     \}  
87.   
88.     **public** **void** setDepositAmount(**double** depositAmount) \{  
89.         **this**.depositAmount = depositAmount;  
90.     \}  
91.   
92.     **public** **double** getPaymentAmount() \{  
93.         **return** paymentAmount;  
94.     \}  
95.   
96.     **public** **void** setPaymentAmount(**double** paymentAmount) \{  
97.         **this**.paymentAmount = paymentAmount;  
98.     \}  
99.   
100.     **public** String getComments() \{  
101.         **return** comments;  
102.     \}  
103.   
104.     **public** **void** setComments(String comments) \{  
105.         **this**.comments = comments;  
106.     \}  
107. \}

6. LedgerDaoImpl.java

package net.etongbao.vasp.ac.dao.impl;

Java代码   ![收藏代码][icon_star.png]

1.  **import** java.sql.PreparedStatement;  
2.  **import** java.sql.SQLException;  
3.  **import** net.etongbao.vasp.ac.dao.LedgerDao;  
4.  **import** net.etongbao.vasp.ac.pojo.Ledger;  
5.  **import** org.springframework.beans.factory.annotation.Autowired;  
6.  **import** org.springframework.jdbc.core.JdbcTemplate;  
7.  **import** org.springframework.jdbc.core.PreparedStatementSetter;  
8.  **import** org.springframework.stereotype.Repository;  
9.    
10. /\*\* 
11.  \* ledger数据操作类 
12.  \*  
13.  \* @author Fu Wei 
14.  \*  
15.  \*/  
16.   
17. @Repository  
18. **public** **class** LedgerDaoImpl **implements** LedgerDao \{  
19.   
20.     **private** **static** **final** String SAVE\_SQL = "insert into ledger\_temp (rcv\_dt, mbr\_nm, chk\_nbr, chk\_dt, pymt\_typ, dpst\_amt, pymt\_amt, comments) values(?,?,?,?,?,?,?,?)";  
21.   
22.     @Autowired  
23.     **private** JdbcTemplate jdbcTemplate;  
24.   
25.     @Override  
26.     **public** **void** save(**final** Ledger item) \{  
27.         jdbcTemplate.update(SAVE\_SQL, **new** PreparedStatementSetter() \{  
28.             **public** **void** setValues(PreparedStatement stmt) **throws** SQLException \{  
29.                 stmt.setDate(1, **new** java.sql.Date(item.getReceiptDate().getTime()));  
30.                 stmt.setString(2, item.getMemberName());  
31.                 stmt.setString(3, item.getCheckNumber());  
32.                 stmt.setDate(4, **new** java.sql.Date(item.getCheckDate().getTime()));  
33.                 stmt.setString(5, item.getPaymentType());  
34.                 stmt.setDouble(6, item.getDepositAmount());  
35.                 stmt.setDouble(7, item.getPaymentAmount());  
36.                 stmt.setString(8, item.getComments());  
37.             \}  
38.         \});  
39.     \}  
40.   
41. \}

7.接口 LedgerDao .java

Java代码   ![收藏代码][icon_star.png]

1.  **package** net.etongbao.vasp.ac.dao;  
2.    
3.  **import** net.etongbao.vasp.ac.pojo.Ledger;  
4.    
5.  **public** **interface** LedgerDao \{  
6.      **public** **void** save(**final** Ledger item) ;  
7.  \}

8. JdbcTemplete 需要的LedgerRowMapper.java

package net.etongbao.vasp.ac.batch.writer;

Java代码   ![收藏代码][icon_star.png]

1.  **import** java.sql.ResultSet;  
2.  **import** java.sql.SQLException;  
3.    
4.  **import** net.etongbao.vasp.ac.pojo.Ledger;  
5.    
6.  **import** org.springframework.jdbc.core.RowMapper;  
7.  **import** org.springframework.stereotype.Component;  
8.     
9.  /\*\* 
10.  \* ledger行的映射类 
11.  \* @author Administrator 
12.  \* 
13.  \*/  
14. @Component("ledgerRowMapper")  
15. **public** **class** LedgerRowMapper **implements** RowMapper \{  
16.     **public** Object mapRow(ResultSet rs, **int** rowNum) **throws** SQLException \{  
17.         Ledger ledger = **new** Ledger();  
18.         ledger.setId(rs.getInt("id"));  
19.         ledger.setReceiptDate(rs.getDate("rcv\_dt"));  
20.         ledger.setMemberName(rs.getString("mbr\_nm"));  
21.         ledger.setCheckNumber(rs.getString("chk\_nbr"));  
22.         ledger.setCheckDate(rs.getDate("chk\_dt"));  
23.         ledger.setPaymentType(rs.getString("pymt\_typ"));  
24.         ledger.setDepositAmount(rs.getDouble("dpst\_amt"));  
25.         ledger.setPaymentAmount(rs.getDouble("pymt\_amt"));  
26.         ledger.setComments(rs.getString("comments"));  
27.         **return** ledger;  
28.     \}  
29. \}

9.关键类LedgerWriter.java ，写入数据，负责数据的添加

Java代码   ![收藏代码][icon_star.png]

1.  **package** net.etongbao.vasp.ac.batch.writer;  
2.    
3.  **import** java.util.List;  
4.    
5.  **import** net.etongbao.vasp.ac.dao.LedgerDao;  
6.  **import** net.etongbao.vasp.ac.pojo.Ledger;  
7.    
8.  **import** org.springframework.batch.item.ItemWriter;  
9.  **import** org.springframework.beans.factory.annotation.Autowired;  
10. **import** org.springframework.stereotype.Component;  
11.   
12. /\*\* 
13.  \* ledger写入数据 
14.  \*  
15.  \* @author Fu Wei 
16.  \*  
17.  \*/  
18. @Component("ledgerWriter")  
19. **public** **class** LedgerWriter **implements** ItemWriter<Ledger> \{  
20.   
21.     @Autowired  
22.     **private** LedgerDao ledgerDao;  
23.   
24.     /\*\* 
25.      \* 写入数据 
26.      \*  
27.      \* @param ledgers 
28.      \*/  
29.     **public** **void** write(List<? **extends** Ledger> ledgers) **throws** Exception \{  
30.         **for** (Ledger ledger : ledgers) \{  
31.             ledgerDao.save(ledger);  
32.         \}  
33.     \}  
34.   
35. \}

classPath:

<?xml version="1.0" encoding="UTF-8"?>

Xml代码   ![收藏代码][icon_star.png]

1.  **<****classpath****>**  
2.      **<****classpathentry** kind="src" path="src"**/>**  
3.      **<****classpathentry** kind="con" path="org.eclipse.jdt.launching.JRE\_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jrockit-jdk1.6.0\_24-R28.1.3-4.0.1"**/>**  
4.      **<****classpathentry** kind="lib" path="lib/aopalliance-1.0.jar"**/>**  
5.      **<****classpathentry** kind="lib" path="lib/c3p0-0.9.1.2.jar"**/>**  
6.      **<****classpathentry** kind="lib" path="lib/commons-collections-3.2.1.jar"**/>**  
7.      **<****classpathentry** kind="lib" path="lib/commons-lang-2.3.jar"**/>**  
8.      **<****classpathentry** kind="lib" path="lib/commons-logging-1.1.1.jar"**/>**  
9.      **<****classpathentry** kind="lib" path="lib/etb-log4j-1.2.16.jar"**/>**  
10.     **<****classpathentry** kind="lib" path="lib/etb-slf4j-api-1.5.8.jar"**/>**  
11.     **<****classpathentry** kind="lib" path="lib/etb-slf4j-log4j12-1.5.8.jar"**/>**  
12.     **<****classpathentry** kind="lib" path="lib/ojdbc6.jar"**/>**  
13.     **<****classpathentry** kind="lib" path="lib/org.springframework.aop-3.0.5.RELEASE.jar"**/>**  
14.     **<****classpathentry** kind="lib" path="lib/org.springframework.asm-3.0.5.RELEASE.jar"**/>**  
15.     **<****classpathentry** kind="lib" path="lib/org.springframework.aspects-3.0.5.RELEASE.jar"**/>**  
16.     **<****classpathentry** kind="lib" path="lib/org.springframework.beans-3.0.5.RELEASE.jar"**/>**  
17.     **<****classpathentry** kind="lib" path="lib/org.springframework.context-3.0.5.RELEASE.jar"**/>**  
18.     **<****classpathentry** kind="lib" path="lib/org.springframework.context.support-3.0.5.RELEASE.jar"**/>**  
19.     **<****classpathentry** kind="lib" path="lib/org.springframework.core-3.0.5.RELEASE.jar"**/>**  
20.     **<****classpathentry** kind="lib" path="lib/org.springframework.expression-3.0.5.RELEASE.jar"**/>**  
21.     **<****classpathentry** kind="lib" path="lib/org.springframework.instrument-3.0.5.RELEASE.jar"**/>**  
22.     **<****classpathentry** kind="lib" path="lib/org.springframework.instrument.tomcat-3.0.5.RELEASE.jar"**/>**  
23.     **<****classpathentry** kind="lib" path="lib/org.springframework.jdbc-3.0.5.RELEASE.jar"**/>**  
24.     **<****classpathentry** kind="lib" path="lib/org.springframework.jms-3.0.5.RELEASE.jar"**/>**  
25.     **<****classpathentry** kind="lib" path="lib/org.springframework.orm-3.0.5.RELEASE.jar"**/>**  
26.     **<****classpathentry** kind="lib" path="lib/org.springframework.oxm-3.0.5.RELEASE.jar"**/>**  
27.     **<****classpathentry** kind="lib" path="lib/org.springframework.test-3.0.5.RELEASE.jar"**/>**  
28.     **<****classpathentry** kind="lib" path="lib/org.springframework.transaction-3.0.5.RELEASE.jar"**/>**  
29.     **<****classpathentry** kind="lib" path="lib/quartz-all-1.6.5.jar"**/>**  
30.     **<****classpathentry** kind="lib" path="lib/spring-batch-core-2.1.6.RELEASE.jar"**/>**  
31.     **<****classpathentry** kind="lib" path="lib/spring-batch-infrastructure-2.1.6.RELEASE.jar"**/>**  
32.     **<****classpathentry** kind="lib" path="lib/spring-batch-test-2.1.6.RELEASE.jar"**/>**  
33.     **<****classpathentry** kind="con" path="org.eclipse.jdt.junit.JUNIT\_CONTAINER/4"**/>**  
34.     **<****classpathentry** kind="output" path="bin"**/>**  
35. **</****classpath****>**

[icon_star.png]: /images/20220724/e88d33d9ff9c41bda7223c0b6f261e1c.png