通知定时任务类：
public class NotifyMCTimer implements Callable<String> {private Map<String,String> resultMap;private String trade_no;private String notifyUrl;private volatile int runTimes = 0; public NotifyMCTimer(Map<String, String> resultMap, String notifyUrl, String trade_no) {super();this.resultMap = resultMap;this.notifyUrl = notifyUrl;this.trade_no = trade_no;} @Overridepublic String call() throws Exception {return HttpUtils.sendPostMessage(resultMap, "utf-8", notifyUrl);}public Map<String, String> getResultMap() {return resultMap;}public void setResultMap(Map<String, String> resultMap) {this.resultMap = resultMap;}public String getTrade_no() {return trade_no;}public void setTrade_no(String trade_no) {this.trade_no = trade_no;} public int getRunTimes() {return runTimes;} public void setRunTimes(int runTimes) {this.runTimes = runTimes;}}
发送定时任务工具类：
public class ScheduledExecutorServiceUtil extends BaseService {private volatile static ScheduledExecutorService scheduledExecutor ; private static MyBlockingQueue<Callable<String>> queue ; private ScheduledExecutorServiceUtil (){} //初始化 public static void startUp() { if (scheduledExecutor == null) { synchronized (ScheduledExecutorServiceUtil.class) { if (scheduledExecutor == null) { scheduledExecutor = Executors.newScheduledThreadPool(10); queue = new MyBlockingQueue<>(100); new Consumer(queue).start(); } } } } //提交无延迟任务 public static void submitNoDelay(Callable<String> task){ NotifyMCTimer timer = (NotifyMCTimer)task; try { ScheduledFuture<String> future = scheduledExecutor.schedule(task, 0, TimeUnit.MILLISECONDS); timer.setRunTimes(timer.getRunTimes()+1);if(future.get().contains("\"result\":0")||future.get().contains("已完款")){System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知成功");return;}else{System.out.println("++++++++++++++++future:"+future.get());logger.info("++++++++++++++++future:"+future.get());System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知失败");submitWithDelay(task,10l);}} catch (Exception e) {System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知出现错误");submitWithDelay(task,10l);return;} } //提交延时任务，成功接到返回值退出方法，失败继续提交任务 public static void submitWithDelay(Callable<String> task,Long delay) { NotifyMCTimer timer = (NotifyMCTimer)task; if(timer.getRunTimes()>=5){ System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"通知达到5次最大次数"); return; }else{ try { ScheduledFuture<String> future = scheduledExecutor.schedule(task, delay, TimeUnit.MINUTES); timer.setRunTimes(timer.getRunTimes()+1); if(future.get().contains("\"result\":0")||future.get().contains("已完款")){ System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知成功"); return; }else { System.out.println("++++++++++++++++future:"+future.get()); logger.info("++++++++++++++++future:"+future.get()); System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知失败"); if(timer.getRunTimes()<5){ submitWithDelay(task,delay); } } } catch (Exception e) { System.out.println(DateUtils.dateToString(DateUtils.PATTEN_HMS)+"trade_no:"+timer.getTrade_no()+"第"+timer.getRunTimes()+"次通知出现错误"); submitWithDelay(task,delay); return; } }} //工具类入口方法，将定时任务放入队列public static void addToQueue(Callable<String> task) throws InterruptedException{startUp();queue.put(task);} //保存任务的队列 static class MyBlockingQueue<E> {private Queue<E> queue = null;private int limit;private Lock lock = new ReentrantLock();private Condition notFull = lock.newCondition(); private Condition notEmpty = lock.newCondition(); public MyBlockingQueue(int limit) {this.limit = limit;queue = new ArrayDeque<>(limit);} public void put(E e) throws InterruptedException {lock.lockInterruptibly();try{while (queue.size() == limit) {notFull.await();}queue.add(e);notEmpty.signal();}finally{lock.unlock();}} public E take() throws InterruptedException {lock.lockInterruptibly();try{while (queue.isEmpty()) {notEmpty.await();}E e = queue.poll();notFull.signal();return e;}finally{lock.unlock();}}} //消费者static class Consumer extends Thread {MyBlockingQueue<Callable<String>> queue; public Consumer(MyBlockingQueue<Callable<String>> queue2) {this.queue = queue2;} @Overridepublic void run() {try {while (true) {Callable<String> task = queue.take();submitNoDelay(task);}} catch (InterruptedException e) {}}}}
调用时直接放到任务队列：
ScheduledExecutorServiceUtil.addToQueue(new NotifyMCTimer(resultMap,paymentRecd.getNotifyUrl(),paymentRecd.getTradeNo()));
方法可能有缺陷，仅供参考
版权声明：本文转载于今日头条，版权归作者所有，如果侵权，请联系本站编辑删除