Skip to content

X
XXL-JOB
提交 ee07e0b7 authored 作者: xuxueli's avatar xuxueli
浏览文件
操作

调度中心API服务改为自研RPC形式，统一底层通讯模型；

上级 1651c63d
隐藏空白字符变更
内嵌 并排
正在显示 包含 252 行增加 和 395 行删除
doc/XXL-JOB官方文档.md
浏览文件 @ ee07e0b7
......@@ -902,6 +902,7 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段
- 3、执行器JobHandler禁止命名冲突；
- 4、执行器集群地址列表进行自然排序；
- 5、调度中心，DAO层代码精简优化并且新增测试用例覆盖；
- 6、调度中心API服务改为自研RPC形式，统一底层通讯模型；
#### TODO LIST
- 1、任务权限管理：执行器为粒度分配权限，核心操作校验权限；
......
xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobApiController.java
浏览文件 @ ee07e0b7
package com.xxl.job.admin.controller;
import com.xxl.job.admin.controller.annotation.PermessionLimit;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.dao.XxlJobInfoDao;
import com.xxl.job.admin.dao.XxlJobLogDao;
import com.xxl.job.admin.dao.XxlJobRegistryDao;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.AdminApiUtil;
import org.apache.commons.lang.StringUtils;
import org.quartz.SchedulerException;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.rpc.codec.RpcRequest;
import com.xxl.job.core.rpc.codec.RpcResponse;
import com.xxl.job.core.rpc.netcom.NetComServerFactory;
import com.xxl.job.core.rpc.serialize.HessianSerializer;
import com.xxl.job.core.util.HttpClientUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import javax.annotation.Resource;
import java.text.MessageFormat;
import java.util.Date;
import java.util.List;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.OutputStream;
/**
* Created by xuxueli on 17/5/10.
......@@ -33,100 +25,50 @@ import java.util.List;
public class JobApiController {
private static Logger logger = LoggerFactory.getLogger(JobApiController.class);
@Resource
public XxlJobLogDao xxlJobLogDao;
@Resource
private XxlJobInfoDao xxlJobInfoDao;
@Resource
private XxlJobRegistryDao xxlJobRegistryDao;
@RequestMapping(value= AdminApiUtil.CALLBACK, method = RequestMethod.POST, consumes = "application/json")
@ResponseBody
@PermessionLimit(limit=false)
public ReturnT<String> callback(@RequestBody List<HandleCallbackParam> callbackParamList){
for (HandleCallbackParam handleCallbackParam: callbackParamList) {
ReturnT<String> callbackResult = callback(handleCallbackParam);
logger.info("JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
(callbackResult.getCode()==ReturnT.SUCCESS_CODE?"success":"fail"), handleCallbackParam, callbackResult);
}
return ReturnT.SUCCESS;
static {
NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz);
}
private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
// valid log item
XxlJobLog log = xxlJobLogDao.load(handleCallbackParam.getLogId());
if (log == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "log item not found.");
}
// trigger success, to trigger child job, and avoid repeat trigger child job
String childTriggerMsg = null;
if (ReturnT.SUCCESS_CODE==handleCallbackParam.getExecuteResult().getCode() && ReturnT.SUCCESS_CODE!=log.getHandleCode()) {
XxlJobInfo xxlJobInfo = xxlJobInfoDao.loadById(log.getJobId());
if (xxlJobInfo!=null && StringUtils.isNotBlank(xxlJobInfo.getChildJobKey())) {
childTriggerMsg = "<hr>";
String[] childJobKeys = xxlJobInfo.getChildJobKey().split(",");
for (int i = 0; i < childJobKeys.length; i++) {
String[] jobKeyArr = childJobKeys[i].split("_");
if (jobKeyArr!=null && jobKeyArr.length==2) {
XxlJobInfo childJobInfo = xxlJobInfoDao.loadById(Integer.valueOf(jobKeyArr[1]));
if (childJobInfo!=null) {
try {
boolean ret = XxlJobDynamicScheduler.triggerJob(String.valueOf(childJobInfo.getId()), String.valueOf(childJobInfo.getJobGroup()));
private RpcResponse doInvoke(HttpServletRequest request) {
try {
// deserialize request
byte[] requestBytes = HttpClientUtil.readBytes(request);
if (requestBytes == null || requestBytes.length==0) {
RpcResponse rpcResponse = new RpcResponse();
rpcResponse.setError("RpcRequest byte[] is null");
return rpcResponse;
}
RpcRequest rpcRequest = (RpcRequest) HessianSerializer.deserialize(requestBytes, RpcRequest.class);
// add msg
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务成功, 子任务Key: {2}, status: {3}, 子任务描述: {4}",
(i+1), childJobKeys.length, childJobKeys[i], ret, childJobInfo.getJobDesc());
} catch (SchedulerException e) {
logger.error("", e);
}
} else {
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务失败, 子任务xxlJobInfo不存在, 子任务Key: {2}",
(i+1), childJobKeys.length, childJobKeys[i]);
}
} else {
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务失败, 子任务Key格式错误, 子任务Key: {2}",
(i+1), childJobKeys.length, childJobKeys[i]);
}
}
// invoke
RpcResponse rpcResponse = NetComServerFactory.invokeService(rpcRequest, null);
return rpcResponse;
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
RpcResponse rpcResponse = new RpcResponse();
rpcResponse.setError("Server-error:" + e.getMessage());
return rpcResponse;
}
}
// handle msg
StringBuffer handleMsg = new StringBuffer();
if (log.getHandleMsg()!=null) {
handleMsg.append(log.getHandleMsg()).append("<br>");
}
if (handleCallbackParam.getExecuteResult().getMsg() != null) {
handleMsg.append(handleCallbackParam.getExecuteResult().getMsg());
}
if (childTriggerMsg !=null) {
handleMsg.append("<br>子任务触发备注：").append(childTriggerMsg);
}
@RequestMapping("/api")
@PermessionLimit(limit=false)
public void api(HttpServletRequest request, HttpServletResponse response) throws IOException {
// success, save log
log.setHandleTime(new Date());
log.setHandleCode(handleCallbackParam.getExecuteResult().getCode());
log.setHandleMsg(handleMsg.toString());
xxlJobLogDao.updateHandleInfo(log);
// invoke
RpcResponse rpcResponse = doInvoke(request);
return ReturnT.SUCCESS;
}
// serialize response
byte[] responseBytes = HessianSerializer.serialize(rpcResponse);
response.setContentType("text/html;charset=utf-8");
response.setStatus(HttpServletResponse.SC_OK);
//baseRequest.setHandled(true);
@RequestMapping(value=AdminApiUtil.REGISTRY, method = RequestMethod.POST, consumes = "application/json")
@ResponseBody
@PermessionLimit(limit=false)
public ReturnT<String> registry(@RequestBody RegistryParam registryParam){
int ret = xxlJobRegistryDao.registryUpdate(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
if (ret < 1) {
xxlJobRegistryDao.registrySave(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
}
return ReturnT.SUCCESS;
OutputStream out = response.getOutputStream();
out.write(responseBytes);
out.flush();
}
}
xxl-job-admin/src/main/java/com/xxl/job/admin/core/schedule/XxlJobDynamicScheduler.java
浏览文件 @ ee07e0b7
......@@ -8,6 +8,7 @@ import com.xxl.job.admin.dao.XxlJobGroupDao;
import com.xxl.job.admin.dao.XxlJobInfoDao;
import com.xxl.job.admin.dao.XxlJobLogDao;
import com.xxl.job.admin.dao.XxlJobRegistryDao;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.rpc.netcom.NetComServerFactory;
import org.quartz.*;
import org.quartz.Trigger.TriggerState;
......@@ -62,6 +63,7 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
public static XxlJobInfoDao xxlJobInfoDao;
public static XxlJobRegistryDao xxlJobRegistryDao;
public static XxlJobGroupDao xxlJobGroupDao;
public static AdminBiz adminBiz;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
......@@ -69,6 +71,7 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
XxlJobDynamicScheduler.xxlJobInfoDao = applicationContext.getBean(XxlJobInfoDao.class);
XxlJobDynamicScheduler.xxlJobRegistryDao = applicationContext.getBean(XxlJobRegistryDao.class);
XxlJobDynamicScheduler.xxlJobGroupDao = applicationContext.getBean(XxlJobGroupDao.class);
XxlJobDynamicScheduler.adminBiz = applicationContext.getBean(AdminBiz.class);
}
@Override
......
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/AdminBizImpl.java 0 → 100644
浏览文件 @ ee07e0b7
package com.xxl.job.admin.service.impl;
import com.xxl.job.admin.controller.JobApiController;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.dao.XxlJobInfoDao;
import com.xxl.job.admin.dao.XxlJobLogDao;
import com.xxl.job.admin.dao.XxlJobRegistryDao;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import org.apache.commons.lang.StringUtils;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.text.MessageFormat;
import java.util.Date;
import java.util.List;
/**
* @author xuxueli 2017-07-27 21:54:20
*/
@Service
public class AdminBizImpl implements AdminBiz {
private static Logger logger = LoggerFactory.getLogger(AdminBizImpl.class);
@Resource
public XxlJobLogDao xxlJobLogDao;
@Resource
private XxlJobInfoDao xxlJobInfoDao;
@Resource
private XxlJobRegistryDao xxlJobRegistryDao;
@Override
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
for (HandleCallbackParam handleCallbackParam: callbackParamList) {
ReturnT<String> callbackResult = callback(handleCallbackParam);
logger.info("JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
(callbackResult.getCode()==ReturnT.SUCCESS_CODE?"success":"fail"), handleCallbackParam, callbackResult);
}
return ReturnT.SUCCESS;
}
private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
// valid log item
XxlJobLog log = xxlJobLogDao.load(handleCallbackParam.getLogId());
if (log == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "log item not found.");
}
// trigger success, to trigger child job, and avoid repeat trigger child job
String childTriggerMsg = null;
if (ReturnT.SUCCESS_CODE==handleCallbackParam.getExecuteResult().getCode() && ReturnT.SUCCESS_CODE!=log.getHandleCode()) {
XxlJobInfo xxlJobInfo = xxlJobInfoDao.loadById(log.getJobId());
if (xxlJobInfo!=null && StringUtils.isNotBlank(xxlJobInfo.getChildJobKey())) {
childTriggerMsg = "<hr>";
String[] childJobKeys = xxlJobInfo.getChildJobKey().split(",");
for (int i = 0; i < childJobKeys.length; i++) {
String[] jobKeyArr = childJobKeys[i].split("_");
if (jobKeyArr!=null && jobKeyArr.length==2) {
XxlJobInfo childJobInfo = xxlJobInfoDao.loadById(Integer.valueOf(jobKeyArr[1]));
if (childJobInfo!=null) {
try {
boolean ret = XxlJobDynamicScheduler.triggerJob(String.valueOf(childJobInfo.getId()), String.valueOf(childJobInfo.getJobGroup()));
// add msg
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务成功, 子任务Key: {2}, status: {3}, 子任务描述: {4}",
(i+1), childJobKeys.length, childJobKeys[i], ret, childJobInfo.getJobDesc());
} catch (SchedulerException e) {
logger.error("", e);
}
} else {
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务失败, 子任务xxlJobInfo不存在, 子任务Key: {2}",
(i+1), childJobKeys.length, childJobKeys[i]);
}
} else {
childTriggerMsg += MessageFormat.format("<br> {0}/{1} 触发子任务失败, 子任务Key格式错误, 子任务Key: {2}",
(i+1), childJobKeys.length, childJobKeys[i]);
}
}
}
}
// handle msg
StringBuffer handleMsg = new StringBuffer();
if (log.getHandleMsg()!=null) {
handleMsg.append(log.getHandleMsg()).append("<br>");
}
if (handleCallbackParam.getExecuteResult().getMsg() != null) {
handleMsg.append(handleCallbackParam.getExecuteResult().getMsg());
}
if (childTriggerMsg !=null) {
handleMsg.append("<br>子任务触发备注：").append(childTriggerMsg);
}
// success, save log
log.setHandleTime(new Date());
log.setHandleCode(handleCallbackParam.getExecuteResult().getCode());
log.setHandleMsg(handleMsg.toString());
xxlJobLogDao.updateHandleInfo(log);
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
int ret = xxlJobRegistryDao.registryUpdate(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
if (ret < 1) {
xxlJobRegistryDao.registrySave(registryParam.getRegistGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue());
}
return ReturnT.SUCCESS;
}
}
xxl-job-admin/src/test/java/com/xxl/job/dao/impl/AdminApiTest.java deleted 100644 → 0
浏览文件 @ 1651c63d
package com.xxl.job.dao.impl;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.enums.RegistryConfig;
import com.xxl.job.core.util.AdminApiUtil;
/**
* Created by xuxueli on 17/5/10.
*/
public class AdminApiTest {
public static void main(String[] args) {
try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), "aaa", "112312312312");
AdminApiUtil.init("http://localhost:8080/xxl-job-admin");
ReturnT<String> registryResult = AdminApiUtil.callApiFailover(AdminApiUtil.REGISTRY, registryParam);
System.out.println(registryResult);
} catch (Exception e) {
e.printStackTrace();
}
}
}
xxl-job-core/src/main/java/com/xxl/job/core/biz/AdminBiz.java 0 → 100644
浏览文件 @ ee07e0b7
package com.xxl.job.core.biz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import java.util.List;
/**
* @author xuxueli 2017-07-27 21:52:49
*/
public interface AdminBiz {
/**
* callback
*
* @param callbackParamList
* @return
*/
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList);
/**
* registry
*
* @param registryParam
* @return
*/
public ReturnT<String> registry(RegistryParam registryParam);
}
xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java
浏览文件 @ ee07e0b7
......@@ -8,7 +8,6 @@ import com.xxl.job.core.rpc.netcom.NetComServerFactory;
import com.xxl.job.core.thread.ExecutorRegistryThread;
import com.xxl.job.core.thread.JobThread;
import com.xxl.job.core.thread.TriggerCallbackThread;
import com.xxl.job.core.util.AdminApiUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
......@@ -30,7 +29,7 @@ public class XxlJobExecutor implements ApplicationContextAware, ApplicationListe
private String ip;
private int port = 9999;
private String appName;
private String adminAddresses;
public static String adminAddresses;
public static String logPath;
public void setIp(String ip) {
......@@ -52,9 +51,6 @@ public class XxlJobExecutor implements ApplicationContextAware, ApplicationListe
// ---------------------------------- job server ------------------------------------
private NetComServerFactory serverFactory = new NetComServerFactory();
public void start() throws Exception {
// admin api util init
AdminApiUtil.init(adminAddresses);
// executor start
NetComServerFactory.putService(ExecutorBiz.class, new ExecutorBizImpl());
serverFactory.start(port, ip, appName);
......
xxl-job-core/src/main/java/com/xxl/job/core/rpc/netcom/jetty/client/JettyClient.java
浏览文件 @ ee07e0b7
......@@ -19,8 +19,14 @@ public class JettyClient {
// serialize request
byte[] requestBytes = HessianSerializer.serialize(request);
// reqURL
String reqURL = request.getServerAddress();
if (reqURL!=null && reqURL.indexOf("http://")==-1) {
reqURL = "http://" + request.getServerAddress() + "/";
}
// remote invoke
byte[] responseBytes = HttpClientUtil.postRequest("http://" + request.getServerAddress() + "/", requestBytes);
byte[] responseBytes = HttpClientUtil.postRequest(reqURL, requestBytes);
if (responseBytes == null || responseBytes.length==0) {
RpcResponse rpcResponse = new RpcResponse();
rpcResponse.setError("RpcResponse byte[] is null");
......
xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java
浏览文件 @ ee07e0b7
package com.xxl.job.core.thread;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.enums.RegistryConfig;
import com.xxl.job.core.util.AdminApiUtil;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import com.xxl.job.core.util.IpUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -26,8 +28,12 @@ public class ExecutorRegistryThread extends Thread {
public void start(final int port, final String ip, final String appName){
// valid
if ( !(AdminApiUtil.allowCallApi() && (appName!=null && appName.trim().length()>0)) ) {
logger.warn(">>>>>>>>>>>> xxl-job, executor registry config fail");
if (appName==null || appName.trim().length()==0) {
logger.warn(">>>>>>>>>>>> xxl-job, executor registry config fail, appName is null.");
return;
}
if (XxlJobExecutor.adminAddresses==null || XxlJobExecutor.adminAddresses.trim().length()==0) {
logger.warn(">>>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
return;
}
......@@ -45,7 +51,19 @@ public class ExecutorRegistryThread extends Thread {
while (!toStop) {
try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, executorAddress);
ReturnT<String> registryResult = AdminApiUtil.callApiFailover(AdminApiUtil.REGISTRY, registryParam);
ReturnT<String> registryResult = null;
for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) {
String apiUrl = addressUrl.concat("/api");
AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject();
registryResult = adminBiz.registry(registryParam);
if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
registryResult = ReturnT.SUCCESS;
break;
}
}
logger.info(">>>>>>>>>>> xxl-job Executor registry {}, RegistryParam:{}, registryResult:{}",
new Object[]{(registryResult.getCode()==ReturnT.SUCCESS_CODE?"success":"fail"), registryParam.toString(), registryResult.toString()});
} catch (Exception e) {
......
xxl-job-core/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
浏览文件 @ ee07e0b7
package com.xxl.job.core.thread;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.AdminApiUtil;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -35,14 +37,32 @@ public class TriggerCallbackThread {
HandleCallbackParam callback = getInstance().callBackQueue.take();
if (callback != null) {
// callback list
// valid
if (XxlJobExecutor.adminAddresses==null || XxlJobExecutor.adminAddresses.trim().length()==0) {
logger.warn(">>>>>>>>>>>> xxl-job callback fail, adminAddresses is null.");
continue;
}
// callback list param
List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
callbackParamList.add(callback);
// callback, will retry if error
try {
ReturnT<String> callbackResult = AdminApiUtil.callApiFailover(AdminApiUtil.CALLBACK, callbackParamList);
ReturnT<String> callbackResult = null;
for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) {
String apiUrl = addressUrl.concat("/api");
AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject();
callbackResult = adminBiz.callback(callbackParamList);
if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
callbackResult = ReturnT.SUCCESS;
break;
}
}
logger.info(">>>>>>>>>>> xxl-job callback, callbackParamList:{}, callbackResult:{}", new Object[]{callbackParamList, callbackResult});
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job TriggerCallbackThread Exception:", e);
......
xxl-job-core/src/main/java/com/xxl/job/core/util/AdminApiUtil.java deleted 100644 → 0
浏览文件 @ 1651c63d
package com.xxl.job.core.util;
import com.xxl.job.core.biz.model.ReturnT;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* @author xuxueli 2017-05-10 21:28:15
*/
public class AdminApiUtil {
private static Logger logger = LoggerFactory.getLogger(AdminApiUtil.class);
public static final String CALLBACK = "/api/callback";
public static final String REGISTRY = "/api/registry";
private static List<String> adminAddressList = null;
public static void init(String adminAddresses){
// admin assress list
if (adminAddresses != null) {
Set<String> adminAddressSet = new HashSet<String>();
for (String adminAddressItem: adminAddresses.split(",")) {
if (adminAddressItem.trim().length()>0) {
adminAddressSet.add(adminAddressItem);
}
}
adminAddressList = new ArrayList<String>(adminAddressSet);
}
}
public static boolean allowCallApi(){
boolean allowCallApi = (adminAddressList!=null && adminAddressList.size()>0);
return allowCallApi;
}
public static ReturnT<String> callApiFailover(String subUrl, Object requestObj) throws Exception {
if (!allowCallApi()) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "allowCallApi fail.");
}
for (String adminAddress: adminAddressList) {
ReturnT<String> registryResult = null;
try {
String apiUrl = adminAddress.concat(subUrl);
registryResult = callApi(apiUrl, requestObj);
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
if (registryResult!=null && registryResult.getCode()==ReturnT.SUCCESS_CODE) {
return ReturnT.SUCCESS;
}
}
return ReturnT.FAIL;
}
private static ReturnT<String> callApi(String finalUrl, Object requestObj) throws Exception {
HttpPost httpPost = new HttpPost(finalUrl);
CloseableHttpClient httpClient = HttpClients.createDefault();
try {
// timeout
RequestConfig requestConfig = RequestConfig.custom()
.setConnectionRequestTimeout(10000)
.setSocketTimeout(10000)
.setConnectTimeout(10000)
.build();
httpPost.setConfig(requestConfig);
// data
if (requestObj != null) {
String json = JacksonUtil.writeValueAsString(requestObj);
StringEntity entity = new StringEntity(json, "utf-8");
entity.setContentEncoding("UTF-8");
entity.setContentType("application/json");
httpPost.setEntity(entity);
}
// do post
HttpResponse response = httpClient.execute(httpPost);
HttpEntity entity = response.getEntity();
if (null != entity) {
String responseMsg = EntityUtils.toString(entity, "UTF-8");
if (response.getStatusLine().getStatusCode() != 200) {
EntityUtils.consume(entity);
return new ReturnT<String>(response.getStatusLine().getStatusCode(),
"StatusCode（+"+ response.getStatusLine().getStatusCode() +"） Error，response：" + responseMsg);
}
EntityUtils.consume(entity);
if (responseMsg!=null && responseMsg.startsWith("{")) {
ReturnT<String> result = JacksonUtil.readValue(responseMsg, ReturnT.class);
return result;
}
}
return ReturnT.FAIL;
} catch (Exception e) {
logger.error("", e);
return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
} finally {
if (httpPost!=null) {
httpPost.releaseConnection();
}
try {
httpClient.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
xxl-job-core/src/main/java/com/xxl/job/core/util/DBUtil.java deleted 100644 → 0
浏览文件 @ 1651c63d
package com.xxl.job.core.util;
import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
/**
* Created by xuxueli on 16/9/30.
*/
public class DBUtil {
private static Connection getConn(DataSource dataSource) {
try {
return dataSource.getConnection();
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
/**
* update
*
* @param dataSource
* @param sql
* @param params
*/
public static int update(DataSource dataSource, String sql, Object params[]) {
Connection connection = getConn(dataSource);
PreparedStatement preparedStatement = null;
int ret = 0;
try {
preparedStatement = connection.prepareStatement(sql);
if (params != null) {
for (int i = 0; i < params.length; i++) {
preparedStatement.setObject(i + 1, params[i]);
}
}
ret = preparedStatement.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
} finally {
release(connection, preparedStatement, null);
}
return ret;
}
/**
* query
*
* @param dataSource
* @param sql
* @param params
* @return
*/
public static List<Map<String, Object>> query(DataSource dataSource, String sql, Object[] params) {
Connection connection = getConn(dataSource);
PreparedStatement preparedStatement = null;
ResultSet resultSet = null;
try {
preparedStatement = connection.prepareStatement(sql);
if (params != null) {
for (int i = 0; i < params.length; i++) {
preparedStatement.setObject(i + 1, params[i]);
}
}
resultSet = preparedStatement.executeQuery();
List<Map<String, Object>> ret = resultSetToList(resultSet);
return ret;
} catch (SQLException e) {
e.printStackTrace();
} finally {
release(connection, preparedStatement, resultSet);
}
return null;
}
private static List<Map<String, Object>> resultSetToList(ResultSet resultSet) throws SQLException {
if (resultSet == null) {
return new ArrayList<Map<String, Object>>();
}
ResultSetMetaData resultSetMetaData = resultSet.getMetaData(); // 得到结果集(rs)的结构信息，比如字段数、字段名等
int columnCount = resultSetMetaData.getColumnCount(); // 返回此 ResultSet 对象中的列数
List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
while (resultSet.next()) {
Map<String, Object> rowData = new HashMap<String, Object>(columnCount);
for (int i = 1; i <= columnCount; i++) {
rowData.put(resultSetMetaData.getColumnName(i), resultSet.getObject(i));
}
list.add(rowData);
}
return list;
}
/**
* release
* @param connection
* @param preparedStatement
* @param resultSet
*/
public static void release(Connection connection, PreparedStatement preparedStatement, ResultSet resultSet) {
if (resultSet != null) {
try {
resultSet.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
if (connection != null) {
try {
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论