HandlerThread.java 2.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
  1. package com.xxl.job.client.handler;
  2. import java.io.PrintWriter;
  3. import java.io.StringWriter;
  4. import java.util.HashMap;
  5. import java.util.Map;
  6. import java.util.concurrent.LinkedBlockingQueue;
  7. import java.util.concurrent.TimeUnit;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import com.xxl.job.client.handler.IJobHandler.JobHandleStatus;
  11. import com.xxl.job.client.util.HttpUtil;
  12. import com.xxl.job.client.util.HttpUtil.RemoteCallBack;
  13. /**
  14. * handler thread
  15. * @author xuxueli 2016-1-16 19:52:47
  16. */
  17. public class HandlerThread extends Thread{
  18. private static Logger logger = LoggerFactory.getLogger(HandlerThread.class);
  19. private IJobHandler handler;
  20. private LinkedBlockingQueue<Map<String, String>> handlerDataQueue;
  21. public HandlerThread(IJobHandler handler) {
  22. this.handler = handler;
  23. handlerDataQueue = new LinkedBlockingQueue<Map<String,String>>();
  24. }
  25. public void pushData(Map<String, String> param) {
  26. handlerDataQueue.offer(param);
  27. }
  28. int i = 1;
  29. @Override
  30. public void run() {
  31. while(true){
  32. try {
  33. i++;
  34. Map<String, String> handlerData = handlerDataQueue.poll();
  35. if (handlerData!=null) {
  36. String trigger_log_url = handlerData.get(HandlerRepository.TRIGGER_LOG_URL);
  37. String trigger_log_id = handlerData.get(HandlerRepository.TRIGGER_LOG_ID);
  38. String handler_params = handlerData.get(HandlerRepository.HANDLER_PARAMS);
  39. // parse param
  40. String[] handlerParams = null;
  41. if (handler_params!=null && handler_params.trim().length()>0) {
  42. handlerParams = handler_params.split(",");
  43. } else {
  44. handlerParams = new String[0];
  45. }
  46. // handle job
  47. JobHandleStatus _status = JobHandleStatus.FAIL;
  48. String _msg = null;
  49. try {
  50. _status = handler.handle(handlerParams);
  51. } catch (Exception e) {
  52. logger.info("HandlerThread Exception:", e);
  53. StringWriter out = new StringWriter();
  54. e.printStackTrace(new PrintWriter(out));
  55. _msg = out.toString();
  56. }
  57. // callback handler info
  58. RemoteCallBack callback = null;
  59. try {
  60. HashMap<String, String> params = new HashMap<String, String>();
  61. params.put(HandlerRepository.TRIGGER_LOG_ID, trigger_log_id);
  62. params.put("status", _status.name());
  63. params.put("msg", _msg);
  64. callback = HttpUtil.post(trigger_log_url, params);
  65. } catch (Exception e) {
  66. logger.info("HandlerThread Exception:", e);
  67. }
  68. logger.info("<<<<<<<<<<< xxl-job thread handle, handlerData:{}, callback_status:{}, callback_msg:{}, callback:{}, thread:{}",
  69. new Object[]{handlerData, _status, _msg, callback, this});
  70. } else {
  71. try {
  72. TimeUnit.MILLISECONDS.sleep(i * 100);
  73. } catch (InterruptedException e) {
  74. e.printStackTrace();
  75. }
  76. if (i>5) {
  77. i= 0;
  78. }
  79. }
  80. } catch (Exception e) {
  81. logger.info("HandlerThread Exception:", e);
  82. }
  83. }
  84. }
  85. }