springboot2.1入门系列九 springboot集成elastic-job

Myth丶恋晨 2022-03-25 10:28 450阅读 0赞

本文为Spring Boot2.1系列的第九篇，代码可以从github下载 [https://github.com/yinww/demo-springboot2.git][https_github.com_yinww_demo-springboot2.git]

应用做了集群后能提供系统的稳定性和可靠性，但是如果集群系统涉及到定时器，一个应用的多个服务器中同时触发了相同的任务，这会出问题的。此时分布式定时任务就可以用来解决这个问题，本文介绍SpringBoot集成当当开源的分布式定时任务框架Elastic-Job。

一、下载Zookeeper

Elastic-Job的失效转移机制是依赖于Zookeeper实现的，因此需要先下载Zookeeper，并且需要3.4.6以上的版本。下载后启动Zookeeper，这里就不详细介绍Zookeeper的安装和启动，请自行查阅相关资料。

二、创建工程demo009

pom.xml的内容为

<project xmlns="http://maven.apache.org/POM/4.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <parent>
            <groupId>com.yinww</groupId>
            <artifactId>demo-springboot2</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </parent>
        <artifactId>demo009</artifactId>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>com.dangdang</groupId>
                <artifactId>elastic-job-lite-core</artifactId>
                <version>2.1.5</version>
            </dependency>
        </dependencies>
    </project>

三、Java代码

主类

package com.yinww.demo.springboot2.demo009;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class Demo009Application {
    
        public static void main(String[] args) {
            SpringApplication.run(Demo009Application.class, args);
        }
    
    }

Zookeeper 配置类

package com.yinww.demo.springboot2.demo009.config;
    
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    
    @Configuration
    @ConditionalOnExpression("'${regCenter.serverList}'.length() > 0")
    public class ZookeeperRegistryCenterConfig {
    
        @Bean(initMethod = "init")
        public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList,
                @Value("${regCenter.namespace}") final String namespace) {
            return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
        }
    
    }

Job配置类

package com.yinww.demo.springboot2.demo009.config;
    
    import javax.annotation.Resource;
    
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
    import com.dangdang.ddframe.job.config.JobCoreConfiguration;
    import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
    import com.dangdang.ddframe.job.lite.api.JobScheduler;
    import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
    import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
    import com.yinww.demo.springboot2.demo009.domain.Mail;
    import com.yinww.demo.springboot2.demo009.job.MailSendJob;
    
    @Configuration
    public class DataflowJobConfig {
    
        @Resource
        private ZookeeperRegistryCenter regCenter;
    
        @Bean
        public DataflowJob<Mail> mailSendJob(){
          return new MailSendJob();
        }
        
        /**
         * 普通邮件定时配置
         * @param mailSendJob
         * @param cron
         * @param shardingTotalCount
         * @param shardingItemParameters
         * @return
         */
        @Bean(initMethod = "init")
        public JobScheduler mailSendJobScheduler(final DataflowJob<Mail> mailSendJob, @Value("${mailSendJob.cron}") final String cron,
            @Value("${shardingCategory.shardingTotalCount}") final int shardingTotalCount,
            @Value("${shardingCategory.shardingItemParameters}") final String shardingItemParameters){
            LiteJobConfiguration jobConfiguration = createJobConfiguration(mailSendJob.getClass(), cron, shardingTotalCount, shardingItemParameters);
          return new JobScheduler(regCenter, jobConfiguration);
        }
        
        @SuppressWarnings("rawtypes")
        private LiteJobConfiguration createJobConfiguration(Class<? extends DataflowJob> jobClass, String cron, int shardingTotalCount
                , String shardingItemParameters) {
            // 定义作业核心配置
            JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build();
            // 定义Dataflow类型配置
            // false 非流式处理数据: 只会在每次作业执行过程中执行一次fetchData方法和processData方法，随即完成本次作业
            DataflowJobConfiguration jobConfiguration = new DataflowJobConfiguration(coreConfig, jobClass.getCanonicalName(), false);
            // 定义Lite作业根配置
            LiteJobConfiguration rootConfig = LiteJobConfiguration.newBuilder(jobConfiguration).overwrite(true).build();
            return rootConfig;
        }
    
    }

业务对象类

package com.yinww.demo.springboot2.demo009.domain;
    
    public class Mail {
        private Long id;
        private String subject;
        private String sendTo;
        private String content;
    
        // getter and setter
    
    }

Job类

package com.yinww.demo.springboot2.demo009.job;
    
    import java.util.List;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
    import com.yinww.demo.springboot2.demo009.domain.Mail;
    
    public class MailSendJob implements DataflowJob<Mail> {
    
        @Override
        public List<Mail> fetchData(ShardingContext shardingContext) {
            System.out.println(shardingContext.getShardingItem() + "====" + shardingContext.getShardingTotalCount());
            return null;
        }
    
        @Override
        public void processData(ShardingContext shardingContext, List<Mail> data) {
            // TODO Auto-generated method stub
        }
    
    }

四、配置文件

application.properties

# zookeeper config
    regCenter.serverList=localhost:2181
    regCenter.namespace=demo009
    
    # 每隔20秒执行
    mailSendJob.cron=0/20 * * * * ?
    
    # 总分片数
    shardingCategory.shardingTotalCount=10
    shardingCategory.shardingItemParameters=0=A,1=B,2=C,3=D,4=E,5=F,6=G,7=H,8=I,9=J

五、运行

执行

java -jar demo009-0.0.1-SNAPSHOT.jar

每隔20秒就能在控制台看到分布式定时任务的输出

0====10
    1====10
    2====10
    3====10
    4====10
    5====10
    6====10
    7====10
    8====10
    9====10

如果结合 [springboot2.1入门系列三 Springboot多环境配置][springboot2.1_ Springboot] 启动多个集群构成的应用，就能看到这10个分片信息在不同的应用控制台中输出

六、踩坑经验

Elastic-Job项目基于成熟的开源产品Quartz和Zookeeper及其客户端Curator进行二次开发，如果与新版的Spring Cloud一起使用，会出现Curator的版本冲突。因为Elastic-Job的最新版是2017年发布的，当当没有持续地对项目进行更新维护，因此在基于新版的Spring Cloud项目中需要谨慎使用Elastic-Job，报错内容为：

Caused by: java.lang.ClassNotFoundException: org.apache.curator.connection.StandardConnectionHandlingPolicy
    	at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[na:1.8.0_172]
    	at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[na:1.8.0_172]
    	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349) ~[na:1.8.0_172]
    	at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[na:1.8.0_172]
    	... 40 common frames omitted

解决这个问题的办法是在pom.xml中强制使用低版本的Curator

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>2.10.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>2.10.0</version>
    </dependency>

本文内容到此结束，更多内容可关注公众号：

![2018111718160012.jpg][]

[https_github.com_yinww_demo-springboot2.git]: https://github.com/yinww/demo-springboot2.git
[springboot2.1_ Springboot]: https://blog.csdn.net/bowei026/article/details/86516912
[2018111718160012.jpg]: /images/20220325/12fe30fec45f49c88c00477134ecf057.png