spring boot 后端 工具篇--分布式定时任务springBoot--elasticjob简单使用（1）

文章目录

前言一、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 引入库

org.apache.shardingsphere.elasticjob

elasticjob-lite-core

3.0.4

org.yaml

snakeyaml

2.2

org.codehaus.groovy

groovy-all

2.4.15

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 {

@Override

public void execute(ShardingContext shardingContext) {

// 分片参数 0=text,1=image,2=radio,3=vedio

String 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 0

break;

case 1:

// do something by sharding item 1

break;

case 2:

// do something by sharding item 2

break;

// 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给到zk

shieldElasticjobIpIsNull();

// 运行任务

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.0

if("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

org.codehaus.groovy

groovy-all

2.4.15

编写如下方法，目的只是为了在特殊情况下改变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.0

if("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 的简单使用进行介绍。

相关阅读