天天看点
返回

ElasticJob‐Lite：扩展作业类型

​

​ElasticJob​

 ​​的作业分类基于​

​class​

 ​​和​

​type​

 ​​两种类型。基于​

​class​

 ​​的作业需要开发者自行通过实现接口的方式织入业务逻辑；基于​

​type​

 ​​的作业则无需编码，只需要提供相应配置即可。基于​

​class​

 ​​的作业接口的方法参数​

​shardingContext​

 ​​包含作业配置、片和运行时信息。可通过​

​getShardingTotalCount()​

 ​​、​

​getShardingItem()​

​等方法分别获取分片总数和运行在本作业服务器的分片序列号等。

​

​ElasticJob​

 ​​目前提供​

​Simple​

 ​​、​

​Dataflow​

 ​​这两种基于​

​class​

 ​​的作业类型，并提供​

​Script​

 ​​、​

​HTTP​

 ​​这两种基于​

​type​

 ​​的作业类型，用户可通过实现​

​SPI​

 ​接口自行扩展作业类型。

添加依赖（​

​3.0.1​

​​是目前最新的​

​Releases​

​版本）：

<dependency>
            <groupId>org.apache.shardingsphere.elasticjob</groupId>
            <artifactId>elasticjob-lite-core</artifactId>
            <version>3.0.1</version>
        </dependency>

本篇博客介绍如何通过实现​

​SPI​

​接口自行扩展作业类型。

扩展作业类型

想要通过实现​

​SPI​

​​接口自行扩展作业类型需要三个步骤（基于​

​class​

​​的作业类型），而基于​

​type​

​的作业类型只需要后面两个步骤：

  1. 定义作业逻辑的执行接口（基于​

    ​type​

     ​​的作业类型不需要此步骤，比如​

    ​Script​

     ​​作业的作业逻辑由脚本程序执行，而​

    ​HTTP​

     ​​作业的作业逻辑由请求的服务端执行，因此基于​

    ​type​

     ​的作业类型不需要定义作业逻辑的执行接口）。
  2. 实现作业逻辑执行接口的​

    ​JobItemExecutor​

     ​。
  3. 通过​

    ​Java SPI​

     ​​的方式声明实现的​

    ​JobItemExecutor​

     ​。

作业逻辑执行接口

​

​KavenJob​

​​接口（继承​

​ElasticJob​

​接口，作业逻辑的执行接口）：

package com.kaven.job.my;

import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;

public interface KavenJob extends ElasticJob {
    void work(ShardingContext shardingContext, String jobExecutorName);
}

​

​KavenJob​

​​接口的实现类​

​MyKavenJob​

​（实现作业逻辑的执行）：

package com.kaven.job.my;

import org.apache.shardingsphere.elasticjob.api.ShardingContext;

import java.text.SimpleDateFormat;
import java.util.Date;

public class MyKavenJob implements KavenJob{

    private static final SimpleDateFormat formatter =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    public void work(ShardingContext shardingContext, String jobExecutorName) {
        String job = shardingContext.getShardingParameter();
        if(job == null || job.trim().equals("")) {
            System.out.println("请指定帮[Kaven]执行的任务名称！");
            throw new RuntimeException();
        }
        System.out.printf("%s[%s]:帮[Kaven]执行%s任务！\n", jobExecutorName, formatter.format(new Date()), job);
    }
}

内置的​

​Simple​

​​和​

​Dataflow​

​作业的作业逻辑执行接口也是这样定义的：

package org.apache.shardingsphere.elasticjob.simple.job;

import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;

public interface SimpleJob extends ElasticJob {
    void execute(ShardingContext var1);
}      
package org.apache.shardingsphere.elasticjob.dataflow.job;

import java.util.List;
import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;

public interface DataflowJob<T> extends ElasticJob {
    List<T> fetchData(ShardingContext var1);

    void processData(ShardingContext var1, List<T> var2);
}

因此想要定义​

​Simple​

​​或​

​Dataflow​

​​作业就只需要分别实现​

​SimpleJob​

​​或​

​DataflowJob​

​接口即可。

JobItemExecutor

​

​ElasticJob​

​​的作业分类基于​

​class​

​​和​

​type​

​​两种类型，因此​

​JobItemExecutor​

​​必须继承或实现​

​ClassedJobItemExecutor​

​​或者​

​TypedJobItemExecutor​

​接口。

​

​KavenJobItemExecutor​

​​接口（继承了​

​ClassedJobItemExecutor​

​​，如果要继承​

​TypedJobItemExecutor​

​​接口来扩展​

​type​

​​类型作业的自定义​

​JobItemExecutor​

​，也是类似的）：

package com.kaven.job.my;

import org.apache.shardingsphere.elasticjob.executor.item.impl.ClassedJobItemExecutor;

public interface KavenJobItemExecutor extends ClassedJobItemExecutor<KavenJob> {
    String getJobExecutorName();
}

​

​KavenJobItemExecutor​

​​接口的实现类​

​MyKavenJobExecutor​

​：

package com.kaven.job.my;

import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;

public class MyKavenJobExecutor implements KavenJobItemExecutor {

    public MyKavenJobExecutor() {}
    
    @Override
    public Class<KavenJob> getElasticJobClass() {
        return KavenJob.class;
    }

    @Override
    public void process(KavenJob kavenJob, JobConfiguration jobConfiguration, JobFacade jobFacade, ShardingContext shardingContext) {
        kavenJob.work(shardingContext, getJobExecutorName());
    }

    @Override
    public String getJobExecutorName() {
        return this.getClass().getName();
    }
}

很显然​

​MyKavenJob​

​​类中的​

​work​

​​方法将在​

​MyKavenJobExecutor​

​​类的​

​process​

​​方法中调用，这是由​

​ElasticJob​

​​控制的。​

​Simple​

​​、​

​Dataflow​

​​、​

​Script​

​​和​

​HTTP​

​​作业也是如此，在​

​JobItemExecutor​

​​中执行作业的分片（以​

​Simple​

​​和​

​Script​

​作业为例）：

package org.apache.shardingsphere.elasticjob.simple.executor;

import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.executor.item.impl.ClassedJobItemExecutor;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;

public final class SimpleJobExecutor implements ClassedJobItemExecutor<SimpleJob> {
    public SimpleJobExecutor() {
    }

    public void process(SimpleJob elasticJob, JobConfiguration jobConfig, JobFacade jobFacade, ShardingContext shardingContext) {
        elasticJob.execute(shardingContext);
    }

    public Class<SimpleJob> getElasticJobClass() {
        return SimpleJob.class;
    }
}      
package org.apache.shardingsphere.elasticjob.script.executor;

import com.google.common.base.Strings;
import java.io.IOException;
import java.util.Properties;
import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.DefaultExecutor;
import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.executor.JobFacade;
import org.apache.shardingsphere.elasticjob.executor.item.impl.TypedJobItemExecutor;
import org.apache.shardingsphere.elasticjob.infra.exception.JobConfigurationException;
import org.apache.shardingsphere.elasticjob.infra.exception.JobSystemException;
import org.apache.shardingsphere.elasticjob.infra.json.GsonFactory;

public final class ScriptJobExecutor implements TypedJobItemExecutor {
    public ScriptJobExecutor() {
    }

    public void process(ElasticJob elasticJob, JobConfiguration jobConfig, JobFacade jobFacade, ShardingContext shardingContext) {
        CommandLine commandLine = CommandLine.parse(this.getScriptCommandLine(jobConfig.getProps()));
        commandLine.addArgument(GsonFactory.getGson().toJson(shardingContext), false);

        try {
            (new DefaultExecutor()).execute(commandLine);
        } catch (IOException var7) {
            throw new JobSystemException("Execute script failure.", new Object[]{var7});
        }
    }

    private String getScriptCommandLine(Properties props) {
        String result = props.getProperty("script.command.line");
        if (Strings.isNullOrEmpty(result)) {
            throw new JobConfigurationException("Cannot find script command line, job is not executed.", new Object[0]);
        } else {
            return result;
        }
    }

    public String getType() {
        return "SCRIPT";
    }
}

声明JobItemExecutor

通过​

​Java SPI​

​​的方式声明实现的​

​JobItemExecutor​

​​，那为什么需要声明实现的​

​JobItemExecutor​

​​？因为​

​ElasticJob​

​​需要知道作业对应的​

​JobItemExecutor​

​​，以便用它来执行该作业的分片。​

​ElasticJob​

​​通过​

​ScheduleJobBootstrap​

​实例来完成定时任务的调度。

​

​Application​

​类（启动类）：

package com.kaven.job;

import com.kaven.job.my.MyKavenJob;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobBootstrap;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;

/**
 * @Author: ITKaven
 * @Date: 2021/11/20 17:05
 * @Leetcode: https://leetcode-cn.com/u/kavenit
 * @Notes:
 */
public class Application {

    public static void main(String[] args) {
        new ScheduleJobBootstrap(createRegistryCenter(), new MyKavenJob(),
                createJobConfiguration()).schedule();
    }
    private static CoordinatorRegistryCenter createRegistryCenter() {
        ZookeeperConfiguration zc = new ZookeeperConfiguration("192.168.1.200:9999", "my-job");
        zc.setConnectionTimeoutMilliseconds(40000);
        zc.setMaxRetries(5);
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zc);
        regCenter.init();
        return regCenter;
    }
    private static JobConfiguration createJobConfiguration() {
        String jobs = "0=看论文,1=做实验,2=打比赛,3=开组会,4=看书,5=其他";
        return JobConfiguration.newBuilder("KavenJob", 6)
                .cron("30 * * * * ?")
                .shardingItemParameters(jobs)
                .overwrite(true)
                .failover(true)
                .build();
    }
}

创建​

​ScheduleJobBootstrap​

​​实例，也会创建​

​JobScheduler​

​实例。

public ScheduleJobBootstrap(CoordinatorRegistryCenter regCenter, ElasticJob elasticJob, JobConfiguration jobConfig) {
        this.jobScheduler = new JobScheduler(regCenter, elasticJob, jobConfig);
    }

    public ScheduleJobBootstrap(CoordinatorRegistryCenter regCenter, String elasticJobType, JobConfiguration jobConfig) {
        this.jobScheduler = new JobScheduler(regCenter, elasticJobType, jobConfig);
    }

而在创建​

​JobScheduler​

​​实例时，还会创建​

​ElasticJobExecutor​

​实例。

public JobScheduler(CoordinatorRegistryCenter regCenter, ElasticJob elasticJob, JobConfiguration jobConfig) {
        ...
        this.jobExecutor = new ElasticJobExecutor(elasticJob, this.jobConfig, this.jobFacade);
        ...
    }

    public JobScheduler(CoordinatorRegistryCenter regCenter, String elasticJobType, JobConfiguration jobConfig) {
        ...
        this.jobExecutor = new ElasticJobExecutor(elasticJobType, this.jobConfig, this.jobFacade);
        ...
    }

在创建​

​ElasticJobExecutor​

​​实例时，会通过​

​JobItemExecutorFactory​

​​类获取作业对应的​

​JobItemExecutor​

​。

public ElasticJobExecutor(final ElasticJob elasticJob, final JobConfiguration jobConfig, final JobFacade jobFacade) {
        this(elasticJob, jobConfig, jobFacade, JobItemExecutorFactory.getExecutor(elasticJob.getClass()));
    }
    
    public ElasticJobExecutor(final String type, final JobConfiguration jobConfig, final JobFacade jobFacade) {
        this(null, jobConfig, jobFacade, JobItemExecutorFactory.getExecutor(type));
    }

​

​JobItemExecutorFactory​

​类：

package org.apache.shardingsphere.elasticjob.executor.item;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.infra.exception.JobConfigurationException;
import org.apache.shardingsphere.elasticjob.executor.item.impl.ClassedJobItemExecutor;
import org.apache.shardingsphere.elasticjob.executor.item.impl.TypedJobItemExecutor;
import org.apache.shardingsphere.elasticjob.infra.spi.ElasticJobServiceLoader;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.ServiceLoader;

/**
 * Job item executor factory.
 */
@SuppressWarnings("rawtypes")
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class JobItemExecutorFactory {
    
    private static final Map<Class, ClassedJobItemExecutor> CLASSED_EXECUTORS = new HashMap<>();
    
    static {
        ElasticJobServiceLoader.registerTypedService(TypedJobItemExecutor.class);
        ServiceLoader.load(ClassedJobItemExecutor.class).forEach(each -> CLASSED_EXECUTORS.put(each.getElasticJobClass(), each));
    }
    
    /**
     * Get executor.
     * 
     * @param elasticJobClass elastic job class
     * @return job item executor
     */
    @SuppressWarnings("unchecked")
    public static JobItemExecutor getExecutor(final Class<? extends ElasticJob> elasticJobClass) {
        for (Entry<Class, ClassedJobItemExecutor> entry : CLASSED_EXECUTORS.entrySet()) {
            if (entry.getKey().isAssignableFrom(elasticJobClass)) {
                return entry.getValue();
            }
        }
        throw new JobConfigurationException("Can not find executor for elastic job class `%s`", elasticJobClass.getName());
    }
    
    /**
     * Get executor.
     *
     * @param elasticJobType elastic job type
     * @return job item executor
     */
    public static JobItemExecutor getExecutor(final String elasticJobType) {
        return ElasticJobServiceLoader.getCachedTypedServiceInstance(TypedJobItemExecutor.class, elasticJobType)
                .orElseThrow(() -> new JobConfigurationException("Cannot find executor for elastic job type `%s`", elasticJobType));
    }
}

​

​ServiceLoader​

​​类就是​

​Java​

​​提供的​

​SPI​

​​，​

​SPI​

​​（​

​Service Provider Interface​

​​）是​

​JDK​

​​内置的一种服务提供发现机制，可以用来启用框架扩展和替换组件，主要是被框架的开发人员使用，不同厂商可以针对同一接口做出不同的实现，比如​

​java.sql.Driver​

​​接口，​

​MySQL​

​​和​

​PostgreSQL​

​​都提供了对应的实现给用户使用，而​

​Java​

​​的​

​SPI​

​​机制可以为某个接口寻找服务实现。​

​Java​

​​中​

​SPI​

​机制主要思想是将装配的控制权移到程序之外，在模块化设计中这个机制尤其重要，其核心思想就是解耦。

​

​ServiceLoader​

​类正常工作的唯一要求是服务提供类必须具有无参构造函数，以便它们可以在加载期间实例化。通过在资源目录的​

​META-INF/services​

​​中放置服务提供者配置文件来标识服务提供者，文件名是服务类型的完全限定名，该文件包含具体的服务提供者类的完全限定名列表，每行一个，每个名称周围的空格和制表符以及空行都将被忽略，该文件必须以​

​UTF-8​

​编码。如下图所示：

ElasticJob‐Lite：扩展作业类型

完成这三个步骤，我们实现的​

​JobItemExecutor​

​​就可以被​

​ElasticJob​

​发现，以便将它们用于对应作业分片的执行。

static {
        // 加载type作业的JobItemExecutor
        ElasticJobServiceLoader.registerTypedService(TypedJobItemExecutor.class);
        // 加载class作业的JobItemExecutor
        ServiceLoader.load(ClassedJobItemExecutor.class).forEach(each -> CLASSED_EXECUTORS.put(each.getElasticJobClass(), each));
    }

在​

​JobItemExecutorFactory​

​​类的静态块中会加载​

​class​

​​和​

​type​

​​作业的​

​JobItemExecutor​

​​，加载​

​class​

​​作业的​

​JobItemExecutor​

​​时，以​

​each.getElasticJobClass()​

​​为​

​key​

​​，​

​each​

​​为​

​value​

​​。而​

​MyKavenJobExecutor​

​​类的​

​getElasticJobClass​

​​方法返回​

​KavenJob.class​

​​，这样作业和​

​JobItemExecutor​

​就对应起来了。

@Override
    public Class<KavenJob> getElasticJobClass() {
        return KavenJob.class;
    }

加载​

​type​

​​作业的​

​JobItemExecutor​

​​是在​

​ElasticJobServiceLoader​

​​类中完成的（也是使用​

​ServiceLoader​

​​类来加载），以​

​instance.getType()​

​​为​

​key​

​​，​

​instance​

​​为​

​value​

​。

public static <T extends TypedSPI> void registerTypedService(final Class<T> typedService) {
        if (TYPED_SERVICES.containsKey(typedService)) {
            return;
        }
        ServiceLoader.load(typedService).forEach(each -> registerTypedServiceClass(typedService, each));
    }
    private static <T extends TypedSPI> void registerTypedServiceClass(final Class<T> typedService, final TypedSPI instance) {
        TYPED_SERVICES.computeIfAbsent(typedService, unused -> new ConcurrentHashMap<>()).putIfAbsent(instance.getType(), instance);
        TYPED_SERVICE_CLASSES.computeIfAbsent(typedService, unused -> new ConcurrentHashMap<>()).putIfAbsent(instance.getType(), instance.getClass());
    }

这也是为什么​

​ScheduleJobBootstrap​

​​构造器的​

​elasticJobType​

​​参数需要全部大写（比如​

​SCRIPT​

​​和​

​HTTP​

​）的原因。

new ScheduleJobBootstrap(createRegistryCenter(), "SCRIPT",
        createJobConfiguration()).schedule();

因为这两种作业对应的​

​JobItemExecutor​

​​就是使用​

​getType()​

​​的返回值作为​

​key​

​进行存储的。

public String getType() {
        return "SCRIPT";
    }
    public String getType() {
        return "HTTP";
    }

这样就通过实现​

​SPI​

​接口自行扩展了作业类型，输出如下图所示：

ElasticJob‐Lite：扩展作业类型
ElasticJob‐Lite：扩展作业类型
ElasticJob‐Lite：扩展作业类型
elastic-job elastic apache java
上一篇: Myeclipse8.x配置svn插件
下一篇: matplotlib简单四图----条形图，点阵图，盒型图，箱形图

继续阅读