• 如何优雅的整合定时批量任务（荣耀典藏版）

     

    目录

    前言

    一、代码具体实现

    1、pom文件配置 

    2、application.yaml文件

    3、Service实现类

    4、SpringBatch配置类

    5、Processor设置

    6、封装实体Bean

    7、启动类上要加上注解

    总结

    前言

    最近赶一个紧急需求，需求内容如下：

    PC网页触发一条设备升级记录（下图），后台要定时批量设备更新。这里定时要用到Quartz，批量数据处理要用到SpringBatch，二者结合，可以完成该需求。

    由于之前，没有用过SpringBatch，于是上网查了下资料，发现可参考的不是很多，于是只能去慢慢的翻看官方文档。 

    https://docs.spring.io/spring-batch/4.1.x/reference/html

    遇到不少问题，就记录一下吧。

    一、代码具体实现

    1、pom文件配置 

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

    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
    12.  
    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/
    15.  
    16. # 每次批量处理的数据量，默认为5000
    17. batch-size: 5000

    3、Service实现类

    触发批处理任务的入口，执行一个job。

    1. @Service("batchService")
    2. public class BatchServiceImpl implements BatchService {
    3.  
    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 {
    3.  
    4.     private static final Logger log = LoggerFactory.getLogger(BatchConfiguration.class);
    5.  
    6.     @Value("${batch-size:5000}")
    7.     private int batchSize;
    8.  
    9. // 框架自动注入
    10.     @Autowired
    11.     public JobBuilderFactory jobBuilderFactory;
    12.  
    13. // 框架自动注入
    14.     @Autowired
    15.     public StepBuilderFactory stepBuilderFactory;
    16.  
    17. // 数据过滤器，对从数据库读出来的数据，注意进行操作
    18.     @Autowired
    19.     public TaskItemProcessor taskItemProcessor;
    20.  
    21.     // 接收job参数
    22.     public Map parameters;
    23.  
    24.     public Object taskId;
    25.  
    26.     @Autowired
    27.     private JdbcTemplate jdbcTemplate;
    28.  
    29. // 读取数据库操作
    30.     @Bean
    31.     @StepScope
    32.     public JdbcCursorItemReader itemReader(DataSource dataSource) {
    33.  
    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 = ?";
    55.  
    56.         return new JdbcCursorItemReaderBuilder()
    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.     }
    64.  
    65. // 将结果写回数据库
    66.     @Bean
    67.     @StepScope
    68.     public ItemWriter itemWriter() {
    69.         return new ItemWriter() {
    70.  
    71.             private int updateTaskStatus(DispatchRequest dispatchRequest, int status) {
    72.                 log.info("update taskId: {}, deviceId: {} to status {}", dispatchRequest.getTaskId(), dispatchRequest.getDeviceId(), status);
    73.  
    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.                 }
    86.  
    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.             }
    95.  
    96.             @Override
    97.             @Transactional
    98.             public void write(List 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.     }
    120.  
    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.     }
    133.  
    134.     /**
    135.      * 定义一个下发更新的 step
    136.      * @return
    137.      */
    138.     @Bean
    139.     public Step updateDeviceStep(JdbcCursorItemReader itemReader,ItemWriter itemWriter) {
    140.         return stepBuilderFactory.get(UUID.randomUUID().toString().replace("-", ""))
    141.                 . chunk(batchSize)
    142.                 .reader(itemReader) //根据taskId从数据库读取更新设备信息
    143.                 .processor(taskItemProcessor) // 每条更新信息，执行下发更新接口
    144.                 .writer(itemWriter)
    145.                 .build();
    146.     }
    147.  
    148.     // job 监听器
    149.     public class JobListener implements JobExecutionListener {
    150.  
    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.         }
    158.  
    159.         @Override
    160.         public void afterJob(JobExecution jobExecution) {
    161.  
    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.                     " ) ";
    176.  
    177.             // 获取更新失败的设备个数
    178.             Integer count = jdbcTemplate.queryForObject(sql, new Object[]{taskId}, Integer.class);
    179.  
    180.             log.info("update device failure count : " + count);
    181.  
    182.             // 下面是使用Quartz触发定时任务
    183.             // 获取任务时间,单位秒
    184. // String time = jdbcTemplate.queryForObject(sql, new Object[]{taskId}, Integer.class);
    185.             // 此处方便测试，应该从数据库中取taskId对应的重试间隔，单位秒
    186.             Integer millSecond = 10;
    187.  
    188.             if(count != null && count > 0){
    189.                 String jobName = "UpgradeTask_" + taskId;
    190.                 String reTaskId = taskId.toString();
    191.                 Map 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.             }
    199.  
    200.         }
    201.     }
    202. }

    5、Processor设置

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

    1. @Component("taskItemProcessor")
    2. public class TaskItemProcessor implements ItemProcessor {
    3.  
    4.     public static final int STATUS_DISPATCH_FAILED = 2;
    5.     public static final int STATUS_DISPATCH_SUCC = 1;
    6.  
    7.     private static final Logger log = LoggerFactory.getLogger(TaskItemProcessor.class);
    8.  
    9.     @Value("${upgrade-dispatch-base-url:http://localhost/api/v2/rpc/dispatch/command/}")
    10.     private String dispatchUrl;
    11.  
    12.     @Autowired
    13.     JdbcTemplate jdbcTemplate;
    14.  
    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();
    25.  
    26.         log.info("request url:" + url);
    27.         RestTemplate restTemplate = new RestTemplate();
    28.         HttpHeaders headers = new HttpHeaders();
    29.         headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
    30.  
    31.         MultiValueMap params = new LinkedMultiValueMap();
    32.  
    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());
    41.  
    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.         }
    49.  
    50.         log.info("request body json :" + jsonOuter);
    51.         HttpEntity requestEntity = new HttpEntity(jsonOuter.toString(),headers);
    52.         int status;
    53.         try {
    54.             ResponseEntity 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.             }
    61.  
    62.         }catch (Exception e){
    63.             status = STATUS_DISPATCH_FAILED;
    64.         }
    65.  
    66.         return new ProcessResult(dispatchRequest, status);
    67.     }
    68. }

    6、封装实体Bean

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

    1. public class DispatchRequest {
    2.  
    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;
    14.  
    15.    // 省略构造函数，setter/getter/tostring方法
    16.    //......
    17.    
    18.     public static class DispatchRequestRowMapper implements RowMapper {
    19.         @Override
    20.         public DispatchRequest mapRow(ResultSet resultSet, int i) throws SQLException {
    21.  
    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 {
    4.  
    5.     public static void main(String[] args) {
    6.         SpringApplication.run(Application.class, args);
    7.     }
    8. }

    总结

    其实SpringBatch并没有想象中那么好用，当从数据库中每次取5000条数据后，进入processor中是逐条处理的，这个时候不能不行操作，等5000条数据处理完之后，再一次性执行ItemWriter方法。

    在使用的过程中，最坑的地方是ItemReader和ItemWriter这两个地方，如何执行自定义的Sql，参考文中代码就行。

    至于Quartz定时功能，很简单，只要定时创建SpringBatch里面的Job，让这个job启动就好了。

    只有当你开始，你才会到达你的理想和目的地，只有当你努力，
    你才会获得辉煌的成功，只有当你播种，你才会有所收获。只有追求，
    才能尝到成功的味道，坚持在昨天叫立足，坚持在今天叫进取，坚持在明天叫成功。欢迎所有小伙伴们点赞+收藏！！！ 

     

  • 相关阅读:
    C++编译期循环获取变量类型
    C语言的静态库和的动态库
    Java 面试八股文有必要背吗？要背多久
    React进阶
    Java线程面试题
    Linux ARM平台开发系列讲解（platform平台子系统） 2.10.1 platform平台子系统介绍
    [vue3.x]实战问题--vue2.x升级vue3.x遇到的问题
    Vue.js+SpringBoot开发生活废品回收系统
    记录一次我虚拟机好不容易连上后的配置
    一文详解视觉Transformer模型压缩和加速策略(量化/低秩近似/蒸馏/剪枝)
  • 原文地址：https://blog.csdn.net/weixin_48321993/article/details/125907098