DispatchTask.java 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. package cn.sagacloud.server;
  2. import cn.sagacloud.guice.InjectorInit;
  3. import cn.sagacloud.mybatis.model.TaskModel;
  4. import cn.sagacloud.mybatis.service.TaskService;
  5. import cn.sagacloud.pojo.ChannelHandlerContextWrapper;
  6. import cn.sagacloud.pojo.Command;
  7. import cn.sagacloud.pojo.TaskStatus;
  8. import cn.sagacloud.utils.CommonUtil;
  9. import org.apache.log4j.Logger;
  10. import java.util.ArrayList;
  11. import java.util.Arrays;
  12. import java.util.Map;
  13. import java.util.Set;
  14. public class DispatchTask implements Runnable {
  15. private static Logger log = Logger.getLogger(DispatchTask.class);
  16. private static TaskService service = InjectorInit.getInjector().getInstance(TaskService.class);
  17. ArrayList<ChannelHandlerContextWrapper> clientList;
  18. private int clientIndex = 0;
  19. public static Map<Integer, TaskModel> tasks;
  20. public DispatchTask(ArrayList<ChannelHandlerContextWrapper> clientList) throws Exception {
  21. this.clientList = clientList;
  22. tasks = service.getAllTaskMapByStatus(Arrays.asList(0, 1, 2));
  23. }
  24. /**
  25. * 是否当前状态下可以执行该命令
  26. * @param taskId task id
  27. * @param cmd command
  28. * @return 写一个状态, 如果不可执行该命令, 返回null
  29. */
  30. public static TaskStatus isAllowCmd(int taskId, Command cmd) {
  31. TaskModel task = DispatchTask.tasks.get(taskId);
  32. if(task == null)
  33. return null;
  34. TaskStatus status = TaskStatus.getTaskStatusById(task.getTask_status());
  35. if(status == null)
  36. return null;
  37. Map<Command, TaskStatus> allowCmd = SchedulerHandler.statusTransitionMap.get(status);
  38. if(allowCmd == null)
  39. return null;
  40. return allowCmd.get(cmd);
  41. }
  42. // 加锁
  43. public static void changeStatusByCmd(int taskId, Command cmd, String clientInfo) {
  44. TaskStatus nextStatus = isAllowCmd(taskId, cmd);
  45. if(nextStatus == null)
  46. return;
  47. TaskModel task = DispatchTask.tasks.get(taskId);
  48. if(task == null)
  49. return;
  50. task.setTask_last_client(clientInfo);
  51. task.setTask_status(TaskStatus.getIdByTaskStatus(nextStatus));
  52. updateTask(task);
  53. }
  54. public static void changeStatusByCmdWithReturnJson(int taskId, Command cmd, String clientInfo, String returnJson) {
  55. TaskStatus nextStatus = isAllowCmd(taskId, cmd);
  56. if(nextStatus == null)
  57. return;
  58. TaskModel task = DispatchTask.tasks.get(taskId);
  59. if(task == null)
  60. return;
  61. task.setTask_last_client(clientInfo);
  62. task.setTask_result_json(returnJson);
  63. task.setTask_status(TaskStatus.getIdByTaskStatus(nextStatus));
  64. updateTask(task);
  65. }
  66. @Override
  67. public void run() {
  68. // 考虑加锁
  69. try {
  70. for (long cnt = Long.MAX_VALUE; cnt > Long.MIN_VALUE; --cnt) {
  71. // 获取到一个在waiting状态的任务
  72. TaskModel toBeSent = getOneAvailableTask();
  73. if (toBeSent != null) {
  74. // 获取到一个潜在可以接受任务的客户端
  75. ChannelHandlerContextWrapper client = getOneAvailableClient();
  76. if (client != null) {
  77. // 将获取到的任务发送给选中客户端
  78. client.sendTask(toBeSent);
  79. }
  80. }
  81. // 检测sending, sent状态的任务, 是否有超时需要重新下载的任务
  82. checkSendingTimeOut();
  83. checkExecuteTimeOut();
  84. try {
  85. Thread.sleep(1000);
  86. } catch (InterruptedException e) {
  87. e.printStackTrace();
  88. }
  89. if (cnt % 500 == 0L) {
  90. syncTask();
  91. }
  92. }
  93. }catch (Exception ex){
  94. log.error(ex.getMessage());
  95. System.out.println("出错");
  96. }
  97. }
  98. private void syncTask() {
  99. try {
  100. Map<Integer, TaskModel> tmpTasks = service.getAllTaskMapByStatus(Arrays.asList(0, 1, 2));
  101. tasks = tmpTasks;
  102. log.info("同步数据库任务完成");
  103. } catch (Exception e) {
  104. e.printStackTrace();
  105. log.error("同步数据库失败");
  106. }
  107. }
  108. private void checkExecuteTimeOut() {
  109. int id = getOneExecuteTimeOutTaskId();
  110. if(id > 0){
  111. // 是否加锁
  112. tasks.get(id).setTask_status(0);
  113. System.out.println("任务" + id + "执行超时, 已重置");
  114. }
  115. }
  116. /**
  117. * 获取一个在执行任务时超时的任务id
  118. * @return
  119. */
  120. private int getOneExecuteTimeOutTaskId() {
  121. if(tasks == null || tasks.size() == 0)
  122. return -1;
  123. long currentTime = CommonUtil.getTime();
  124. for(Integer id : tasks.keySet()){
  125. TaskModel task = tasks.get(id);
  126. long expectTime = task.getTask_expected_finish_time();
  127. long sendTime = task.getTask_sent_time();
  128. if(sendTime <= 0)
  129. continue;
  130. if(currentTime - sendTime > expectTime){
  131. return id;
  132. }
  133. }
  134. return -1;
  135. }
  136. private void checkSendingTimeOut() {
  137. for(ChannelHandlerContextWrapper client : clientList){
  138. int id = client.getOneSendingTimeOutTaskId();
  139. if(id > 0){
  140. // 是否加锁
  141. tasks.get(id).setTask_status(0);
  142. System.out.println("任务" + id + "发送超时, 已重置");
  143. }
  144. }
  145. }
  146. private ChannelHandlerContextWrapper getOneAvailableClient() {
  147. int clientCount ;
  148. if(clientList == null || clientList.size() == 0)
  149. return null;
  150. else {
  151. clientCount = clientList.size();
  152. if(clientCount == 0)
  153. return null;
  154. clientIndex %= clientCount;
  155. }
  156. ChannelHandlerContextWrapper wrapper = null;
  157. do{
  158. try {
  159. wrapper = clientList.get(clientIndex);
  160. }catch (Exception ex){
  161. wrapper = null;
  162. clientCount = clientList.size();
  163. }
  164. if(!wrapper.isLastRefuseTimeOutPassed()){
  165. wrapper = null;
  166. }
  167. if(wrapper.getClientInfo() == null || wrapper.getClientInfo().length() == 0)
  168. wrapper = null;
  169. if(clientCount == 0)
  170. return null;
  171. clientIndex = (clientIndex + 1) % clientCount;
  172. }while(wrapper == null && clientIndex != 0);
  173. if(wrapper != null)
  174. System.out.println("当前选中index :" + clientIndex);
  175. return wrapper;
  176. }
  177. private TaskModel getOneAvailableTask() {
  178. //检测到在waiting状态的任务, 返回
  179. for(Integer taskId : tasks.keySet()){
  180. TaskModel task = tasks.get(taskId);
  181. if(task.getTask_status() == 0){
  182. return task;
  183. }
  184. }
  185. return null;
  186. }
  187. private static void updateTask(TaskModel task){
  188. try {
  189. service.updateTask(task);
  190. } catch (Exception e) {
  191. e.printStackTrace();
  192. log.error("任务:" + task.getId() + "update失败");
  193. }
  194. }
  195. }