springboot整合elasticJob实战(纯代码开发三种任务类型用法)以及分片系统,事件追踪详解

一 springboot整合

介绍就不多说了,只有这个框架是当当网开源的,支持分布式调度,分布式系统中非常合适(两个服务同时跑不会重复,并且可灵活配置分开分批处理数据,贼方便)!

这里主要还是用到zookeeper,如果没有zk环境,可以百度或者参考我之前的博客搭建

添加依赖,这里有一点,如果是在springcloud中的话,需要排除自带的curator依赖,因为cloud已经集成一些,会冲突:

  <!-- elastic-job -->

         <dependency>

             <groupId>com.dangdang</groupId>

             <artifactId>elastic-job-lite-core</artifactId>

             <version>2.1.5</version>

             <exclusions>

                 <exclusion>

                     <artifactId>curator-client</artifactId>

                     <groupId>org.apache.curator</groupId>

                 </exclusion>

                 <exclusion>

                     <artifactId>curator-framework</artifactId>

                     <groupId>org.apache.curator</groupId>

                 </exclusion>

                 <exclusion>

                     <artifactId>curator-recipes</artifactId>

                     <groupId>org.apache.curator</groupId>

                 </exclusion>

             </exclusions>

         </dependency>

         <dependency>

             <groupId>com.dangdang</groupId>

             <artifactId>elastic-job-lite-spring</artifactId>

             <version>2.1.5</version>

         </dependency>

         <dependency>

             <groupId>org.apache.curator</groupId>

             <artifactId>curator-framework</artifactId>

             <version>2.10.0</version>

         </dependency>

         <dependency>

             <groupId>org.apache.curator</groupId>

             <artifactId>curator-client</artifactId>

             <version>2.10.0</version>

         </dependency>

         <dependency>

             <groupId>org.apache.curator</groupId>

             <artifactId>curator-recipes</artifactId>

             <version>2.10.0</version>

         </dependency>

     </dependencies>

然后就是配置zk注册中心,分布式功能主要依赖这个,所有属性都从yml中注入,这里注意一点,可以把超时时间设置大一点:

@Configuration

public class ElasticRegCenterConfig {

    /**

     * 配置zookeeper注册中心

     */

    @Bean(initMethod = "init")  // 需要配置init执行初始化逻辑

    public ZookeeperRegistryCenter regCenter(

            @Value("${regCenter.serverList}") final String serverList,

            @Value("${regCenter.namespace}") final String namespace) {

        ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration(serverList, namespace);

        zookeeperConfiguration.setMaxRetries(3); //设置重试次数,可设置其他属性

        zookeeperConfiguration.setSessionTimeoutMilliseconds(500000); //设置会话超时时间,尽量大一点,否则项目无法正常启动

        return new ZookeeperRegistryCenter(zookeeperConfiguration);

    }

}

然后就是配置job了,其实和spring的quartz配置都差不多,一个job类,一个调度类

这里先贴我的yml配置,任务执行周期,分片个数都从这里注入即可,分片使用后面单独说明:

二 simplejob

job类:

@Component

public class MySimpleJob implements SimpleJob {

    @Override

    public void execute(ShardingContext shardingContext) {

        System.out.println(shardingContext.getJobName()+"执行:"+

                "分片参数:"+shardingContext.getShardingParameter()+

                ",当前分片项:"+shardingContext.getShardingItem()+

                ",time:"+ LocalDate.now());

    }

}

配置类,这里用到了一个工具方法,工具类放下面:

/**

 * 配置MySimpleJob

 */

@Configuration

public class MySimpleJobConf {

    @Autowired ZookeeperRegistryCenter regCenter;

    @Autowired MySimpleJob mySimpleJob;

    /**

     * 配置任务调度: 参数:  任务

     *                    zk注册中心

     *                    任务详情

     */

    @Bean(initMethod = "init")

    public JobScheduler simpleJobScheduler(@Value("${mySimpleJob.cron}") final String cron,  //yml注入

                                           @Value("${mySimpleJob.shardingTotalCount}") final int shardingTotalCount,

                                           @Value("${mySimpleJob.shardingItemParameters}") final String shardingItemParameters) {

        return new SpringJobScheduler(mySimpleJob, regCenter,

                                      ElasticJobUtils.getSimpleJobConfiguration(

                                              mySimpleJob.getClass(),

                                              cron,

                                              shardingTotalCount,

                                              shardingItemParameters)

                                              //,new MyElasticJobListener() 可配置监听器

        );

    }

}

工具类:

public class ElasticJobUtils {

    /**

     * 创建简单任务详细信息

     */

    public static LiteJobConfiguration getSimpleJobConfiguration(final Class<? extends SimpleJob> jobClass, //任务类

                                                                final String cron,    // 运行周期配置

                                                                final int shardingTotalCount,  //分片个数

                                                                final String shardingItemParameters) {  // 分片参数

        return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(

                JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount)

                        .shardingItemParameters(shardingItemParameters).build()

                , jobClass.getCanonicalName())

        ).overwrite(true).build();

    }

    /**

     * 创建流式作业配置

     */

    public static LiteJobConfiguration getDataFlowJobConfiguration(final Class<? extends DataflowJob> jobClass, //任务类

                                                                   final String cron,    // 运行周期配置

                                                                   final int shardingTotalCount,  //分片个数

                                                                   final String shardingItemParameters,

                                                                   final Boolean streamingProcess   //是否是流式作业

                                                                   ) {  // 分片参数

        return LiteJobConfiguration.newBuilder(new DataflowJobConfiguration(

                JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount)

                        .shardingItemParameters(shardingItemParameters).build()

                // true为流式作业,除非fetchData返回数据为null或者size为0,否则会一直执行

                // false 非流式,只会按配置时间执行一次

                , jobClass.getCanonicalName(),streamingProcess)

        ).overwrite(true).build();

    }

}

测试:

三 dataflowjob

job类:

@Component

public class MyDataFlowJob implements DataflowJob<String> {

    @Override

    public List<String> fetchData(ShardingContext shardingContext) { //抓取数据

        System.out.println("---------获取数据---------");

        return Arrays.asList("1","2","3");

    }

    @Override

    public void processData(ShardingContext shardingContext, List<String> list) {//处理数据

        System.out.println("---------处理数据---------");

        list.forEach(x-> System.out.println("数据处理:"+x));

    }

}

配置类:

@Configuration

public class MyDataFlowJobConf {

    @Autowired ZookeeperRegistryCenter regCenter;

    @Autowired MyDataFlowJob myDataFlowJob;

    /**

     * 配置任务调度: 参数:  任务

     *                    zk注册中心

     *                    任务详情

     */

    @Bean(initMethod = "init")

    public JobScheduler dataFlowJobScheduler(@Value("${myDataFlowJob.cron}") final String cron,  //yml注入

                                           @Value("${myDataFlowJob.shardingTotalCount}") final int shardingTotalCount,

                                           @Value("${myDataFlowJob.shardingItemParameters}") final String shardingItemParameters) {

        return new SpringJobScheduler(myDataFlowJob, regCenter,

                ElasticJobUtils.getDataFlowJobConfiguration(

                        myDataFlowJob.getClass(),

                        cron,

                        shardingTotalCount,

                        shardingItemParameters,true)

                //,new MyElasticJobListener() 可配置监听器

        );

    }

}

测试:

需要注意一点流式作业如果数据不为空会一直跑

四 scriptjob

脚本任务有一点,不需要创建类实例,否则会报错,参数直接传null即可

配置类:

@Configuration

public class MyScriptJobConf {

    @Autowired ZookeeperRegistryCenter regCenter;

    /**

     * 配置任务调度: 参数:  任务

     *                    zk注册中心

     *                    任务详情

     */

    @Bean(initMethod = "init")

    public JobScheduler scriptJobScheduler(@Value("${myScriptJob.cron}") final String cron,  //yml注入

                                           @Value("${myScriptJob.shardingTotalCount}") final int shardingTotalCount,

                                           @Value("${myScriptJob.shardingItemParameters}") final String shardingItemParameters) {

        return new SpringJobScheduler(null, regCenter,

                                      ElasticJobUtils.getScriptJobConfiguration(

                                              "script_job",

                                              cron,

                                              shardingTotalCount,

                                              //命令或者脚本路径

                                              shardingItemParameters,"echo hello")

                                              //,new MyElasticJobListener() 可配置监听器

        );

    }

}

工具添加静态方法:

/**

     * 创建脚本作业配置

     */

    public static LiteJobConfiguration getScriptJobConfiguration(final String jobName, //任务名字

                                                                 final String cron,    // 运行周期配置

                                                                 final int shardingTotalCount,  //分片个数

                                                                 final String shardingItemParameters,

                                                                 final String scriptCommandLine   //是脚本路径或者命令

    ) {  // 分片参数

        return LiteJobConfiguration.newBuilder(new ScriptJobConfiguration(

                JobCoreConfiguration.newBuilder(jobName, cron, shardingTotalCount)

                        .shardingItemParameters(shardingItemParameters).build()

                // 此处配置文件路径或者执行命令

                , scriptCommandLine)

        ).overwrite(true).build();

    }

测试:

五分片用法

分片的目的就是通过配置分片个数,让不同的分片参数到不同的服务中去,比如配置了分片个数是2,那么分片一会到服务一中,分片二到服务二中

项目中根据分片参数来决定哪个服务处理哪些数据,比如 0=客户甲,1=客户乙,但是分片item是从1开始

分片算法默认是平均,可自定义,然后参数就是上面yml那种配置,比如2,就是 0=,1= 4就是0=,1=,2=,3=,两个服务的话服务一就是0,1的参数,服务二就是2,3的参数,并且分片item是3,4

然后要注意一点的是,这个分片识别是根据ip的,也就是说同一台电脑,跑两个程序没用,两个程序都会全部执行,还是会重复

主要是这个分片保证分布式中处理数据不重复,分片也会转移,即一个服务挂了之后,分片参数和item会自动转移到剩下服务中

六事件追踪(即任务信息持久化到mysql)

需要提前创建btach_log数据库

配置数据源Bean,在任务配置中添加event

@Configuration

@ConfigurationProperties(prefix = "spring.datasource")

public class JobDataSourceConf {

    private String url;

    private String username;

    private String password;

    private String driver_class_name;

    @Bean

    @Primary

    public DataSource hikariDataSource() {

        HikariDataSource dataSource = new HikariDataSource();

        dataSource.setJdbcUrl(url);

        dataSource.setUsername(username);

        dataSource.setPassword(password);

        dataSource.setDriverClassName(driver_class_name);

        return dataSource;

    }

程序会自动创建两张表并添加记录

七容易踩的坑

一配置类中配置bean的时候,方法名不要重复,否则会发现任务不跑,

二测试分布式的时候,必须跑在ip不一样的服务上,否则不会实现分片

三我的版本再pom里面,springboot版本是2.0.6,版本不一样可能用法也有些区别

四理论上xml更简单,但是我个人比较喜欢代码风格,哈哈

五脚本任务不能新建实例,参数传null,且确认命令是否有权限

巴特西

springboot整合elasticJob实战(纯代码开发三种任务类型用法)以及分片系统,事件追踪详解

一 springboot整合

二 simplejob

三 dataflowjob

四 scriptjob

五分片用法

六事件追踪(即任务信息持久化到mysql)

七容易踩的坑

最新文章

热门文章

巴特西

springboot整合elasticJob实战(纯代码开发三种任务类型用法)以及分片系统,事件追踪详解

一 springboot整合

二 simplejob

三 dataflowjob

四 scriptjob

五 分片用法

六 事件追踪(即任务信息持久化到mysql)

七 容易踩的坑

最新文章

热门文章

五分片用法

六事件追踪(即任务信息持久化到mysql)

七容易踩的坑