远程调度增强校验:1、调度中心与执行器时间偏移超过60秒拒绝任务执行;2、执行器校验调度uuid避免一次调度重复执行;
This commit is contained in:
parent
53fd312d37
commit
ee9eca0db7
|
@ -57,6 +57,7 @@ public class RemoteHttpJobBean extends QuartzJobBean {
|
||||||
HashMap<String, String> params = new HashMap<String, String>();
|
HashMap<String, String> params = new HashMap<String, String>();
|
||||||
params.put(HandlerRepository.TRIGGER_LOG_URL, PropertiesUtil.getString(HandlerRepository.TRIGGER_LOG_URL));
|
params.put(HandlerRepository.TRIGGER_LOG_URL, PropertiesUtil.getString(HandlerRepository.TRIGGER_LOG_URL));
|
||||||
params.put(HandlerRepository.TRIGGER_LOG_ID, String.valueOf(jobLog.getId()));
|
params.put(HandlerRepository.TRIGGER_LOG_ID, String.valueOf(jobLog.getId()));
|
||||||
|
params.put(HandlerRepository.TRIGGER_TIMESTAMP, String.valueOf(System.currentTimeMillis()));
|
||||||
params.put(HandlerRepository.HANDLER_NAME, jobDataMap.get(HandlerRepository.HANDLER_NAME));
|
params.put(HandlerRepository.HANDLER_NAME, jobDataMap.get(HandlerRepository.HANDLER_NAME));
|
||||||
params.put(HandlerRepository.HANDLER_PARAMS, jobDataMap.get(HandlerRepository.HANDLER_PARAMS));
|
params.put(HandlerRepository.HANDLER_PARAMS, jobDataMap.get(HandlerRepository.HANDLER_PARAMS));
|
||||||
|
|
||||||
|
|
|
@ -22,6 +22,7 @@ public class HandlerRepository {
|
||||||
|
|
||||||
public static final String TRIGGER_LOG_ID = "trigger_log_id";
|
public static final String TRIGGER_LOG_ID = "trigger_log_id";
|
||||||
public static final String TRIGGER_LOG_URL = "trigger_log_url";
|
public static final String TRIGGER_LOG_URL = "trigger_log_url";
|
||||||
|
public static final String TRIGGER_TIMESTAMP = "trigger_timestamp";
|
||||||
|
|
||||||
public static ConcurrentHashMap<String, HandlerThread> handlerTreadMap = new ConcurrentHashMap<String, HandlerThread>();
|
public static ConcurrentHashMap<String, HandlerThread> handlerTreadMap = new ConcurrentHashMap<String, HandlerThread>();
|
||||||
|
|
||||||
|
@ -41,6 +42,13 @@ public class HandlerRepository {
|
||||||
RemoteCallBack callback = new RemoteCallBack();
|
RemoteCallBack callback = new RemoteCallBack();
|
||||||
callback.setStatus(RemoteCallBack.FAIL);
|
callback.setStatus(RemoteCallBack.FAIL);
|
||||||
|
|
||||||
|
// encryption check
|
||||||
|
long timestamp = _param.get(HandlerRepository.TRIGGER_TIMESTAMP)!=null?Long.valueOf(_param.get(HandlerRepository.TRIGGER_TIMESTAMP)):-1;
|
||||||
|
if (System.currentTimeMillis() - timestamp > 60000) {
|
||||||
|
callback.setMsg("Timestamp check failed.");
|
||||||
|
return JacksonUtil.writeValueAsString(callback);
|
||||||
|
}
|
||||||
|
|
||||||
// push data to queue
|
// push data to queue
|
||||||
String handler_name = _param.get(HandlerRepository.HANDLER_NAME);
|
String handler_name = _param.get(HandlerRepository.HANDLER_NAME);
|
||||||
if (handler_name!=null && handler_name.trim().length()>0) {
|
if (handler_name!=null && handler_name.trim().length()>0) {
|
||||||
|
|
|
@ -7,6 +7,7 @@ import java.util.Map;
|
||||||
import java.util.concurrent.LinkedBlockingQueue;
|
import java.util.concurrent.LinkedBlockingQueue;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
import org.eclipse.jetty.util.ConcurrentHashSet;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
|
@ -23,14 +24,18 @@ public class HandlerThread extends Thread{
|
||||||
|
|
||||||
private IJobHandler handler;
|
private IJobHandler handler;
|
||||||
private LinkedBlockingQueue<Map<String, String>> handlerDataQueue;
|
private LinkedBlockingQueue<Map<String, String>> handlerDataQueue;
|
||||||
|
private ConcurrentHashSet<String> logIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID
|
||||||
|
|
||||||
public HandlerThread(IJobHandler handler) {
|
public HandlerThread(IJobHandler handler) {
|
||||||
this.handler = handler;
|
this.handler = handler;
|
||||||
handlerDataQueue = new LinkedBlockingQueue<Map<String,String>>();
|
handlerDataQueue = new LinkedBlockingQueue<Map<String,String>>();
|
||||||
|
logIdSet = new ConcurrentHashSet<String>();
|
||||||
}
|
}
|
||||||
|
|
||||||
public void pushData(Map<String, String> param) {
|
public void pushData(Map<String, String> param) {
|
||||||
handlerDataQueue.offer(param);
|
if (param.get(HandlerRepository.TRIGGER_LOG_ID)!=null && !logIdSet.contains(param.get(HandlerRepository.TRIGGER_LOG_ID))) {
|
||||||
|
handlerDataQueue.offer(param);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
int i = 1;
|
int i = 1;
|
||||||
|
@ -38,12 +43,13 @@ public class HandlerThread extends Thread{
|
||||||
public void run() {
|
public void run() {
|
||||||
while(true){
|
while(true){
|
||||||
try {
|
try {
|
||||||
i++;
|
|
||||||
Map<String, String> handlerData = handlerDataQueue.poll();
|
Map<String, String> handlerData = handlerDataQueue.poll();
|
||||||
if (handlerData!=null) {
|
if (handlerData!=null) {
|
||||||
|
i= 0;
|
||||||
String trigger_log_url = handlerData.get(HandlerRepository.TRIGGER_LOG_URL);
|
String trigger_log_url = handlerData.get(HandlerRepository.TRIGGER_LOG_URL);
|
||||||
String trigger_log_id = handlerData.get(HandlerRepository.TRIGGER_LOG_ID);
|
String trigger_log_id = handlerData.get(HandlerRepository.TRIGGER_LOG_ID);
|
||||||
String handler_params = handlerData.get(HandlerRepository.HANDLER_PARAMS);
|
String handler_params = handlerData.get(HandlerRepository.HANDLER_PARAMS);
|
||||||
|
logIdSet.remove(trigger_log_id);
|
||||||
|
|
||||||
// parse param
|
// parse param
|
||||||
String[] handlerParams = null;
|
String[] handlerParams = null;
|
||||||
|
@ -68,9 +74,8 @@ public class HandlerThread extends Thread{
|
||||||
// callback handler info
|
// callback handler info
|
||||||
RemoteCallBack callback = null;
|
RemoteCallBack callback = null;
|
||||||
try {
|
try {
|
||||||
|
|
||||||
HashMap<String, String> params = new HashMap<String, String>();
|
HashMap<String, String> params = new HashMap<String, String>();
|
||||||
params.put(HandlerRepository.TRIGGER_LOG_ID, trigger_log_id);
|
params.put("trigger_log_id", trigger_log_id);
|
||||||
params.put("status", _status.name());
|
params.put("status", _status.name());
|
||||||
params.put("msg", _msg);
|
params.put("msg", _msg);
|
||||||
callback = HttpUtil.post(trigger_log_url, params);
|
callback = HttpUtil.post(trigger_log_url, params);
|
||||||
|
@ -80,6 +85,8 @@ public class HandlerThread extends Thread{
|
||||||
logger.info("<<<<<<<<<<< xxl-job thread handle, handlerData:{}, callback_status:{}, callback_msg:{}, callback:{}, thread:{}",
|
logger.info("<<<<<<<<<<< xxl-job thread handle, handlerData:{}, callback_status:{}, callback_msg:{}, callback:{}, thread:{}",
|
||||||
new Object[]{handlerData, _status, _msg, callback, this});
|
new Object[]{handlerData, _status, _msg, callback, this});
|
||||||
} else {
|
} else {
|
||||||
|
i++;
|
||||||
|
logIdSet.clear();
|
||||||
try {
|
try {
|
||||||
TimeUnit.MILLISECONDS.sleep(i * 100);
|
TimeUnit.MILLISECONDS.sleep(i * 100);
|
||||||
} catch (InterruptedException e) {
|
} catch (InterruptedException e) {
|
||||||
|
|
Loading…
Reference in New Issue