Java 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. }