Java應用多機器部署解決大量定時任務問題
今天來說一個Java多機部署下定時任務的處理方案。
需求: 有兩臺服務器同時部署了同一套代碼, 代碼中寫有spring自帶的定時任務,但是每次執(zhí)行定時任務時只需要一臺機器去執(zhí)行。
當拿到這個需求時我腦子中立馬出現了兩個簡單的解決方案:
- 利用ip進行判斷, 兩臺機器ip肯定不一樣, 指定某一臺機器的ip運行。
- 只在一臺機器上部署定時任務的代碼。
最后兩個方案又都被自己否決了。 第一條,如果指定ip的機器出現了問題怎么辦? 例如說宕機了, 那么該制定ip的機器上的定時任務是不是就無法運行了?如果以后該服務器遷移導致ip變化怎么辦?
第二條, 同上, 還有就是要維護兩套代碼很不方便。
因為上面兩個假設都不成立, 只能另找他法。 于是便想到利用mysql去解決, 之前了解過一點mysql的鎖機制, 知道如果有同時的兩個任務去寫數據庫中同一條記錄, 只有一條會成功, 這是利用了mysql的排他鎖。
下面就開始代碼演示, 這里主要想給大家的是一個思路的提示, 代碼還是很簡單的。
首先需要單獨創(chuàng)建一張表
CREATE TABLE `t_schedule_cluster` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '@cname:主鍵', `execute` int(1) NOT NULL COMMENT '@cname:執(zhí)行狀態(tài)', `version` int(11) NOT NULL COMMENT '@cname:版本號\r\n ', `task_name` varchar(128) NOT NULL COMMENT '@cname:任務名稱\r\n ', `execute_ip` varchar(32) DEFAULT NULL COMMENT '@cname:執(zhí)行ip\r\n ', `update_time` datetime DEFAULT NULL COMMENT '@cname:修改時間\r\n ', PRIMARY KEY (`id`), KEY `Index_series_id` (`execute`) ) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8 COMMENT='@cname:多機定時任務調度';
看一下建成后的表結構, 注釋寫的已經很清楚了, 初始化時需要添加一些定時任務的名稱(task_name), 這個要和你代碼中保持一致, 后面會提到:

代碼
首先看下我代碼中用到的spring定時任務:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd"
default-lazy-init="true">
<description>使用Spring的 Scheduled的定時任務配置</description>
<!--支持annotation的方式-->
<task:annotation-driven />
<task:scheduler id="springScheduler" pool-size="10"/>
<task:scheduled-tasks scheduler="springScheduler">
<!-- 測試使用, 項目啟動后每隔一分鐘執(zhí)行一次 -->
<task:scheduled ref="listCarAction" method="listCar" cron="0 0/1 0 * * ?"/>
<task:scheduled ref="listCarAction" method="listCar" cron="0 0/1 0 * * ?"/>
</task:scheduled-tasks>
</beans>
相信大家都是用過這種定時任務的設置方法, 因為它是spring自帶的, 所以使用起來很方便, 這里我指定了兩個定時任務來模擬兩臺機器的情況, 兩個定時任務都是項目啟動后每隔一分鐘執(zhí)行一次。
然后看看這個listCar中的代碼:
//定時任務的名稱, 這個和數據庫中的task_name是保持一致的, 保證要執(zhí)行該定時任務。
public static final String LIST_CAR_TASK = "listCarTask";
private ScheduleClusterTask scheduleClusterTask;
//這個時間是根據spring-scheduler.xml中配置的定時刷新時間, 比如說我們以后要設置這個定時任務時4小時刷新一次
public static final long maxExpireTime = 4 * 3600;
public void listCar() {
if (scheduleClusterTask.isValidMachine(maxExpireTime, CommonConstants.ScheduleTaskName.LIST_CAR_TASK)) {
//執(zhí)行具體的task方法,
doTask();
//將execute狀態(tài)更新為0
scheduleClusterTask.end(LIST_CAR_TASK);
}
}
最后看下最核心的代碼:ScheduleClusterTask.java
/**
* 多機定時任務工具類
* Created by WangMeng on 2017/4/12.
*/
@Service
public class ScheduleClusterTask {
private ScheduleClusterEntityService scheduleClusterEntityService;
/**
* 這里因為兩臺機器都有同樣的定時任務, 會同時執(zhí)行這個方法,只有一臺機器可以執(zhí)行成功,返回true。
* @param maxExpireTime 最大的檢查時間。
* @param taskName 任務名稱。
* @return
*/
public boolean isValidMachine(long maxExpireTime, String taskName) {
boolean isValid = false;
try {
//通過taskName去數據庫中查找到該條記錄, 如果大家使用的是mybatis這里需要改一下, 就是一個簡單的查詢操作
ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName));
int execute = carIndexEntity.getExecute();
String ip = InetAddress.getLocalHost().getHostAddress();
long currentTimeMillis = System.currentTimeMillis();
long time = carIndexEntity.getUpdateTime().getTime();
if (execute == 0 && time + maxExpireTime - 1000 < currentTimeMillis) {
isValid = checkMachine(taskName, carIndexEntity, ip);
} else if (time + maxExpireTime - 1000 < currentTimeMillis){
//這里要判斷下, 如果上一次執(zhí)行出現異常導致execute沒有更新為0, 那么這里要判斷上一次更新時間的間隔。
isValid = checkMachine(taskName, carIndexEntity, ip);
}
} catch (UnknownHostException e) {
e.printStackTrace();
}
return isValid;
}
/**
* end方法主要是將excute(是否正在執(zhí)行的標志位,0:沒有執(zhí)行, 1:正在執(zhí)行)更新為0
* @param taskName
* @return
*/
public boolean end (String taskName) {
ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName));
//將execute狀態(tài)更新為0
return scheduleClusterEntityService.end(carIndexEntity);
}
private boolean checkMachine(String taskName, ScheduleClusterEntity carIndexRefresh, String ip) {
return scheduleClusterEntityService.start(taskName, carIndexRefresh.getVersion(), ip);
}
@Autowired
public void setScheduleClusterEntityService(ScheduleClusterEntityService scheduleClusterEntityService) {
this.scheduleClusterEntityService = scheduleClusterEntityService;
}
}
這里還有start方法, 看看怎樣的操作:
@Repository
public class DefaultScheduleClusterEntityDao extends AbstractDao<ScheduleClusterEntity> implements ScheduleClusterEntityDao {
@Override
public boolean start(String taskName, int version, String ip) {
String sql = "update t_schedule_cluster set execute = 1, " +
"version = ?, execute_ip = ?, update_time = ?" +
" where task_name = ? and version = ?";
Sql s = new Sql(sql);
s.addParam(version + 1);
s.addParam(ip);
s.addParam(SqlTimeUtils.nowTimestamp());
s.addParam(taskName);
s.addParam(version);
return 1 == executeUpdate(s);
}
}
核心的代碼到了這里就沒有了, 代碼確實是非常非常的簡單, 有興趣的話大家可以在本地測試一下就可以。
當然還有更多很好地解決方案, 我這里秉承的是最簡單的處理方式。
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持腳本之家。
相關文章
java中排序報:Comparison method violates its general contract異常的解
這篇文章主要給大家介紹了關于java中排序報:Comparison method violates its general contract異常的解決方法,文中介紹的非常詳細,對大家具有一定的參考學習價值,需要的朋友們下面來一起看看吧。2017-06-06
mybatis Interceptor對UpdateTime自動處理的實現方法
這篇文章主要給大家介紹了關于使用mybatis Interceptor對UpdateTime自動處理的實現方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面來一起看看吧2018-12-12

