文章目录
- 前言
- 一、elasticjob 介绍：
- 二、elasticjob 使用：
- 2.1 部署zookeeper：
- 2.2 引入库
- 2.2 定义任务：
- 2.3 任务执行：
- 2.4 任务执行控制台输出：
- 三、elasticjob 启动错误：
- 3.1 KeeperErrorCode = OperationTimeout：
- 3.2 .HostException: ip is null：
- 总结
前言
本文对 elasticjob 的简单使用进行介绍。
一、elasticjob 介绍：
ElasticJob 是一个分布式任务调度框架，由当当网开发并开源。它基于 Zookeeper 实现分布式协调，采用经典的分片算法，能够实现弹性扩容和缩容的分布式任务调度。ElasticJob 能够灵活地应用于各种环境中，如易用性、稳定性、弹性可伸缩等方面表现优异。
以下是 ElasticJob 的一些主要特性：
-
分布式任务调度：ElasticJob 基于 Zookeeper 实现分布式协调，支持分布式自动负载均衡调度。
-
弹性扩缩容：ElasticJob 支持弹性扩容和缩容，在任务节点伸缩时能够自动调整任务分片。
-
丰富的定时调度策略：ElasticJob 提供了各种灵活的调度策略，如简单的固定频率、CRON 等。
-
支持多种任务处理逻辑：ElasticJob 支持处理数据流、打印日志、脚本处理等多种任务处理逻辑。
-
统计和监控：ElasticJob 提供了完善的统计和监控功能，可以监控任务执行状态和运行情况。
总体来说，ElasticJob 是一个功能丰富、可靠且易于集成的分布式任务调度框架，广泛应用于企业系统中的定时任务调度、数据处理等场景。
二、elasticjob 使用：
2.1 部署zookeeper：
因为job需要注册到zk 上，依赖于zk 的leader选举，所以需要先进行zk 的安装；
阿里云轻量服务器–Docker–Zookeeper&Kafka安装；
window Zookeeper zk 启动；
2.2 引入库
<!-- https://mvnrepository.com/artifact/org.apache.shardingsphere.elasticjob/elasticjob-lite-core -->
<!-- 定时任务核心库 -->
<dependency><groupId>org.apache.shardingsphere.elasticjob</groupId><artifactId>elasticjob-lite-core</artifactId><version>3.0.4</version><!--<version>3.0.1</version>-->
</dependency>
<!-- https://mvnrepository.com/artifact/org.yaml/snakeyaml -->
<!-- SnakeYaml用于解析YAML -->
<dependency><groupId>org.yaml</groupId><artifactId>snakeyaml</artifactId><!--<version>1.27</version>--><version>2.2</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.codehaus.groovy/groovy-all -->
<!--在程序运行时任意修改代码逻辑 -->
<dependency><groupId>org.codehaus.groovy</groupId><artifactId>groovy-all</artifactId><version>2.4.15</version>
</dependency>
2.2 定义任务：
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class MyJob implements SimpleJob {@Overridepublic void execute(ShardingContext shardingContext) {// 分片参数 0=text,1=image,2=radio,3=vedioString shardingParameter= shardingContext.getShardingParameter();String jobParameter= shardingContext.getJobParameter();log.debug("job 执行 error,job名称：{},分片数量:{},分片：{},分片参数:{},jobParamer:{}", shardingContext.getJobName(), shardingContext.getShardingTotalCount(),shardingContext.getShardingItem(), shardingParameter,jobParameter);if ("text".equals(jobParameter)) {// do something by sharding}switch (shardingContext.getShardingItem()) {case 0:// do something by sharding item 0break;case 1:// do something by sharding item 1break;case 2:// do something by sharding item 2break;// case n: ...}}
}2.3 任务执行：
import groovy.lang.GroovyShell;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.api.JobExtraConfiguration;
import org.apache.shardingsphere.elasticjob.infra.env.HostException;
import org.apache.shardingsphere.elasticjob.infra.env.IpUtils;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobBootstrap;
import org.apache.shardingsphere.elasticjob.lite.internal.snapshot.SnapshotService;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
import org.apache.shardingsphere.elasticjob.tracing.event.JobEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
import org.springframework.util.StringUtils;public class MyJobDemo {public static void main(String[] args) {// 电脑连接无线网时 连接zk 可能出现ip is null 错误，此时设置一个ip给到zkshieldElasticjobIpIsNull();// 运行任务new ScheduleJobBootstrap(createRegistryCenter(), new MyJob(), createJobConfiguration()).schedule();}// 配置zookeeper 连接private static CoordinatorRegistryCenter createRegistryCenter() {ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("localhost:2181", "my-job");zookeeperConfiguration.setConnectionTimeoutMilliseconds(10000);zookeeperConfiguration.setSessionTimeoutMilliseconds(10000);zookeeperConfiguration.setMaxSleepTimeMilliseconds(10000);CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);regCenter.init();return regCenter;// ... 分片参数 分片从0开始到分片总数-1}// 配置job 运行时机private static JobConfiguration createJobConfiguration() {// 创建作业配置 /*** myjob-param job 的名称 同一个zk命名空间下 需要唯一* 1 分片个数* cron 任务运行的cron 表达式* overwrite 运行job 的配置被覆盖写入，默认为false* shardingItemParameters 分片参数（随后同 分片个数一同介绍）* jobParameter job的参数（job 业务端在执行任务的时候可以接收到该参数）**/JobConfiguration jobConfiguration = JobConfiguration.newBuilder("myjob-param", 1).cron("0/5 * * * * ?").overwrite(true)// .shardingItemParameters("0=Beijing,1=Shanghai,2=Guangzhou").jobParameter("jobparamer").build();return jobConfiguration;}/*** 屏蔽org.apache.shardingsphere.elasticjob.infra.env.IpUtils.getIp()抛出* HostException(ip is null) 的异常导致windows本地程序无法启动*/private static void shieldElasticjobIpIsNull(){try {IpUtils.getIp();} catch (HostException e) {//抛出HostException 且 异常信息为 "ip is null" 时，设置ip地址为 0.0.0.0if("ip is null".equals(e.getMessage())){String code = "org.apache.shardingsphere.elasticjob.infra.env.IpUtils.cachedIpAddress=\"0.0.0.0\";";GroovyShell groovy = new GroovyShell();groovy.evaluate(code);}}}}2.4 任务执行控制台输出：
三、elasticjob 启动错误：
3.1 KeeperErrorCode = OperationTimeout：
报错位置在 ZookeeperRegistryCenter 的 init() 方法中：
这里等待一段时间后如果还没有连接到zk 就会报错，默认的等待时间是 3000ms * 3 = 9s ，此时可以考虑增加 maxSleepTimeMilliseconds 的时间：
private static CoordinatorRegistryCenter createRegistryCenter() {ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("139.196.92.249:2181", "my-job");zookeeperConfiguration.setConnectionTimeoutMilliseconds(10000);zookeeperConfiguration.setSessionTimeoutMilliseconds(10000);// 增加 maxSleepTimeMilliseconds 时间zookeeperConfiguration.setMaxSleepTimeMilliseconds(10000);CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);regCenter.init();return regCenter;// ... 分片参数 分片从0开始到分片总数-1}
3.2 .HostException: ip is null：
错误代码：
这里会会获取到本机的ip 进行遍历找到一个符合要求的ip 然后进行返回，如果所有的ip 都不通过，则抛出ip is null 的问题；
处理：引入： groovy-all
<dependency><groupId>org.codehaus.groovy</groupId><artifactId>groovy-all</artifactId><version>2.4.15</version>
</dependency>
编写如下方法，目的只是为了在特殊情况下改变cachedIpAddress的值：
/*** 屏蔽org.apache.shardingsphere.elasticjob.infra.env.IpUtils.getIp()抛出* HostException(ip is null) 的异常导致windows本地程序无法启动*/
private static void shieldElasticjobIpIsNull(){try {IpUtils.getIp();} catch (HostException e) {//抛出HostException 且 异常信息为 "ip is null" 时，设置ip地址为 0.0.0.0if("ip is null".equals(e.getMessage())){String code = "org.apache.shardingsphere.elasticjob.infra.env.IpUtils.cachedIpAddress=\"0.0.0.0\";";GroovyShell groovy = new GroovyShell();groovy.evaluate(code);}}
}
启动类的main方法内部一开始就调用上面这个shieldElasticjobIpIsNull初始化cachedIpAddress：
public static void main(String[] args) {//屏蔽org.apache.shardingsphere.elasticjob.infra.env.IpUtils.getIp()抛出//HostException(ip is null) 的异常导致windowes本地程序无法启动的问题shieldElasticjobIpIsNull();new ScheduleJobBootstrap(createRegistryCenter(), new MyJob(), createJobConfiguration()).schedule();}
总结
本文对 elasticjob 的简单使用进行介绍。
