添加异步调度

2026-10-29 16:13:47 +00:00 · 2017-07-29 10:18:36 +08:00
parent b99df28104
commit f203f040ba
11 changed files with 469 additions and 74 deletions
@@ -20,5 +20,10 @@
      <groupId>com.egridcloud.udf</groupId>
      <artifactId>udf-starter-core</artifactId>
    </dependency>
+    <!-- udf-starter-rms -->
+    <dependency>
+      <groupId>com.egridcloud.udf</groupId>
+      <artifactId>udf-starter-rms</artifactId>
+    </dependency>
  </dependencies>
 </project>
@@ -0,0 +1,57 @@
+/**
+ * RmsJobStats.java
+ * Created at 2016-09-27
+ * Created by wangkang
+ * Copyright (C) 2016 egridcloud.com, All rights reserved.
+ */
+package com.egridcloud.udf.scheduler.client;
+
+/**
+ * 描述 : RmsJobStats
+ *
+ * @author wangkang
+ *
+ */
+public enum RmsJobStats {
+  /**
+   * 描述 : 正在执行
+   */
+  EXECUTING(1),
+  /**
+   * 描述 : 完成
+   */
+  COMPLETE(2),
+  /**
+   * 描述 : 错误
+   */
+  ERROR(-1);
+
+  /**
+   * <p>
+   * Field value: 参数值
+   * </p>
+   */
+  private Integer value = null;
+
+  /**
+   * <p>
+   * Description: 构造函数
+   * </p>
+   * 
+   * @param value 构造函数
+   */
+  private RmsJobStats(Integer value) {
+    this.value = value;
+  }
+
+  /**
+   * <p>
+   * Description: 放回int值
+   * </p>
+   * 
+   * @return value 放回int值
+   */
+  public Integer value() {
+    return this.value;
+  }
+}
@@ -0,0 +1,132 @@
+/**
+ * SchClientService.java
+ * Created at 2017-06-12
+ * Created by Administrator
+ * Copyright (C) 2016 egridcloud.com, All rights reserved.
+ */
+package com.egridcloud.udf.scheduler.client;
+
+import java.util.Date;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.retry.annotation.Recover;
+import org.springframework.retry.annotation.Retryable;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import com.egridcloud.udf.core.RestResponse;
+import com.egridcloud.udf.rms.Rms;
+import com.egridcloud.udf.scheduler.client.domain.RmsJobParam;
+import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
+
+/**
+ * 描述 : SchClientService
+ *
+ * @author Administrator
+ *
+ */
+@Component
+public class SchClientHandle implements ApplicationContextAware {
+
+  /**
+   * 描述 : 日志
+   */
+  private static final Logger LOGGER = LoggerFactory.getLogger(SchClientHandle.class);
+
+  /**
+   * 描述 : spring上下文
+   */
+  private ApplicationContext applicationContext;
+
+  /**
+   * 描述 : rms
+   */
+  @Autowired
+  private Rms rms;
+
+  /**
+   * 描述 : 异步执行
+   *
+   * @param receiveTime 接收时间
+   * @param param 参数
+   */
+  @Async
+  public void asyncHandle(Date receiveTime, RmsJobParam param) {
+    //定义返回值
+    RmsJobResult result = new RmsJobResult();
+    result.setParam(param);
+    result.setClientReceiveTime(receiveTime);
+    result.setFireInstanceId(param.getFireInstanceId());
+    //执行(并且记录开始和结束时间)
+    result.setClientStartExecuteTime(new Date());
+    try {
+      this.handle(param);
+      result.setClientEndExecuteTime(new Date());
+      result.setStats(RmsJobStats.COMPLETE.value());
+      //回调
+      this.callback(result);
+    } catch (Exception e) {
+      result.setClientEndExecuteTime(new Date());
+      result.setStats(RmsJobStats.ERROR.value());
+      result.setErrorMsg(ExceptionUtils.getStackTrace(e));
+      //回调
+      this.callback(result);
+      //抛出异常
+      LOGGER.error("asyncHandle error:", e);
+      throw new SchException(e);
+    }
+
+  }
+
+  /**
+   * 描述 : 同步执行
+   *
+   * @param param 参数
+   */
+  public void handle(RmsJobParam param) {
+    //判断bean是否存在
+    if (!applicationContext.containsBean(param.getBeanName())) {
+      throw new SchException(param.getBeanName() + " not definition");
+    }
+    //获得bean
+    IExecutor bean = applicationContext.getBean(param.getBeanName(), IExecutor.class);
+    //执行
+    bean.handle(param.getJobDataMap());
+  }
+
+  /**
+   * 描述 : 回调(重处理)
+   *
+   * @param result result
+   */
+  @Retryable(maxAttempts = 3, value = Exception.class)
+  private void callback(RmsJobResult result) {
+    LOGGER.info("try to callback");
+    final String serviceCode = "SCH_CLIENT_CALLBACK_1";
+    rms.call(serviceCode, result, null, new ParameterizedTypeReference<RestResponse<String>>() {
+    }, null);
+  }
+
+  /**
+   * 描述 : 重试失败后执行的方法
+   *
+   * @param e e
+   */
+  @Recover
+  public void recover(Exception e) {
+    LOGGER.error("try to callback failed:", e);
+  }
+
+  @Override
+  public void setApplicationContext(ApplicationContext arg0) throws BeansException {
+    applicationContext = arg0;
+  }
+
+}
@@ -38,12 +38,36 @@ public class RmsJobParam implements Serializable {
  @ApiModelProperty(value = "bean名称", required = true, dataType = "string")
  private String beanName;

+  /**
+   * 描述 : 是否异步
+   */
+  @ApiModelProperty(value = "是否异步", required = true, dataType = "boolean")
+  private Boolean async;
+
  /**
   * 描述 : 作业参数
   */
  @ApiModelProperty(value = "作业参数", required = true, dataType = "object")
  private Map<String, Object> jobDataMap; //NOSONAR

+  /**
+   * 描述 : 获取async
+   *
+   * @return the async
+   */
+  public Boolean getAsync() {
+    return async;
+  }
+
+  /**
+   * 描述 : 设置async
+   *
+   * @param async the async to set
+   */
+  public void setAsync(Boolean async) {
+    this.async = async;
+  }
+
  /**
   * 描述 : 获取jobDataMap
   *
@@ -39,10 +39,10 @@ public class RmsJobResult implements Serializable {
  private RmsJobParam param;

  /**
-   * 描述 : 客户端初始化时间
+   * 描述 : 客户端接收时间
   */
-  @ApiModelProperty(value = "客户端初始化时间", required = true, dataType = "long")
-  private Date clientInitTime;
+  @ApiModelProperty(value = "客户端接收时间", required = true, dataType = "long")
+  private Date clientReceiveTime;

  /**
   * 描述 : 客户端开始时间
@@ -57,21 +57,87 @@ public class RmsJobResult implements Serializable {
  private Date clientEndExecuteTime;

  /**
-   * 描述 : 获取clientInitTime
-   *
-   * @return the clientInitTime
+   * 描述 : 状态
   */
-  public Date getClientInitTime() {
-    return clientInitTime;
+  @ApiModelProperty(value = "状态", required = true, dataType = "int")
+  private Integer stats;
+
+  /**
+   * 描述 : 错误信息
+   */
+  @ApiModelProperty(value = "错误信息", required = true, dataType = "string")
+  private String errorMsg;
+
+  /**
+   * 描述 : 获取param
+   *
+   * @return the param
+   */
+  public RmsJobParam getParam() {
+    return param;
  }

  /**
-   * 描述 : 设置clientInitTime
+   * 描述 : 设置param
   *
-   * @param clientInitTime the clientInitTime to set
+   * @param param the param to set
   */
-  public void setClientInitTime(Date clientInitTime) {
-    this.clientInitTime = clientInitTime;
+  public void setParam(RmsJobParam param) {
+    this.param = param;
+  }
+
+  /**
+   * 描述 : 获取errorMsg
+   *
+   * @return the errorMsg
+   */
+  public String getErrorMsg() {
+    return errorMsg;
+  }
+
+  /**
+   * 描述 : 设置errorMsg
+   *
+   * @param errorMsg the errorMsg to set
+   */
+  public void setErrorMsg(String errorMsg) {
+    this.errorMsg = errorMsg;
+  }
+
+  /**
+   * 描述 : 获取stats
+   *
+   * @return the stats
+   */
+  public Integer getStats() {
+    return stats;
+  }
+
+  /**
+   * 描述 : 设置stats
+   *
+   * @param stats the stats to set
+   */
+  public void setStats(Integer stats) {
+    this.stats = stats;
+  }
+
+  /**
+   * 描述 : 获取clientReceiveTime
+   *
+   * @return the clientReceiveTime
+   */
+  public Date getClientReceiveTime() {
+    return clientReceiveTime;
+  }
+
+  /**
+   * 描述 : 设置clientReceiveTime
+   *
+   * @param clientReceiveTime the clientReceiveTime to set
+   */
+  public void setClientReceiveTime(Date clientReceiveTime) {
+    this.clientReceiveTime = clientReceiveTime;
  }

  /**
@@ -110,24 +176,6 @@ public class RmsJobResult implements Serializable {
    this.clientEndExecuteTime = clientEndExecuteTime;
  }

-  /**
-   * 描述 : 获取param
-   *
-   * @return the param
-   */
-  public RmsJobParam getParam() {
-    return param;
-  }
-
-  /**
-   * 描述 : 设置param
-   *
-   * @param param the param to set
-   */
-  public void setParam(RmsJobParam param) {
-    this.param = param;
-  }
-
  /**
   * 描述 : 获取fireInstanceId
   *
@@ -8,15 +8,13 @@ package com.egridcloud.udf.scheduler.client.web;

 import java.util.Date;

-import org.springframework.beans.BeansException;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ApplicationContextAware;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;

 import com.egridcloud.udf.core.RestResponse;
-import com.egridcloud.udf.scheduler.client.IExecutor;
-import com.egridcloud.udf.scheduler.client.SchException;
+import com.egridcloud.udf.scheduler.client.RmsJobStats;
+import com.egridcloud.udf.scheduler.client.SchClientHandle;
 import com.egridcloud.udf.scheduler.client.domain.RmsJobParam;
 import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;

@@ -27,44 +25,35 @@ import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
 *
 */
@RestController
-public class SchClientController implements ISchClientController, ApplicationContextAware {
+public class SchClientController implements ISchClientController {

  /**
-   * 描述 : spring上下文
+   * 描述 : schClientHandle
   */
-  private ApplicationContext applicationContext;
+  @Autowired
+  private SchClientHandle schClientHandle;

  @Override
  public RestResponse<RmsJobResult> execute(@RequestBody RmsJobParam param) {
-    //客户端初始化时间
-    Date clientInitTime = new Date();
+    //记录来接收时间
+    Date receiveTime = new Date();
    //定义返回值
    RmsJobResult result = new RmsJobResult();
-    result.setFireInstanceId(param.getFireInstanceId());
    result.setParam(param);
-    //判断bean是否存在
-    if (!applicationContext.containsBean(param.getBeanName())) {
-      throw new SchException(param.getBeanName() + " not definition");
+    result.setClientReceiveTime(receiveTime);
+    result.setFireInstanceId(param.getFireInstanceId());
+    //执行(区分同步跟异步)
+    if (param.getAsync()) {
+      schClientHandle.asyncHandle(receiveTime, param);
+      result.setStats(RmsJobStats.EXECUTING.value());
+    } else {
+      result.setClientStartExecuteTime(new Date());
+      schClientHandle.handle(param);
+      result.setClientEndExecuteTime(new Date());
+      result.setStats(RmsJobStats.COMPLETE.value());
    }
-    //获得bean
-    IExecutor bean = applicationContext.getBean(param.getBeanName(), IExecutor.class);
-    //客户端开始时间
-    Date clientStartExecuteTime = new Date();
-    //执行
-    bean.handle(param.getJobDataMap());
-    //客户端结束时间
-    Date clientEndExecuteTime = new Date();
-    //设置返回值
-    result.setClientInitTime(clientInitTime);
-    result.setClientStartExecuteTime(clientStartExecuteTime);
-    result.setClientEndExecuteTime(clientEndExecuteTime);
    //返回
    return new RestResponse<>(result);
  }

-  @Override
-  public void setApplicationContext(ApplicationContext arg0) throws BeansException {
-    applicationContext = arg0;
-  }
-
 }
@@ -19,8 +19,15 @@ public interface IRmsJobLog {
  /**
   * 描述 : save
   *
-   * @param generalJobResult generalJobResult
+   * @param result result
   */
-  public void save(RmsJobResult generalJobResult);
+  public void save(RmsJobResult result);
+
+  /**
+   * 描述 : delete
+   *
+   * @param fireInstanceId fireInstanceId
+   */
+  public void delete(String fireInstanceId);

 }
@@ -8,10 +8,12 @@ package com.egridcloud.udf.scheduler.job;

 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.quartz.DisallowConcurrentExecution;
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.quartz.JobDataMap;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -19,6 +21,7 @@ import org.springframework.http.ResponseEntity;
 import com.egridcloud.udf.core.RestResponse;
 import com.egridcloud.udf.rms.Rms;
 import com.egridcloud.udf.scheduler.IRmsJobLog;
+import com.egridcloud.udf.scheduler.client.RmsJobStats;
 import com.egridcloud.udf.scheduler.client.SchException;
 import com.egridcloud.udf.scheduler.client.domain.RmsJobParam;
 import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
@@ -29,9 +32,13 @@ import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
 * @author Administrator
 *
 */
-@DisallowConcurrentExecution
 public class RmsJob extends AbstractBaseJob {

+  /**
+   * 描述 : 日志
+   */
+  private static final Logger LOGGER = LoggerFactory.getLogger(RmsJob.class);
+
  @Override
  protected void executeInternal(JobExecutionContext jobExecutionContext)
      throws JobExecutionException {
@@ -40,6 +47,7 @@ public class RmsJob extends AbstractBaseJob {
    //获得必要字段
    String serviceCode = jobDataMap.getString("serviceCode");
    String beanName = jobDataMap.getString("beanName");
+    Boolean async = jobDataMap.getBoolean("async");
    //校验
    if (StringUtils.isBlank(serviceCode)) {
      throw new SchException("serviceCode can not be empty");
@@ -53,21 +61,37 @@ public class RmsJob extends AbstractBaseJob {
    RmsJobParam rmsJobParam = new RmsJobParam();
    rmsJobParam.setFireInstanceId(jobExecutionContext.getFireInstanceId());
    rmsJobParam.setBeanName(beanName);
+    rmsJobParam.setAsync(async);
    rmsJobParam.setJobDataMap(jobDataMap.getWrappedMap());
    //请求
-    ResponseEntity<RestResponse<RmsJobResult>> result = rms.call(serviceCode, rmsJobParam, null,
-        new ParameterizedTypeReference<RestResponse<RmsJobResult>>() {
-        }, null);
-    //判断http状态
-    if (result.getStatusCode() != HttpStatus.OK) {
-      throw new SchException(result.getStatusCode().toString());
+    try {
+      ResponseEntity<RestResponse<RmsJobResult>> result = rms.call(serviceCode, rmsJobParam, null,
+          new ParameterizedTypeReference<RestResponse<RmsJobResult>>() {
+          }, null);
+      //判断http状态
+      if (result.getStatusCode() != HttpStatus.OK) {
+        //抛出异常
+        throw new SchException("http状态:" + result.getStatusCode().toString());
+      }
+      //记录
+      saveRmsJobResult(result.getBody().getResult());
+    } catch (Exception e) {
+      //定义返回值
+      RmsJobResult result = new RmsJobResult();
+      result.setParam(rmsJobParam);
+      result.setFireInstanceId(rmsJobParam.getFireInstanceId());
+      result.setStats(RmsJobStats.ERROR.value());
+      result.setErrorMsg(ExceptionUtils.getStackTrace(e));
+      //记录
+      saveRmsJobResult(result);
+      //抛出异常
+      LOGGER.error("RmsJob error:", e);
+      throw new JobExecutionException(e);
    }
-    //记录结果
-    saveRmsJobResult(result.getBody().getResult());
  }

  /**
-   * 描述 : 保存(会吃掉注入IGeneralJobResult实现类的异常)
+   * 描述 : 保存
   *
   * @param result result
   */
@@ -0,0 +1,20 @@
+/**
+ * RmsJobDisallowConcurrent.java
+ * Created at 2017-06-04
+ * Created by Administrator
+ * Copyright (C) 2016 egridcloud.com, All rights reserved.
+ */
+package com.egridcloud.udf.scheduler.job;
+
+import org.quartz.DisallowConcurrentExecution;
+
+/**
+ * 描述 : RmsJobDisallowConcurrent
+ *
+ * @author Administrator
+ *
+ */
+@DisallowConcurrentExecution
+public class RmsJobDisallowConcurrent extends RmsJob {
+
+}
@@ -0,0 +1,49 @@
+/**
+ * ISchClientCallbackController.java
+ * Created at 2017-06-11
+ * Created by Administrator
+ * Copyright (C) 2016 egridcloud.com, All rights reserved.
+ */
+package com.egridcloud.udf.scheduler.web;
+
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+
+import com.egridcloud.udf.core.RestResponse;
+import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+
+/**
+ * 描述 : ISchClientCallbackController
+ *
+ * @author Administrator
+ *
+ */
+@Api(value = "调度客户端回调服务", consumes = "application/json", produces = "application/json",
+    protocols = "http")
+@RequestMapping(value = "/service/scheduler/client/callback")
+public interface ISchClientCallbackController {
+
+  /**
+   * 描述 : 客户端回调
+   *
+   * @param result 执行结果
+   * @return 结果
+   */
+  @ApiOperation(value = "SCH_CLIENT_CALLBACK_1", notes = "客户端回调")
+  @ApiImplicitParams({
+      @ApiImplicitParam(paramType = "header", name = "rmsApplicationName", value = "rms应用名称",
+          required = true, dataType = "string"),
+      @ApiImplicitParam(paramType = "header", name = "rmsSign", value = "rms认证秘钥", required = true,
+          dataType = "string"),
+      @ApiImplicitParam(paramType = "header", name = "rmsServiceCode", value = "rms接口编号",
+          required = true, dataType = "string") })
+  @RequestMapping(method = RequestMethod.POST)
+  public RestResponse<String> callback(@RequestBody RmsJobResult result);
+
+}
@@ -0,0 +1,40 @@
+/**
+ * JobController.java
+ * Created at 2017-05-26
+ * Created by Administrator
+ * Copyright (C) 2016 egridcloud.com, All rights reserved.
+ */
+package com.egridcloud.udf.scheduler.web;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.egridcloud.udf.core.RestResponse;
+import com.egridcloud.udf.scheduler.IRmsJobLog;
+import com.egridcloud.udf.scheduler.client.domain.RmsJobResult;
+
+/**
+ * 描述 : JobController
+ *
+ * @author Administrator
+ *
+ */
+@RestController
+public class SchClientCallbackController implements ISchClientCallbackController {
+
+  /**
+   * 描述 : rmsJobLog
+   */
+  @Autowired(required = false)
+  private IRmsJobLog rmsJobLog;
+
+  @Override
+  public RestResponse<String> callback(@RequestBody RmsJobResult result) {
+    if (rmsJobLog != null) {
+      rmsJobLog.save(result);
+    }
+    return new RestResponse<>();
+  }
+
+}