如何使用 SpringBoot SpringBatch Quartz 来整合定时批量任务呢 ？

1、pom文件

1. <dependencies>
2. <dependency>
3.   <groupId>org.springframework.boot</groupId>
4.   <artifactId>spring-boot-starter-web</artifactId>
5. </dependency>
6. <dependency>
7.   <groupId>org.postgresql</groupId>
8.   <artifactId>postgresql</artifactId>
9. </dependency>
10. <dependency>
11.   <groupId>org.springframework.boot</groupId>
12.   <artifactId>spring-boot-starter-jdbc</artifactId>
13. </dependency>
14. <dependency>
15.   <groupId>org.springframework.boot</groupId>
16.   <artifactId>spring-boot-starter-batch</artifactId>
17. </dependency>
18. <dependency>
19.   <groupId>org.projectlombok</groupId>
20.   <artifactId>lombok</artifactId>
21. </dependency>
22. <dependency>
23.   <groupId>org.springframework.boot</groupId>
24.   <artifactId>spring-boot-starter-batch</artifactId>
25. </dependency>
26. </dependencies>

2、application.yaml文件

1. spring:
2.  datasource:
3.    username: thinklink
4.    password: thinklink
5.    url: jdbc:postgresql://172.16.205.54:5432/thinklink
6.    driver-class-name: org.postgresql.Driver
7.  batch:
8.    job:
9.      enabled: false
10. server:
11.  port: 8073
13. #upgrade-dispatch-base-url: http://172.16.205.125:8080/api/rpc/dispatch/command/
14. upgrade-dispatch-base-url: http://172.16.205.211:8080/api/noauth/rpc/dispatch/command/
16. # 每次批量处理的数据量，默认为5000
17. batch-size: 5000

3、Service实现类，触发批处理任务的入口，执行一个job

1. @Service("batchService")
2. public class BatchServiceImpl implements BatchService {
4. // 框架自动注入
5.    @Autowired
6.    private JobLauncher jobLauncher;
7.    @Autowired
8.    private Job updateDeviceJob;
9.    /**
10.     * 根据 taskId 创建一个Job
11.     * @param taskId
12.     * @throws Exception
13.     */
14.    @Override
15.    public void createBatchJob(String taskId) throws Exception {
16.        JobParameters jobParameters = new JobParametersBuilder()
17.                .addString("taskId", taskId)
18.                .addString("uuid", UUID.randomUUID().toString().replace("-",""))
19.                .toJobParameters();
20.        // 传入一个Job任务和任务需要的参数
21.        jobLauncher.run(updateDeviceJob, jobParameters);
22.    }
23. }

4、SpringBatch配置类，此部分最重要

1. @Configuration
2. public class BatchConfiguration {
4.    private static final Logger log = LoggerFactory.getLogger(BatchConfiguration.class);
6.    @Value("${batch-size:5000}")
7.    private int batchSize;
9. // 框架自动注入
10.    @Autowired
11.    public JobBuilderFactory jobBuilderFactory;
13. // 框架自动注入
14.    @Autowired
15.    public StepBuilderFactory stepBuilderFactory;
17. // 数据过滤器，对从数据库读出来的数据，注意进行操作
18.    @Autowired
19.    public TaskItemProcessor taskItemProcessor;
21.    // 接收job参数
22.    public Map<String, JobParameter> parameters;
24.    public Object taskId;
26.    @Autowired
27.    private JdbcTemplate jdbcTemplate;
29. // 读取数据库操作
30.    @Bean
31.    @StepScope
32.    public JdbcCursorItemReader<DispatchRequest> itemReader(DataSource dataSource) {
34.        String querySql = " SELECT " +
35.                " e. ID AS taskId, " +
36.                " e.user_id AS userId, " +
37.                " e.timing_startup AS startTime, " +
38.                " u.device_id AS deviceId, " +
39.                " d.app_name AS appName, " +
40.                " d.compose_file AS composeFile, " +
41.                " e.failure_retry AS failureRetry, " +
42.                " e.tetry_times AS retryTimes, " +
43.                " e.device_managered AS deviceManagered " +
44.                " FROM " +
45.                " eiot_upgrade_task e " +
46.                " LEFT JOIN eiot_upgrade_device u ON e. ID = u.upgrade_task_id " +
47.                " LEFT JOIN eiot_app_detail d ON e.app_id = d. ID " +
48.                " WHERE " +
49.                " ( " +
50.                " u.device_upgrade_status = 0 " +
51.                " OR u.device_upgrade_status = 2" +
52.                " )" +
53.                " AND e.tetry_times > u.retry_times " +
54.                " AND e. ID = ?";
56.        return new JdbcCursorItemReaderBuilder<DispatchRequest>()
57.                .name("itemReader")
58.                .sql(querySql)
59.                .dataSource(dataSource)
60.                .queryArguments(new Object[]{parameters.get("taskId").getValue()})
61.                .rowMapper(new DispatchRequest.DispatchRequestRowMapper())
62.                .build();
63.    }
65. // 将结果写回数据库
66.    @Bean
67.    @StepScope
68.    public ItemWriter<ProcessResult> itemWriter() {
69.        return new ItemWriter<ProcessResult>() {
71.            private int updateTaskStatus(DispatchRequest dispatchRequest, int status) {
72.                log.info("update taskId: {}, deviceId: {} to status {}", dispatchRequest.getTaskId(), dispatchRequest.getDeviceId(), status);
74.                Integer retryTimes = jdbcTemplate.queryForObject(
75.                        "select retry_times from eiot_upgrade_device where device_id = ? and upgrade_task_id = ?",
76.                        new Object[]{ dispatchRequest.getDeviceId(), dispatchRequest.getTaskId()}, Integer.class
77.                );
78.                retryTimes += 1;
79.                int updateCount = jdbcTemplate.update("update eiot_upgrade_device set device_upgrade_status = ?, retry_times = ? " +
80.                        "where device_id = ? and upgrade_task_id = ?", status, retryTimes, dispatchRequest.getDeviceId(), dispatchRequest.getTaskId());
81.                if (updateCount <= 0) {
82.                    log.warn("no task updated");
83.                } else {
84.                    log.info("count of {} task updated", updateCount);
85.                }
87.                // 最后一次重试
88.                if (status == STATUS_DISPATCH_FAILED && retryTimes == dispatchRequest.getRetryTimes()) {
89.                    log.info("the last retry of {} failed, inc deviceManagered", dispatchRequest.getTaskId());
90.                    return 1;
91.                } else {
92.                    return 0;
93.                }
94.            }
96.            @Override
97.            @Transactional
98.            public void write(List<? extends ProcessResult> list) throws Exception {
99.                Map taskMap = jdbcTemplate.queryForMap(
100.                        "select device_managered, device_count, task_status from eiot_upgrade_task where id = ?",
101.                        list.get(0).getDispatchRequest().getTaskId() // 我们认定一个批量里面，taskId都是一样的
102.                        );
103.                int deviceManagered = (int)taskMap.get("device_managered");
104.                Integer deviceCount = (Integer) taskMap.get("device_count");
105.                if (deviceCount == null) {
106.                    log.warn("deviceCount of task {} is null", list.get(0).getDispatchRequest().getTaskId());
107.                }
108.                int taskStatus = (int)taskMap.get("task_status");
109.                for (ProcessResult result: list) {
110.                    deviceManagered += updateTaskStatus(result.getDispatchRequest(), result.getStatus());
111.                }
112.                if (deviceCount != null && deviceManagered == deviceCount) {
113.                    taskStatus = 2; //任务状态 0:待升级，1:升级中，2:已完成
114.                }
115.                jdbcTemplate.update("update eiot_upgrade_task set device_managered = ?, task_status = ? " +
116.                        "where id = ?", deviceManagered, taskStatus, list.get(0).getDispatchRequest().getTaskId());
117.            }
118.        };
119.    }
121.    /**
122.     * 定义一个下发更新的 job
123.     * @return
124.     */
125.    @Bean
126.    public Job updateDeviceJob(Step updateDeviceStep) {
127.        return jobBuilderFactory.get(UUID.randomUUID().toString().replace("-", ""))
128.                .listener(new JobListener()) // 设置Job的监听器
129.                .flow(updateDeviceStep)// 执行下发更新的Step
130.                .end()
131.                .build();
132.    }
134.    /**
135.     * 定义一个下发更新的 step
136.     * @return
137.     */
138.    @Bean
139.    public Step updateDeviceStep(JdbcCursorItemReader<DispatchRequest> itemReader,ItemWriter<ProcessResult> itemWriter) {
140.        return stepBuilderFactory.get(UUID.randomUUID().toString().replace("-", ""))
141.                .<DispatchRequest, ProcessResult> chunk(batchSize)
142.                .reader(itemReader) //根据taskId从数据库读取更新设备信息
143.                .processor(taskItemProcessor) // 每条更新信息，执行下发更新接口
144.                .writer(itemWriter)
145.                .build();
146.    }
148.    // job 监听器
149.    public class JobListener implements JobExecutionListener {
151.        @Override
152.        public void beforeJob(JobExecution jobExecution) {
153.            log.info(jobExecution.getJobInstance().getJobName() + " before... ");
154.            parameters = jobExecution.getJobParameters().getParameters();
155.            taskId = parameters.get("taskId").getValue();
156.            log.info("job param taskId : " + parameters.get("taskId"));
157.        }
159.        @Override
160.        public void afterJob(JobExecution jobExecution) {
162.            log.info(jobExecution.getJobInstance().getJobName() + " after... ");
163.            // 当所有job执行完之后，查询设备更新状态，如果有失败，则要定时重新执行job
164.            String sql = " SELECT " +
165.                    " count(*) " +
166.                    " FROM " +
167.                    " eiot_upgrade_device d " +
168.                    " LEFT JOIN eiot_upgrade_task u ON d.upgrade_task_id = u. ID " +
169.                    " WHERE " +
170.                    " u. ID = ? " +
171.                    " AND d.retry_times < u.tetry_times " +
172.                    " AND ( " +
173.                    " d.device_upgrade_status = 0 " +
174.                    " OR d.device_upgrade_status = 2 " +
175.                    " ) ";
177.            // 获取更新失败的设备个数
178.            Integer count = jdbcTemplate.queryForObject(sql, new Object[]{taskId}, Integer.class);
180.            log.info("update device failure count : " + count);
182.            // 下面是使用Quartz触发定时任务
183.            // 获取任务时间,单位秒
184. // String time = jdbcTemplate.queryForObject(sql, new Object[]{taskId}, Integer.class);
185.            // 此处方便测试，应该从数据库中取taskId对应的重试间隔，单位秒
186.            Integer millSecond = 10;
188.            if(count != null && count > 0){
189.                String jobName = "UpgradeTask_" + taskId;
190.                String reTaskId = taskId.toString();
191.                Map<String,Object> params = new HashMap<>();
192.                params.put("jobName",jobName);
193.                params.put("taskId",reTaskId);
194.                if (QuartzManager.checkNameNotExist(jobName))
195.                {
196.                    QuartzManager.scheduleRunOnceJob(jobName, RunOnceJobLogic.class,params,millSecond);
197.                }
198.            }
200.        }
201.    }
202. }

5、Processor，处理每条数据，可以在此对数据进行过滤操作

1. @Component("taskItemProcessor")
2. public class TaskItemProcessor implements ItemProcessor<DispatchRequest, ProcessResult> {
4.    public static final int STATUS_DISPATCH_FAILED = 2;
5.    public static final int STATUS_DISPATCH_SUCC = 1;
7.    private static final Logger log = LoggerFactory.getLogger(TaskItemProcessor.class);
9.    @Value("${upgrade-dispatch-base-url:http://localhost/api/v2/rpc/dispatch/command/}")
10.    private String dispatchUrl;
12.    @Autowired
13.    JdbcTemplate jdbcTemplate;
15.    /**
16.     * 在这里，执行 下发更新指令 的操作
17.     * @param dispatchRequest
18.     * @return
19.     * @throws Exception
20.     */
21.    @Override
22.    public ProcessResult process(final DispatchRequest dispatchRequest) {
23.        // 调用接口，下发指令
24.        String url = dispatchUrl + dispatchRequest.getDeviceId()+"/"+dispatchRequest.getUserId();
26.        log.info("request url:" + url);
27.        RestTemplate restTemplate = new RestTemplate();
28.        HttpHeaders headers = new HttpHeaders();
29.        headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
31.        MultiValueMap<String, String> params = new LinkedMultiValueMap<String, String>();
33.        JSONObject jsonOuter = new JSONObject();
34.        JSONObject jsonInner = new JSONObject();
35.        try {
36.            jsonInner.put("jobId",dispatchRequest.getTaskId());
37.            jsonInner.put("name",dispatchRequest.getName());
38.            jsonInner.put("composeFile", Base64Util.bytesToBase64Str(dispatchRequest.getComposeFile()));
39.            jsonInner.put("policy",new JSONObject().put("startTime",dispatchRequest.getPolicy()));
40.            jsonInner.put("timestamp",dispatchRequest.getTimestamp());
42.            jsonOuter.put("method","updateApp");
43.            jsonOuter.put("params",jsonInner);
44.        } catch (JSONException e) {
45.            log.info("JSON convert Exception :" + e);
46.        }catch (IOException e) {
47.            log.info("Base64Util bytesToBase64Str :" + e);
48.        }
50.        log.info("request body json :" + jsonOuter);
51.        HttpEntity<String> requestEntity = new HttpEntity<String>(jsonOuter.toString(),headers);
52.        int status;
53.        try {
54.            ResponseEntity<String> response = restTemplate.postForEntity(url,requestEntity,String.class);
55.            log.info("response :" + response);
56.            if (response.getStatusCode() == HttpStatus.OK) {
57.                status = STATUS_DISPATCH_SUCC;
58.            } else {
59.                status = STATUS_DISPATCH_FAILED;
60.            }
62.        }catch (Exception e){
63.            status = STATUS_DISPATCH_FAILED;
64.        }
66.        return new ProcessResult(dispatchRequest, status);
67.    }
68. }

6、封装数据库返回数据的实体Bean，注意静态内部类

1. public class DispatchRequest {
3.    private String taskId;
4.    private String deviceId;
5.    private String userId;
6.    private String name;
7.    private byte[] composeFile;
8.    private String policy;
9.    private String timestamp;
10.    private String md5;
11.    private int failureRetry;
12.    private int retryTimes;
13.    private int deviceManagered;
15.   // 省略构造函数，setter/getter/tostring方法
16.   //......
18.    public static class DispatchRequestRowMapper implements RowMapper<DispatchRequest> {
19.        @Override
20.        public DispatchRequest mapRow(ResultSet resultSet, int i) throws SQLException {
22.            DispatchRequest dispatchRequest = new DispatchRequest();
23.            dispatchRequest.setTaskId(resultSet.getString("taskId"));
24.            dispatchRequest.setUserId(resultSet.getString("userId"));
25.            dispatchRequest.setPolicy(resultSet.getString("startTime"));
26.            dispatchRequest.setDeviceId(resultSet.getString("deviceId"));
27.            dispatchRequest.setName(resultSet.getString("appName"));
28.            dispatchRequest.setComposeFile(resultSet.getBytes("composeFile"));
29.            dispatchRequest.setTimestamp(DateUtil.DateToString(new Date()));
30.            dispatchRequest.setRetryTimes(resultSet.getInt("retryTimes"));
31.            dispatchRequest.setFailureRetry(resultSet.getInt("failureRetry"));
32.            dispatchRequest.setDeviceManagered(resultSet.getInt("deviceManagered"));
33.            return dispatchRequest;
34.        }
35.    }
36. }

7、启动类上要加上注解

1. @SpringBootApplication
2. @EnableBatchProcessing
3. public class Application {
5.    public static void main(String[] args) {
6.        SpringApplication.run(Application.class, args);
7.    }
8. }