分布式任务调度框架--SkySchedule介绍

 

概述

 

SkySchedule是基于netty实现的分布式任务调度框架，不依赖zookeeper等其他服务。主要原理是：客户端启动时通过netty与SkySchedule服务端建立长连接，通过长连接发送心跳消息，服务端可以统计到当前存活的客户端列表，为每个客户端分配“任务编号”。通过“任务id”对“客户端总数”取模，每个客户端获取对应“任务编号”的任务，实现分布式任务调度。

 

该分布式任务调度适用场景

 

生产者：假设有一个系统时刻都在生成任务，这些任务不会立即执行，需要暂时存放到一个mysql任务表表。

 

任务表：每个任务以一条记录的形式存放在msyql任务表中，表结构大致如下：

列名	类型	描述
id	Int	连续自增主键
task_data	varchar(50)	任务数据
task_time	datetime	任务执行时间
status	Int	状态：0 新建 1执行成功 2-执行失败

 

消费者：为了防止任务积压，并且能在最短的时间内把已经到期的任务执行完，我们需要多台机器并行执行这些任务。

 

分布式任务调度器SkySchedule职责：

1、通过分布式调度，把任务均分到各个消费者服务器。

2、保证任务100%被执行。一般情况下任务执行有三种情况（跟mq的消息处理类似）：

At most once 任务可能没有执行，但绝不会重复执行。

At least one 任务绝对会执行，但可能会重复执行。

Exactly once 每条任务肯定会被执行一次且仅执行一次，很多时候这是用户所想要的。

由于在我所在项目的业务场景中，允许重复执行，目前SkySchedule只支持到At least one。Exactly once后续会提供。

3、某台消费者服务器挂掉，能自动重新分配任务到其他消费者服务器（重新分配的这部分任务执行可能会有短时间延迟）。如果增加消费者服务器，也能对剩余的任务自动重新分配。

4、为了防止单点故障，SkySchedule服务需要部署到多台服务器上的，如果其中某台挂掉或出现故障，存活的SkySchedule服务器能正常对外服务。

 

部署方式：

 

 

多分组集群模式：

这种方式，理论上可以为任意数量的系统提供分布式任务调度服务。

SkySchedule以集群方式部署，并做分组，比如：每三台server做为一个分组。每个分组，可以为多个系统提供分布式调度服务。上图红色框部分表示SkySchedule集群。

 

蓝色框部分，表示多个需要分布式任务调度的子系统。以“系统1”为例：“系统1”中的每台服务器启动时，向指定的SkySchedule分组中的每台server建立一个长连接。

 

这种方式需要开发一个集群管理页面来管理自己的服务器列表，并进行分组管理。以及系统接入注册管理页面，为某个系统分配到指定的SkySchedule分组为其服务。这部分相关管理功能没有提供开源源码，可以根据自己公司业务实现。

 

单分组集群模式：

如果只有少量或单个系统需要分布式任务调度服务，可以把SkySchedule只部署在少量服务器上（比如三台，防止单点故障），并把服务ip和端口整理成配置文件。

在需要分布式调度服务的系统中，引入该配置文件。程序启动时读取配置文件，与SkySchedule服务端建立长连接。

 

目前SkySchedule服务端只实现了“单分组集群模式”，多分组模式还需要添加相关管理功能才能实现。

以下内容都是基于“单分组集群模式”进行讲解。

 

客户端任务分配：

1、在需要任务调度的“子系统”中引入SkySchedule客户端jar包。

2、引入SkySchedule服务ip、端口列表配置文件。

3、使用jar包中ClientNode中的两个常量来调整从mysql任务表获取任务的sql语句

ClientNode. totalNode : 表示“总节点数”，也就是所有存活的“消费者客户端”个数。

ClientNode. nodeNum : 表示当前客户端“任务编号”。

当某个“消费者客户端”挂掉或新增时，SkySchedule服务端会计算最新的“总节点数”和每台客户端对应的“任务编号”，同步给每个客户端。

每台客户端获取已到期任务sql语句为：

Select * from task_info where (id % totalNode) = nodeNum and task_time < now()

task_info为任务表名，where条件有个取模操作，表示只取自己的属于任务。具体原理如下：

 

任务分配原理：

为了方便理解 举个例子：假如当前总共有3个客户端，分别的：

对于每个客户端而言totalNode都为3。

对于“客户端1”其任务编号nodeNum=0

对于“客户端2”其任务编号nodeNum=1

对于“客户端3”其任务编号nodeNum=2

 

在通过使用msyql任务表中的任务id对3取模，

如果结果为0，说明该任务会分配给“客户端1”执行，

如果结果为1，说明该任务会分配给“客户端2”执行，

如果结果为2，说明该任务会分配给“客户端3”执行，

提供这种方式就可以保证所有任务会被100%的执行。

另外mysql任务表的id是连续自增的，也可以理论上做到平台分配。

 

这就是SkySchedule分布式任务调度的实现原理，其实跟淘宝的tbschedule原理差不多，只是SkySchedule是通过netty实现，不需要借助其他服务，相对较为轻量。tbschedule需要借助zookeeper才能正常服务。

 

下图为： 生产者、消费者、SkySchedule服务端、SkySchedule客户端关系图：

 

 

 

源码github地址：https://github.com/gantianxing/skySchedule

目前只是初始版，能满足上述分布式任务调度基本需求。同时欢迎意见，后续会抽时间慢慢完善，同时也希望感兴趣的朋友贡献代码。

 

代码使用详解

 

源码中一共有四个模块：sky-server、sky-common、sky-client、sky-test，关系如下：

 

 

1、服务端：涉及到的模块为：sky-server、sky-common，通过maven编译打包会生成一个sky-server-1.0-SNAPSHOT.war的war，直接部署这个war包到jdk1.8+tomcat 8中环境中。注意这里必须是jdk1.8+tomcat 8的环境，如果无法启动检查下端口“9991”是否被占用。

假设一共部署了3台服务器：192.168.1.100、192.168.1.101、192.168.1.102。

 

2、服务端：涉及到的模块为：sky-client、sky-common，通过maven编译打包上述代码，还会生成两个jar包：sky-client-1.0-SNAPSHOT.jar、sky-common-1.0-SNAPSHOT.jar。把这两个jar包引入到需要做分布式调度的“应用服务工程”中。再进行少量配置即可，配置过程如下：

 

sky-test是一个模拟测试“应用服务工程”，必须为spring工程，建议spring使用4.0以上（sky-test使用的是4.3.1.RELEASE）。创建SkySchedule-client.properties，内容为：

#SkySchedule 服务ip端口列表 多个以","间隔
server.ip.port = localhost:9991,localhost:9992
 
#如果服务端挂掉,重连服务端间隔时间，单位秒
RE.CONN.WAIT.SECONDS=5
 
#向服务端发现心跳请求间隔时间，单位秒
TASK.REQ.WAIT.SECONDS=15
 
#25秒没有收到服务器返回，断开链接，放到重连map
READ.WAIT.SECONDS=25
 
#如果空闲20秒发送一次ping信息
WRITE.WAIT.SECONDS=20
 
#系统编号
group.id=1000
 
#用户名
user.name=moon
 
#密码
password=walker
 

 

把server.ip.port属性改为你自己的SkySchedule服务端列表，其他属性可以保持不变。比如把server.ip.port改为上述提到的：192.168.1.100:9991,192.168.1.101:9991,192.168.1.102:9991。

如果有多个不同类型的应用需要接入SkySchedule，需要对每个应用指定一个不同的“系统编号”group.id，默认是1000，如果只有一个系统需要接入，可以不用修改。

 

然后需要解析SkySchedule-client.properties到spring的Environment中，可以使用xml装配方式，也可以使用spring bean装配方式。由于sky-test构造的无web.xml配置的web工程，这里采用的后者，装配方式如下：

@Configuration
@ComponentScan(basePackages = {"com.sky.schedule.client"})
@PropertySource("classpath:SkySchedule-client.properties")
public class RootConfig {
 
    @Bean
    public PropertySourcesPlaceholderConfigurer placeholderConfigurer(){
        return new PropertySourcesPlaceholderConfigurer();
    }
}

启动多个sky-test实例，每个实例分配的totalNode、nodeNum是否正常。

 

测试

 

下面我们直接使用源码中的代码进行测试，启动一个sky-server(SkySchedule服务端)；依次启动两个sky-test，用来模拟两个需要分布式调度的“应用服务端”(SkySchedule客户端)。下面是测试步骤：

 

1、启动一个sky-test：观察sky-server服务端日志，每隔20秒，会打印一条消息：

return task info groupId:1000,total:1,nodeNum:0

说明当前客户端总节点数为1，当前客户端“任务编号”为0，也就是说所有任务都由该客户端执行。

 

2、再启动一个sky-test：观察sky-server服务端日志，每隔20秒，会打印两条消息：

return task info groupId:1000,total:2,nodeNum:1

return task info groupId:1000,total:2,nodeNum:0

说明当前客户端数为2,两个客户端的“任务编号”分别为0、1，也就是说一个客户端会执行id尾数为：0,2,4,6,8的任务，另一个客户端会执行id尾数为：1,3,5,7,9。

动态的增加客户端，会重新分配任务，测试通过。

 

3、停止其中一个sky-test（模拟其中一个客户端挂掉）：观察sky-server服务端日志，每隔20秒，只会打印一条消息：

return task info groupId:1000,total:1,nodeNum:0

模拟其中一个客户端挂掉，会重新分配任务，测试通过。

 

通过这三个测试，可以发现当客户端个数发生变化时，SkySchedule会动态的重写分配任务。

 

另外，如果你是直接运行github中的代码，没有修改SkySchedule-client.properties中的server.ip.port属性。在进行上述测试时，细心的你会发现每隔5秒会打印一条错误信息：

ERROR netty.NettyClient - connect server failed-localhost:9992

 

主要原因是server.ip.port配置了两个服务端：localhost:9991,localhost:9992。但实际上我们只部署一个localhost:9991实例。假如你再启动一个localhost:9992服务端实例，这个错误会消失。这里模拟的就是：如果一个SkySchedule服务端挂掉（或者重启），恢复正常后，客户端能自动重连。

 

也就是说客户端在启动时，至少要保证有一个SkySchedule服务端是启动的。在客户端已经启动的情况下，假如所有服务端都挂掉，也不影响，只是这时如果一个客户端挂掉，就无法实现任务动态分配了。简单的说，只要有一个SkySchedule服务端存活就能保证分布式任务调度正常运行，但最好启动多个SkySchedule服务端，防止单点故障。

 

总结

 

最后总结下，使用SkySchedule做分布式调度，你只需要3步即可完成：

1、启动多个服务端，不用改任何配置，使用默认配置即可。

2、把sky-client-1.0-SNAPSHOT.jar、sky-common-1.0-SNAPSHOT.jar这两个jar包引入到需要分布式任务调度的“应用服务”工程。

3、把sky-test工程下的SkySchedule-client.properties复制到“应用服务”工程的classpath下，修改其中server.ip.port、group.id两个属性。通过xml或者spring bean装配的方式，把配置文件注入到spring容器中（Environment）。

 

由于底层是采用netty建立的长连接，性能非常优异，对“应用系统”来说几乎毫无感知，也无需配置zookeeper等服务。并且服务端代码也非常轻量，感兴趣的朋友可以看下源码。

 

其实现在要实现分布式任务调度的手段其实很多，比如我同事采用redis+mq实现过一个分布式任务调度，但个人觉得依赖的外部基础服务太多，如果redis、mq其中任何一个出现问题就会对应用服务产生影响。但采用SkySchedule就可以完全避免这些问题，并且对应用服务不会产生额外负担，只需维持一个netty长连接即可。

 

最后感谢我的同事“曾老师”对group分组支持部分的完善。

 

在使用过程中有任何问题或建议，请直接留言或者站内信。一周内会不定期回复。

 

SkySchedule server端管理页面使用介绍：http://moon-walker.iteye.com/blog/2391954

 

转载请注明出处：

http://moon-walker.iteye.com/blog/2386504

 

 

 

 

评论

2 楼 moon_walker 2017-08-31  

sxp2558 写道

你好，skySchedule调度 的工作原理 我这边认真分析过，发现基于netty进行的分布式调度效率非常不错，我在此基础上进行了二次开发，新增了netty 触发的流式调度功能，以及不间断的调度工作，比目前ejob 的框架确实轻了不少，在项目中也好开展，目前这套框架还处于雏形，希望作者能够继续开发下去，同时也希望我们能组建一个群，共同探讨后续的开发，本人是在金融公司从事架构开发以及核心库开发工作！ 我的QQ 382925939

已+你QQ

1 楼 sxp2558 2017-08-31  

你好，skySchedule调度 的工作原理 我这边认真分析过，发现基于netty进行的分布式调度效率非常不错，我在此基础上进行了二次开发，新增了netty 触发的流式调度功能，以及不间断的调度工作，比目前ejob 的框架确实轻了不少，在项目中也好开展，目前这套框架还处于雏形，希望作者能够继续开发下去，同时也希望我们能组建一个群，共同探讨后续的开发，本人是在金融公司从事架构开发以及核心库开发工作！ 我的QQ 382925939