分布式调度Elastic-Job实战解析

妖狐艹你老母 2023-10-02 20:31 5阅读 0赞

分布式调度Elastic-Job实战解析今天分享，承接上篇的内容，Elastic-Job是一个去中心化，支持随机分片策略的轻量级解决方案，下面我们分析其实战使用。

1、springBoot中pom核心包引用

<?xml version="1.0" encoding="UTF-8"?>
    <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">
        <parent>
            <artifactId>schedule</artifactId>
            <groupId>com.enjoy</groupId>
            <version>1.0</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
        <artifactId>job-sharding-elastic</artifactId>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <maven.compiler.source>1.7</maven.compiler.source>
            <maven.compiler.target>1.7</maven.compiler.target>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>com.github.yinjihuan</groupId>
                <artifactId>elastic-job-spring-boot-starter</artifactId>
                <version>1.0.2</version>
            </dependency>
        </dependencies>
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-dependencies</artifactId>
                    <version>2.1.0.RELEASE</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
            </dependencies>
        </dependencyManagement>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <version>2.2.0.RELEASE</version>
                </plugin>
            </plugins>
        </build>
        <repositories>
            <repository>
                <id>jitpack.io</id>
                <url>https://jitpack.io</url>
            </repository>
        </repositories>
    </project>

2、application.properties 配置文件信息

# zk注册中心
    elastic.job.zk.serverLists=127.0.0.1:2181
    elastic.job.zk.namespace=enjoy_elastic7
    
    spring.main.allow-bean-definition-overriding=true
    server.port=8283

注册中心的作用是做协调、调度使用。

3、两种任务的使用方式

实现 SimpleJob 接口

package com.enjoy.job;
    
    import com.cxytiandi.elasticjob.annotation.ElasticJobConf;
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import com.enjoy.business.EnjoyBusiness;
    import org.springframework.beans.factory.annotation.Autowired;
    
    @ElasticJobConf(name = "EnjoySimpleJob",cron = "0/5 * * * * ?"
    		,shardingItemParameters = "0=beijing|shenzhen|tianjin,1=shanghai",shardingTotalCount = 2
    		,listener = "com.enjoy.handle.MessageElasticJobListener"
    		,jobExceptionHandler = "com.enjoy.handle.CustomJobExceptionHandler"
    )
    public class EnjoySimpleJob implements SimpleJob {
    	@Autowired
    	private EnjoyBusiness enjoyBusiness;
    
    	public void execute(ShardingContext context) {
    		System.out.println("EnjoySimpleJob,当前分片："+context.getShardingParameter());
    
    		//当前起始
    		//context.getShardingParameter(),回返切片信息beijing
    		String sql = enjoyBusiness.getSql(context.getShardingParameter());
    		enjoyBusiness.process(sql);
    	}
    
    }

实现 DataflowJob<String> 接口

package com.enjoy.job;
    
    import com.cxytiandi.elasticjob.annotation.ElasticJobConf;
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
    import com.dangdang.ddframe.job.api.simple.SimpleJob;
    import com.enjoy.business.EnjoyBusiness;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import java.util.Arrays;
    import java.util.List;
    
    @ElasticJobConf(name = "EnjoyDataflowJob",cron = "0/5 * * * * ?"
    		,shardingItemParameters = "0=beijing,1=shanghai",shardingTotalCount = 2
    		,listener = "com.enjoy.handle.MessageElasticJobListener",jobExceptionHandler = "com.enjoy.handle.CustomJobExceptionHandler"
    )
    public class EnjoyDataflowJob implements DataflowJob<String> {
    	@Autowired
    	private EnjoyBusiness enjoyBusiness;
    
    	@Override
    	public List fetchData(ShardingContext context) {
    		String sql = enjoyBusiness.getSql(context.getShardingParameter());
    		return Arrays.asList(sql);
    	}
    
    	@Override
    	public void processData(ShardingContext context, List<String> list) {
    		System.out.println("EnjoyDataflowJob,当前分片："+context.getShardingParameter());
    		enjoyBusiness.process(list.get(0));
    	}
    }

shardingItemParameters 参数分片策略是，如果分了两个片，这时候假如启动了一台服务器， 此服务会循环执行这两个片；假如启动了两个服务器，那么每个服务器执行一个片；假如启动了三个或者更多的服务器，那么ZK会随机找两台服务分别执行一个片。

4、业务类

package com.enjoy.business;
    
    import org.springframework.stereotype.Service;
    
    @Service
    public class EnjoyBusiness {
    
    	private String sql_tpl = "select * from xxx where param = %s";
    
    	public String getSql(String param){
    		return String.format(sql_tpl,param);
    	}
    
    	/**
    	 * index指示，现在跑的是第几片
    	 * total指示，总共有几片
         */
    	public void process(String sql) {
    
    		//分片任务
    		System.out.println("当前执行sql："+sql);
    	}
    
    }

5、监听器

package com.enjoy.handle;
    
    import com.dangdang.ddframe.job.executor.ShardingContexts;
    import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /**
     * 作业监听器
     */
    public class MessageElasticJobListener implements ElasticJobListener {
    
        @Override
        public void beforeJobExecuted(ShardingContexts shardingContexts) {
            String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            String msg = date + " 【任务开始执行-" + shardingContexts.getJobName() + "】";
            System.out.println("给管理发邮件："+msg);
    
        }
    
        @Override
        public void afterJobExecuted(ShardingContexts shardingContexts) {
        	String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            String msg = date + " 【任务执行结束-" + shardingContexts.getJobName() + "】" ;
            System.out.println("给管理发邮件："+msg);
        }
    
    }

6、异常处理

package com.enjoy.handle;
    
    import com.dangdang.ddframe.job.executor.handler.JobExceptionHandler;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * 自定义异常处理
     */
    public class CustomJobExceptionHandler implements JobExceptionHandler {
    
        private Logger logger = LoggerFactory.getLogger(CustomJobExceptionHandler.class);
    
        @Override
        public void handleException(String jobName, Throwable cause) {
            logger.error(String.format("Job '%s' exception occur in job processing", jobName), cause);
            System.out.println("给管理发邮件：【"+jobName+"】任务异常。" + cause.getMessage());
        }
    
    }

7、简单的启动类

package com.enjoy;
    
    import com.cxytiandi.elasticjob.annotation.EnableElasticJob;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    @EnableElasticJob //重要注解
    public class ElasticJobApp {
        public static void main(String[] args) {
            SpringApplication.run(ElasticJobApp.class,args);
        }
    }

8、执行步骤图

![6b5ab47e22ee4a46bb291d86ed087d76.png][]

以上就是Elastic-Job的实战使用方式，比较轻量级，节约资源，没有页面配置，下篇我们分析 Elastic-Job 和xxl-job的区别和优缺点，敬请期待！

[6b5ab47e22ee4a46bb291d86ed087d76.png]: https://img-blog.csdnimg.cn/6b5ab47e22ee4a46bb291d86ed087d76.png