第六章 - 分布式定时任务框架ElasticJob之SpringBoot整合DataflowJob流式作业

逃离我推掉我的手 2023-10-03 20:37 57阅读 0赞

## 系列文章目录 ##

[第一章 - 分布式定时任务框架ElasticJob之JavaApi整合Simple作业][- _ElasticJob_JavaApi_Simple]  
[第二章 - 分布式定时任务框架ElasticJob之JavaApi整合DataflowJob作业][- _ElasticJob_JavaApi_DataflowJob]  
[第三章 - 分布式定时任务框架ElasticJob之Spring框架整合SimpleJob作业][- _ElasticJob_Spring_SimpleJob]  
[第四章 - 分布式定时任务框架ElasticJob之Spring框架整合DataflowJob作业][- _ElasticJob_Spring_DataflowJob]  
[第五章 - 分布式定时任务框架ElasticJob之SpringBoot整合SimpleJob作业（实战一）][- _ElasticJob_SpringBoot_SimpleJob]  
[第五章 - 分布式定时任务框架ElasticJob之SpringBoot整合SimpleJob作业（实战二)][- _ElasticJob_SpringBoot_SimpleJob 1]

--------------------

#### 文章目录 ####

*  系列文章目录
 *  前言
 *  一、新建MyDataflowJob
 *  二、新建注解@ElasticDataflowJob
 *  三、启动SpringBoot项目
 *  总结

--------------------

## 前言 ##

本章节介绍SpringBoot整合DataflowJob流式作业，这种类型的定时任务相对于SimpleJob来讲还是后者使用更频繁，这种作业我们了解就行，它主要新建任务类时需要继承DataflowJob接口，实现它的两个方法，一个是取数据的方法，一个是处理数据的方法，下文将详细说明一个流式作业的创建到运行。

--------------------

## 一、新建MyDataflowJob ##

> 需要继承DataflowJob需要指明要处理的数据类型，fetchData是取数据的方法，取完数据，给processData方法去执行，这种流式作业就是一个方法取数据，另一个方法处理数据。

![在这里插入图片描述][9743ce3a29ff42a7b974cb6e6b580e13.png]

> 先定义好等会需要用到的数据：

private List<Integer> list = new ArrayList<>();
    {
        
        list.add(0);
        list.add(1);
        list.add(2);
        list.add(3);
        list.add(4);
        list.add(5);
        list.add(6);
        list.add(7);
        list.add(8);
        list.add(9);
    }

> 抓取数据方法：

public List fetchData(ShardingContext shardingContext) {
        
        List<Integer> rtnList = new ArrayList<>();
        // 取数据规则： 数字 % 分片总数 == 当前分片项
        for (Integer index : list){
        
            if (index % shardingContext.getShardingTotalCount() == shardingContext.getShardingItem()){
        
                rtnList.add(index);
                // 每次只获取一个
                break;
            }
        }
        // 模拟数据抓取过程，耗时3s
        try {
        
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        
            throw new RuntimeException(e);
        }
        log.info("我是分片项：" + shardingContext.getShardingItem() + ", 我抓取的数据是：" + rtnList);
        return rtnList;
    }

> 处理数据方法：

public void processData(ShardingContext shardingContext, List data) {
        
        // 数据处理
        list.removeAll(data);
        // 模拟数据处理
        try {
        
            Thread.sleep(5000);
        } catch (InterruptedException e) {
        
            throw new RuntimeException(e);
        }
        log.info("我是分片项：" + shardingContext.getShardingItem() + ",我移除的数据是：" + data);
    }

## 二、新建注解@ElasticDataflowJob ##

> 在MyDataflowJob类上加上这个注解，streamingProcess是否流式处理

![在这里插入图片描述][44def4046e704067ae6f94af9ded8640.png]  
![在这里插入图片描述][a638065af2964dcfae848d04eb2da4cb.png]

> 新建DataflowJobAutoConfig–流式作业自动装配类，该类的意义也是通过反射将 DataflowJob作业自动注册进JobScheduler
> 
>  *  @Configuration : 定义一个配置类
>  *  @ConditionalOnBean(CoordinatorRegistryCenter.class) ：启动条件，发现协调注册中心后启动
>  *  @AutoConfigureAfter(ZookeeperAutoConfig.class) ：在zookeeper注册中心配置类之后启动

![在这里插入图片描述][b18d040c4fad43a49a78cd68e5912808.png]

> 注册进JobScheduler

public void initDataflow(){
        
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(ElasticDataflowJob.class);
        for (Map.Entry<String, Object> entry : beans.entrySet()){
        
            Object instance = entry.getValue();
            Class<?>[] interfaces = instance.getClass().getInterfaces();
            for (Class<?> superInterface : interfaces){
        
                // 如果继承的接口是DataflowJob,就将他注册进JobScheduler,实现自动启动
                if (superInterface == DataflowJob.class){
        
                    ElasticDataflowJob annotation = instance.getClass().getAnnotation(ElasticDataflowJob.class);
                    String jobName = annotation.jobName();
                    String cron = annotation.cron();
                    int shardingTotalCount = annotation.shardingTotalCount();
                    boolean overwrite = annotation.overwrite();
                    boolean streamingProcess = annotation.streamingProcess();
    
                    //第一步job核心配置
                    JobCoreConfiguration jcc = JobCoreConfiguration
                            .newBuilder(jobName, cron, shardingTotalCount)
                            .build();
    
                    //第二步job类型配置
                    JobTypeConfiguration jtc = new DataflowJobConfiguration(jcc,
                            instance.getClass().getCanonicalName(), streamingProcess);
    
                    //第三步job根的配置
                    LiteJobConfiguration ljc = LiteJobConfiguration
                            .newBuilder(jtc)
                            .overwrite(overwrite)
                            .build();
                    // 注册进JobScheduler
                    new JobScheduler(zkCenter, ljc).init();
                }
            }
        }
    }

## 三、启动SpringBoot项目 ##

> DataflowJob类型定时任务已经跑起来了

![在这里插入图片描述][ff9d65712bc5445ba9722f2b7531970a.png]  
此外需要注意的是，我们也要将DataflowAutoConfig类配置到spring.factories配置文件中，让spring发现它。  
![在这里插入图片描述][eb78e18341ca4a66b0c864e8d635324d.png]

--------------------

## 总结 ##

SpringBoot整合流式作业就介绍到这儿，比较简单，代码点击[此处][Link 1]下载

[- _ElasticJob_JavaApi_Simple]: https://blog.csdn.net/qq_40187702/article/details/118344565
[- _ElasticJob_JavaApi_DataflowJob]: https://blog.csdn.net/qq_40187702/article/details/120108006
[- _ElasticJob_Spring_SimpleJob]: https://blog.csdn.net/qq_40187702/article/details/124111195
[- _ElasticJob_Spring_DataflowJob]: https://blog.csdn.net/qq_40187702/article/details/124697053
[- _ElasticJob_SpringBoot_SimpleJob]: https://blog.csdn.net/qq_40187702/article/details/127200584?spm=1001.2014.3001.5501
[- _ElasticJob_SpringBoot_SimpleJob 1]: https://blog.csdn.net/qq_40187702/article/details/127216222?spm=1001.2014.3001.5501
[9743ce3a29ff42a7b974cb6e6b580e13.png]: https://img-blog.csdnimg.cn/9743ce3a29ff42a7b974cb6e6b580e13.png
[44def4046e704067ae6f94af9ded8640.png]: https://img-blog.csdnimg.cn/44def4046e704067ae6f94af9ded8640.png
[a638065af2964dcfae848d04eb2da4cb.png]: https://img-blog.csdnimg.cn/a638065af2964dcfae848d04eb2da4cb.png
[b18d040c4fad43a49a78cd68e5912808.png]: https://img-blog.csdnimg.cn/b18d040c4fad43a49a78cd68e5912808.png
[ff9d65712bc5445ba9722f2b7531970a.png]: https://img-blog.csdnimg.cn/ff9d65712bc5445ba9722f2b7531970a.png
[eb78e18341ca4a66b0c864e8d635324d.png]: https://img-blog.csdnimg.cn/eb78e18341ca4a66b0c864e8d635324d.png
[Link 1]: https://gitee.com/climb_high/elastic-job.git