qq交流羣:812321371
mysql
Elastic-Job
是一個分佈式調度解決方案,由兩個相互獨立的子項目Elastic-Job-Lite
和Elastic-Job-Cloud
組成。Elastic-Job-Lite
定位爲輕量級無中心化解決方案,使用jar包的形式提供分佈式任務的協調服務。基於quartz
定時任務框架爲基礎的,所以具有quartz
的大部分功能使用zookeeper
作協調,調度中心,更加輕量級支持任務的分片支持彈性擴容,能夠水平擴展, 當任務再次運行時,會檢查當前的服務器數量,從新分片,分片結束以後纔會繼續執行任務失效轉移,容錯處理,當一臺調度服務器宕機或者跟zookeeper
斷開鏈接以後,會當即中止做業,而後再去尋找其餘空閒的調度服務器,來運行剩餘的任務提供運維界面,能夠管理做業和註冊中心。git
因爲項目爲微服務,單模塊可能在兩個實例以上的數量,定時器就會出現多實例同時執行的狀況。通常定時器缺乏管理界面,沒法監控定時器是否執行成功。市面上常見的解決方案爲定時器加鎖的操做,或者採用第3方分佈式定時器。分佈式定時器有多種方案,好比阿里內部的ScheduledX
,噹噹網的Elastic job
,我的開源的xxl-job
等。github
分片:任務的分佈式執行,須要將一個任務拆分爲多個獨立的任務項,而後由分佈式的服務器分別執行某一個或幾個分片項。例如:有一個遍歷數據庫某張表的做業,現有2臺服務器。爲了快速的執行做業,那麼每臺服務器應執行做業的50%
。 爲知足此需求,可將做業分紅2片,每臺服務器執行1片。做業遍歷數據的邏輯應爲:服務器A遍歷ID以奇數結尾的數據;服務器B遍歷ID以偶數結尾的數據。 若是分紅10片,則做業遍歷數據的邏輯應爲:每片分到的分片項應爲ID%10,而服務器A被分配到分片項0,1,2,3,4
;服務器B被分配到分片項5,6,7,8,9
,直接的結果就是服務器A遍歷ID
以0-4
結尾的數據;服務器B遍歷ID
以5-9
結尾的數據。spring
歷史軌跡:Elastic-Job
提供了事件追蹤功能,可經過事件訂閱的方式處理調度過程的重要事件,用於查詢、統計和監控。sql
elasticjob
因爲噹噹網Elastic job
處於1年間未更新階段,相關jar處於可使用階段功能不全。考慮到使用場景爲多項目使用,將elastic-job-lite-spring
簡單封裝便於使用。數據庫
ps:實際version版本請使用最新版
bootstrap
<dependency>
<groupId>com.purgeteam</groupId>
<artifactId>elasticjob-spring-boot-starter</artifactId>
<version>0.1.1.RELEASE</version>
</dependency>複製代碼
ps: 須要mysql
,zookeeper
支持,請提早搭建好。服務器
配置bootstrap.yml
或者application.yml
。app
加入如下配置:框架
spring:
elasticjob:
datasource: # job須要的記錄數據源
url: jdbc:mysql://127.0.0.1:3306/batch_log?useUnicode=true&characterEncoding=utf-8&verifyServerCertificate=false&useSSL=false&requireSSL=false
driver-class-name: com.mysql.cj.jdbc.Driver
username: root
password: Rtqw123OpnmER
regCenter: # 註冊中心
serverList: 127.0.0.1:2181
namespace: elasticJobDemo複製代碼
建立定時器類(惟一不一樣的地方在於將@Scheduled
改成實現SimpleJob
接口便可)定時器實現方法編寫在execute
方法裏。
@Slf4j
@Component
public class MySimpleJob implements SimpleJob {
// @Scheduled(cron = "0 0/1 * * * ?")
@Override
public void execute(ShardingContext shardingContext) {
log.info(String.format("Thread ID: %s, 做業分片總數: %s, " +
"當前分片項: %s.當前參數: %s," +
"做業名稱: %s.做業自定義參數: %s",
Thread.currentThread().getId(),
shardingContext.getShardingTotalCount(),
shardingContext.getShardingItem(),
shardingContext.getShardingParameter(),
shardingContext.getJobName(),
shardingContext.getJobParameter()
));
// 分片大體以下:根據配置的分片參數執行相應的邏輯
switch (context.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: ...
}
}
}複製代碼
log:Thread ID: 66, 做業分片總數: 1, 當前分片項: 0.當前參數: Beijing,做業名稱: PropertiesSimpleJob.做業自定義參數: test複製代碼
將ZookeeperRegistryCenter
和JobEventConfiguration
注入。建立JobScheduler
@Bean(initMethod = "init")
。在mySimpleJobScheduler
方法裏先經過ElasticJobUtils#getLiteJobConfiguration
獲取LiteJobConfiguration
對象。建立SpringJobScheduler
對象返回便可。
@Configuration
public class MyJobConfig {
// job 名稱
private static final String JOB_NAME = "MySimpleJob";
// 定時器cron參數
private static final String CRON = "0 0/1 * * * ?";
// 定時器分片
private static final int SHARDING_TOTAL_COUNT = 1;
// 分片參數
private static final String SHARDING_ITEM_PARAMETERS = "0=Beijing,1=Shanghai,2=Guangzhou";
// 自定義參數
private static final String JOB_PARAMETERS = "parameter";
@Resource
private ZookeeperRegistryCenter regCenter;
@Resource
private JobEventConfiguration jobEventConfiguration;
@Bean(initMethod = "init")
public JobScheduler mySimpleJobScheduler(final MySimpleJob mySimpleJob) {
LiteJobConfiguration liteJobConfiguration = ElasticJobUtils
.getLiteJobConfiguration(mySimpleJob.getClass(), JOB_NAME, CRON,
SHARDING_TOTAL_COUNT, SHARDING_ITEM_PARAMETERS, JOB_PARAMETERS);
// 參數:1.定時器實例,2.註冊中心類,3.LiteJobConfiguration,
// 3.歷史軌跡(不須要能夠省略)
return new SpringJobScheduler(mySimpleJob, regCenter, liteJobConfiguration, jobEventConfiguration);
}
}複製代碼
ElasticJobUtils#getLiteJobConfiguration
參數簡介:
/**
* 獲取 {@link LiteJobConfiguration} 對象
*
* @param jobClass 定時器實現類
* @param jobName 定時器名稱
* @param cron 定時參數
* @param shardingTotalCount 做業分片總數
* @param shardingItemParameters 當前參數 能夠爲null
* @param jobParameters 做業自定義參數 能夠爲null
* @return {@link LiteJobConfiguration}
*/
public static LiteJobConfiguration getLiteJobConfiguration(
final Class<? extends SimpleJob> jobClass,
final String jobName,
final String cron,
final int shardingTotalCount,
final String shardingItemParameters,
final String jobParameters) {
...
return ...;
}複製代碼
固然也能夠用下面的@Configuration
實現簡化,配置bootstrap.yml
或者application.yml
。
spring:
elasticjob:
scheduled:
jobConfigMap: // 爲map集合
PropertiesSimpleJob: // 定時器key名稱
jobName: PropertiesSimpleJob // job名稱
cron: 0 0/1 * * * ? // cron表達式
shardingTotalCount: 2 // 分片數量
shardingItemParameters: 0=123,1=332 // 分片參數
jobParameters: test // 自定義參數複製代碼
注入SpringJobSchedulerFactory
,在propertiesSimpleJobScheduler
方法裏調用gerSpringJobScheduler
方法便可。
@Configuration
public class PropertiesSimpleJobConfig {
@Resource
private SpringJobSchedulerFactory springJobSchedulerFactory;
@Bean(initMethod = "init")
public JobScheduler propertiesSimpleJobScheduler(final PropertiesSimpleJob job) {
// 參數:1.定時器實例,2.配置名稱,3.是否開啓歷史軌跡
return springJobSchedulerFactory.getSpringJobScheduler(job,"PropertiesSimpleJob", true);
}
}複製代碼
ps:這個註解包含了上述方式,簡化定時器注入。
繼承SimpleJob
實現方法execute
。
在AnnotationSimpleJob
類上加入註解@ElasticJobScheduler
便可。下面爲完整註解。
@Slf4j
@ElasticJobScheduler(
name = "AnnotationSimpleJob", // 定時器名稱
cron = "0/8 * * * * ?", // 定時器表達式
shardingTotalCount = 1, // 做業分片總數 默認爲1
shardingItemParameters = "0=Beijing,1=Shanghai,2=Guangzhou", // 分片序列號和參數用等號分隔 不須要參數能夠不加
jobParameters = "123", // 做業自定義參數 不須要參數能夠不加
isEvent = true // 是否開啓數據記錄 默認爲true
)
public class AnnotationSimpleJob implements SimpleJob {
@Override
public void execute(ShardingContext shardingContext) {
log.info(String.format("Thread ID: %s, 做業分片總數: %s, " +
"當前分片項: %s.當前參數: %s," +
"做業名稱: %s.做業自定義參數: %s",
Thread.currentThread().getId(),
shardingContext.getShardingTotalCount(),
shardingContext.getShardingItem(),
shardingContext.getShardingParameter(),
shardingContext.getJobName(),
shardingContext.getJobParameter()
));
}
}複製代碼
分佈式job能夠解決多個項目同一個定時器都執行的問題,配合elastic-job控制檯能夠直觀監控定時器執行狀況等。
示例代碼地址:elastic-job-spring-boot
做者GitHub:
Purgeyao 歡迎關注本文由博客一文多發平臺 OpenWrite 發佈!