从Zookeeper 到 Elastic Job 的Simple Job使用（二）

MickyInvQ

发布于 2020-09-27 10:51:40

1.3K0

发布于 2020-09-27 10:51:40

文章被收录于专栏：InvQ的专栏InvQ的专栏

elastic job demo

一、zookeeper要有

上一篇文章写过了。。

二、maven引入

                <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-core</artifactId>
            <version>2.1.5</version>
        </dependency>
        <!-- 使用springframework自定义命名空间时引入 -->
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-spring</artifactId>
            <version>2.1.5</version>
        </dependency>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.9</version>
        </dependency>

三、demo开发

   package com.test.elasticjob;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;


public class MyElasticJob implements SimpleJob {

    @Override
    public void execute(ShardingContext context) {
        System.out.println(context.toString());
        switch (context.getShardingItem()) {
            case 0:
                System.out.println("------------->>>>0");
                break;
            case 1:
                System.out.println("------------->>>>1");
                break;
            case 2:
                System.out.println("------------->>>>2");
                break;
            default:
                System.out.println("------------->>>>default");
                break;
        }
    }

    public static void main(String[] args) {
        LiteJobConfiguration jobConfiguration = new MyElasticJob().createJobConfiguration();
        new JobScheduler(createRegistryCenter(), jobConfiguration).init();
    }

    private static CoordinatorRegistryCenter createRegistryCenter() {
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration("177.17.161.215:2181", "elastic-job-demo"));
        regCenter.init();
        return regCenter;
    }

    protected LiteJobConfiguration createJobConfiguration() {
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder("demoSimpleJob", "0/5 * * * * ?", 4).shardingItemParameters("0=A,1=B,2=C,3=D").build();
        // 定义SIMPLE类型配置
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, MyElasticJob.class.getCanonicalName());
        // 定义Lite作业根配置
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
        return simpleJobRootConfig;
    }

}

运行ok

查看远端zookeeper，

比之前多了一个

命名空间。

但是，如果此时将代码改为

       public void execute(ShardingContext context) {
        System.out.println(context.toString());
        switch (context.getShardingParameter()) {
            case "A":
                System.out.println("------------->>>>0");
                break;
            case "B":
                System.out.println("------------->>>>1");
                break;
            case "C":
                System.out.println("------------->>>>2");
                break;
            default:
                System.out.println("------------->>>>default");
                break;
        }
    }

此时会报空指针异常，为什呢？

按理说，我赋值的是shardingparameter，但是结果确实jobparameter，因为我一开始使用了jobparameter，然后改成shardingparameter

跟踪源码，发现一开始都没问题

但是到了

发现我的sharadingparameter 已经变为了 jobparameter了

因为代码是直接从zk上读了。

我们去zk上看下

果然是从zk上直接读的。我们换个名字看下

发现又可以了。说明是zk记录了第一次的配置，后面就直接从zk上拿了。所以，有一个配置

增加后，我们再看一下

发现也没问题了。

再次看一下zk，发现有两个job了。

elastic job 实现原理

弹性分布式实现

第一台服务器上线触发主服务器选举。主服务器一旦下线，则重新触发选举，选举过程中阻塞，只有主服务器选举完成，才会执行其他任务。
某作业服务器上线时会自动将服务器信息注册到注册中心，下线时会自动更新服务器状态。
主节点选举，服务器上下线，分片总数变更均更新重新分片标记。
定时任务触发时，如需重新分片，则通过主服务器分片，分片过程中阻塞，分片结束后才可执行任务。如分片过程中主服务器下线，则先选举主服务器，再分片。
通过上一项说明可知，为了维持作业运行时的稳定性，运行过程中只会标记分片状态，不会重新分片。分片仅可能发生在下次任务触发前。
每次分片都会按服务器IP排序，保证分片结果不会产生较大波动。
实现失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。

注册中心数据结构

注册中心在定义的命名空间下，创建作业名称节点，用于区分不同作业，所以作业一旦创建则不能修改作业名称，如果修改名称将视为新的作业。
作业名称节点下又包含4个数据子节点，分别是config, instances, sharding, servers和leader。

config节点

作业配置信息，以JSON格式存储刚刚，我们也看到了。

instances节点

作业运行实例信息，子节点是当前作业运行实例的主键。作业运行实例主键由作业运行服务器的IP地址和PID构成。作业运行实例主键均为临时节点，当作业实例上线时注册，下线时自动清理。注册中心监控这些节点的变化来协调分布式作业的分片以及高可用。可在作业运行实例节点写入TRIGGER表示该实例立即执行一次。当我们运行job时候，那么就会有。

如果下线那么就没有

sharding节点

作业分片信息，子节点是分片项序号，从零开始，至分片总数减一。分片项序号的子节点存储详细信息。每个分片项下的子节点用于控制和记录分片运行状态。节点详细信息说明：

servers节点

作业服务器信息，子节点是作业服务器的IP地址。可在IP地址节点写入DISABLED表示该服务器禁用。在新的cloud native架构下，servers节点大幅弱化，仅包含控制服务器是否可以禁用这一功能。为了更加纯粹的实现job核心，servers功能未来可能删除，控制服务器是否禁用的能力应该下放至自动化部署系统。

目前就我跑得一台机器。