配置DB Job Repository 项目举例(三)

一、项目创建步骤

1.项目结构

BatchMain.java:

package com.xj.demo2;

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 BatchMain {
    public static void main(String[] args) {

        ApplicationContext context = new ClassPathXmlApplicationContext("demo2/job/demo2-job.xml");
        //Spring Batch的作业启动器,
        JobLauncher launcher = (JobLauncher) context.getBean("jobLauncher");
        //在batch.xml中配置的一个作业
        Job job  = (Job)context.getBean("billJob");

        try{
            //开始执行这个作业,获得处理结果(要运行的job,job参数对象)
            JobExecution result = launcher.run(job, new JobParameters());
            System.out.println(result.toString());
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

CreditBill.java:

package com.xj.demo2;


public class CreditBill {
    //银行卡账户ID
    private String accountID = "";
    //持卡人姓名
    private String name = "";
    //消费金额
    private double amount = 0;
    //消费日期
    private String date = "";
    //消费场所
    private String address = "";

    public String getAccountID() {
        return accountID;
    }

    public void setAccountID(String accountID) {
        this.accountID = accountID;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public double getAmount() {
        return amount;
    }

    public void setAmount(double amount) {
        this.amount = amount;
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    @Override
    public String toString() {
        return this.accountID + "," + this.name + "," + this.amount + "," + this.date + "," + this.address;
    }
}

CreditBillProcessor:

package com.xj.demo2;

import org.springframework.batch.item.ItemProcessor;


public class CreditBillProcessor implements ItemProcessor {
    @Override
    public CreditBill process(CreditBill bill) throws Exception {

        System.out.println(bill.toString());
        //做一些简单的处理
        bill.setAccountID(bill.getAccountID() + "1");
        bill.setName(bill.getName() + "2");
        bill.setAmount(bill.getAmount() + 3);
        bill.setDate(bill.getDate() + "4");
        bill.setAddress(bill.getAddress() + 5);

        return bill;
    }
}

demo2-inputFile.csv:

4101231234656,tom,100.00,2013-12-31 12:00:08,Lu lit
4101236543210,tom,120.00,2013-12-31 12:00:08,Lu Zui

demo2-job.xml:

demo2-jobContext.xml:




    
    

    
    
        
    

    
    
        
            com.mysql.jdbc.Driver
        
        
            jdbc:mysql://127.0.0.1:3306/spring_batch_demo2
        
        
        
    

    
    
        
        
    

    
    

    
    
    
        
        
        
        
            
                
                
                
                
                    
                        
                    
                
            
        
    

    
    
        
        
        
        
            
                accountID
                name
                amount
                date
                address

二、配套数据库建立

1.寻找对应的数据库脚本语句

使用数据库的仓库时，需要首先根据Spring Batch框架提供的数据脚本完成数据库的初始化，数据库脚本的位置放在:

spring-batch-core3.0.7.RELEASEspring-batch-core-3.0.7.RELEASE.jarorgspringframeworkbatchcore

Spring Batch框架JobRepository支持如下的数据库：DB2、Derby、H2、HSQLDB、MySQL、Oracle、PostgreSQL、SQLServer、Sybase。

因为我用的是MySQL数据库，所以我选择:schema-mysql.sql

schema-mysql.sql:

-- Autogenerated: do not edit this file

CREATE TABLE BATCH_JOB_INSTANCE  (
	JOB_INSTANCE_ID BIGINT  NOT NULL PRIMARY KEY ,
	VERSION BIGINT ,
	JOB_NAME VARCHAr(100) NOT NULL,
	JOB_KEY VARCHAr(32) NOT NULL,
	constraint JOB_INST_UN unique (JOB_NAME, JOB_KEY)
) ENGINE=InnoDB;

CREATE TABLE BATCH_JOB_EXECUTION  (
	JOB_EXECUTION_ID BIGINT  NOT NULL PRIMARY KEY ,
	VERSION BIGINT  ,
	JOB_INSTANCE_ID BIGINT NOT NULL,
	CREATE_TIME DATETIME NOT NULL,
	START_TIME DATETIME DEFAULT NULL ,
	END_TIME DATETIME DEFAULT NULL ,
	STATUS VARCHAr(10) ,
	EXIT_CODE VARCHAr(2500) ,
	EXIT_MESSAGE VARCHAr(2500) ,
	LAST_UPDATeD DATETIME,
	JOB_CONFIGURATION_LOCATION VARCHAr(2500) NULL,
	constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
	references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
) ENGINE=InnoDB;

CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
	JOB_EXECUTION_ID BIGINT NOT NULL ,
	TYPE_CD VARCHAr(6) NOT NULL ,
	KEY_NAME VARCHAr(100) NOT NULL ,
	STRING_VAL VARCHAr(250) ,
	DATE_VAL DATETIME DEFAULT NULL ,
	LONG_VAL BIGINT ,
	DOUBLE_VAL DOUBLE PRECISION ,
	IDENTIFYING CHAr(1) NOT NULL ,
	constraint JOB_EXEC_PARAMS_FK foreign key (JOB_EXECUTION_ID)
	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ENGINE=InnoDB;

CREATE TABLE BATCH_STEP_EXECUTION  (
	STEP_EXECUTION_ID BIGINT  NOT NULL PRIMARY KEY ,
	VERSION BIGINT NOT NULL,
	STEP_NAME VARCHAr(100) NOT NULL,
	JOB_EXECUTION_ID BIGINT NOT NULL,
	START_TIME DATETIME NOT NULL ,
	END_TIME DATETIME DEFAULT NULL ,
	STATUS VARCHAr(10) ,
	COMMIT_COUNT BIGINT ,
	READ_COUNT BIGINT ,
	FILTER_COUNT BIGINT ,
	WRITE_COUNT BIGINT ,
	READ_SKIP_COUNT BIGINT ,
	WRITE_SKIP_COUNT BIGINT ,
	PROCESS_SKIP_COUNT BIGINT ,
	ROLLBACK_COUNT BIGINT ,
	EXIT_CODE VARCHAr(2500) ,
	EXIT_MESSAGE VARCHAr(2500) ,
	LAST_UPDATED DATETIME,
	constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)
	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ENGINE=InnoDB;

CREATE TABLE BATCH_STEP_EXECUTION_ConTEXT  (
	STEP_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY,
	SHORT_ConTEXT VARCHAr(2500) NOT NULL,
	SERIALIZED_ConTEXT TEXT ,
	constraint STEP_EXEC_CTX_FK foreign key (STEP_EXECUTION_ID)
	references BATCH_STEP_EXECUTION(STEP_EXECUTION_ID)
) ENGINE=InnoDB;

CREATE TABLE BATCH_JOB_EXECUTION_ConTEXT  (
	JOB_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY,
	SHORT_ConTEXT VARCHAr(2500) NOT NULL,
	SERIALIZED_ConTEXT TEXT ,
	constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ENGINE=InnoDB;

CREATE TABLE BATCH_STEP_EXECUTION_SEQ (
	ID BIGINT NOT NULL,
	UNIQUE_KEY CHAr(1) NOT NULL,
	constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
) ENGINE=InnoDB;

INSERT INTO BATCH_STEP_EXECUTION_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_STEP_EXECUTION_SEQ);

CREATE TABLE BATCH_JOB_EXECUTION_SEQ (
	ID BIGINT NOT NULL,
	UNIQUE_KEY CHAr(1) NOT NULL,
	constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
) ENGINE=InnoDB;

INSERT INTO BATCH_JOB_EXECUTION_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_JOB_EXECUTION_SEQ);

CREATE TABLE BATCH_JOB_SEQ (
	ID BIGINT NOT NULL,
	UNIQUE_KEY CHAr(1) NOT NULL,
	constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
) ENGINE=InnoDB;

INSERT INTO BATCH_JOB_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_JOB_SEQ);

2.创建数据库，运行schema-mysql.sql中的脚本

3.各个表的关系

（1）各个表的结构关系

（2）instance、execution和context之间的关系

1>一个Job可以拥有一到多个Step

2>一个Step可以有一到多个Step Execution（当一个Step执行失败，下次重新执行该任务时，会为该Step重新生成一个Step Execution）

3>一个Job Execution可以有一到多个Step Execution(当一个Job由多个Step组成时，每个Step执行都会生成一个新的Step Execution，则一个Job Execution会拥有多个Step Execution)

4>一个Job Execution对应一个Job Execution Context

5>每个Step Execution对应一个Step Execution Context

6>同一个Job中的Step Executionton共用Job Execution Context，因此如果同一个Job的不同Step间需要共享数据时，则可以通过Job Execution的上下文共享数据。

三、配套数据库建立

1.运行结果：

控制台：

demo2-outputFile.csv:

batch_job_instance:

batch_job_execution:

batch_step_execution:

四、一些问题总结

1.当程序第一次运行成功之后，demo2-outputFile文件被成功写入，再重新将demo2-outputFile文件清空，再次运行程序时，发现程序运行成功，没有报错，但是demo2-outputFile文件里却没有被写入，这是为什么呢？

查看batch_job_execution表发现这样的信息:

大概意思是：所有的steps已经执行完毕了，或者该job没有配置step。

查看batch_step_execution表：

分析：

当执行一个Job的时候，就会生成一个Job Instance，然后Job Instance会生成一个Job Execution，并根据step的id即“billStep”生成对应的Step Execution时发现该step的状态是“COMPLETED”即已经执行完毕了。这时就不执行该id为“billStep”的step了，直接在batch_job_execution中记录一条信息，表示没有可执行的step。

解决方法：

在batch_step_execution中删除执行成功的step的信息，因为涉及到外键，所以先要把batch_step_execution_context中对应的信息删除，执行脚本：

DELETe FROM batch_step_execution_context WHERe STEP_EXECUTION_ID = "1";
DELETe FROM batch_step_execution WHERe STEP_EXECUTION_ID = "1";

再执行程序，就会发现又可以写入了：

控制台：

其他解决方法：

1>更改Step的id名称，即将“billStep”更换成其他名称也可以执行。

2>更改Job的名称,即将“billJob”更换成其他名称也可以执行。

3>若不想更改Job的名称，可以为“billJob”添加参数也行，毕竟Job Instance = Job Name + Job Parameters。

见下面代码：

BatchMain.java:

package com.xj.demo2;

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


public class BatchMain {
    public static void main(String[] args) {

        ApplicationContext context = new ClassPathXmlApplicationContext("demo2/job/demo2-job.xml");
        //Spring Batch的作业启动器,
        JobLauncher launcher = (JobLauncher) context.getBean("jobLauncher");
        //在batch.xml中配置的一个作业
        Job job  = (Job)context.getBean("billJob");

        try{
            //开始执行这个作业,获得处理结果(要运行的job,job参数对象)
            JobExecution result = launcher.run(job, new JobParametersBuilder().addString("date","20211108").toJobParameters());
            System.out.println(result.toString());
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

可以看出，其中为billJob新增了String类型的参数“date”:“20211108”：

//开始执行这个作业,获得处理结果(要运行的job,job参数对象)
JobExecution result = launcher.run(job, new JobParametersBuilder().addString("date","20211108").toJobParameters());

运行程序后，查看数据库：

batch_job_instance表:

batch_job_execution表：

batch_step_execution表：

有一点儿需要说明，采用Job Instance = Job Name + Job Parameters这种方式后，若第一次Job已经运行成功了，在不改变Job Name和Job Parameters时，再次运行程序，会报错：

org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException: A job instance already exists and is complete for parameters={date=20211108}. If you want to run this job again, change the parameters.

2.删除数据库的脚本

若是数据建立之后，想要清空数据表或者删除数据表时，因为有外键相关联，所以操作表需要一定的顺序：

清空表：

DELETe FROM batch_job_execution_context;
DELETe FROM batch_job_execution_params;
DELETe FROM batch_job_execution_seq;
DELETe FROM batch_step_execution_context;
DELETe FROM batch_step_execution;
DELETe FROM batch_job_execution;
DELETe FROM batch_job_seq;
DELETe FROM batch_step_execution_seq;
DELETe FROM batch_job_instance;

删除表：

DROP TABLE batch_job_execution_context;
DROP TABLE batch_job_execution_params;
DROP TABLE batch_job_execution_seq;
DROP TABLE batch_step_execution_context;
DROP TABLE batch_step_execution;
DROP TABLE batch_job_execution;
DROP TABLE batch_job_seq;
DROP TABLE batch_step_execution_seq;
DROP TABLE batch_job_instance;

配置DB Job Repository 项目举例(三)

Java相关栏目本月热门文章