Redisson move

yshop-api/src/main/java/co/yixiang/modules/task/CancelOrderService.java

@@ -0,0 +1,41 @@
+package co.yixiang.modules.task;
+
+import cn.hutool.core.util.ObjectUtil;
+import co.yixiang.modules.order.entity.YxStoreOrder;
+import co.yixiang.modules.order.service.YxStoreOrderService;
+import co.yixiang.utils.BeanUtil;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+
+@Component
+@Slf4j
+public class CancelOrderService implements ExecuteJob {
+    @Override
+    public void execute(Map<String,Object> job) {
+        if(job.get("delayJobName").equals("CANCEL_ORVERTIME_ORDER")){
+            log.info("系统开始处理延时任务---订单超时未付款---" + job.get("orderId"));
+
+            YxStoreOrderService yxorderService = BeanUtil.getBean(YxStoreOrderService.class);
+
+            YxStoreOrder order = null;
+            try {
+                order = yxorderService.getOne(new QueryWrapper<YxStoreOrder>().eq("id", job.get("orderId")).eq("is_del",0));
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            if(ObjectUtil.isNull(order)) {
+                return;
+            }
+            if(order.getPaid() != 0){
+                return;
+            }
+            yxorderService.cancelOrderByTask((int)job.get("orderId"));
+            log.info("系统结束处理延时任务---订单超时未付款取消---" + job.get("orderId"));
+        }
+
+    }
+}

yshop-api/src/main/java/co/yixiang/modules/task/DelayJobService.java

@@ -0,0 +1,33 @@
+package co.yixiang.modules.task;
+
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RQueue;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by kl on 2018/7/20.
+ * Content :订单延时job服务
+ */
+@Component
+public class DelayJobService {
+
+    @Autowired
+    private RedissonClient client;
+
+    /**
+     * 添加超时任务到redis队列
+     * @param job  任务
+     * @param delay 超时时间
+     */
+    public void submitJob(Map<String,Object> job, Long delay){
+        RQueue<Map<String,Object>> blockingQueue = client.getQueue(JobTimer.CUSTOMER_JOB_TIMER_JOBS);
+        RDelayedQueue<Map<String,Object>> delayedQueue = client.getDelayedQueue(blockingQueue);
+        delayedQueue.offer(job,delay,TimeUnit.MILLISECONDS);
+        delayedQueue.destroy();
+    }
+}

yshop-api/src/main/java/co/yixiang/modules/task/ExecuteJob.java

@@ -0,0 +1,12 @@
+package co.yixiang.modules.task;
+
+import java.util.Map;
+
+/**
+ * Created by kl on 2018/7/20.
+ * Content :延时job执行器接口
+ */
+public interface ExecuteJob {
+
+    void execute(Map<String,Object> job);
+}

yshop-api/src/main/java/co/yixiang/modules/task/JobTimer.java

@@ -0,0 +1,64 @@
+package co.yixiang.modules.task;
+
+import org.redisson.api.RBlockingQueue;
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 消费已经到点的延时job服务，通过job参数调用业务执行器实现
+ */
+@Component
+public class JobTimer {
+
+    static final String CUSTOMER_JOB_TIMER_JOBS = "customer_job_timer_jobs";
+    @Autowired
+    private RedissonClient client;
+
+    @Autowired
+    private ExecuteJob service;
+
+    ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
+    @PostConstruct
+    public void startJobTimer() {
+        RBlockingQueue<Map<String,Object>> blockingQueue = client.getBlockingQueue(CUSTOMER_JOB_TIMER_JOBS);
+        RDelayedQueue<Map<String,Object>> delayedQueue = client.getDelayedQueue(blockingQueue);
+        new Thread(() -> {
+            while (true) {
+                try {
+                    Map<String,Object> job = blockingQueue.take();
+                    executorService.execute(new ExecutorTask(job));
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    try {
+                        TimeUnit.SECONDS.sleep(60);
+                    } catch (Exception ex) {
+
+                    }
+                }finally {
+                    delayedQueue.destroy();
+                }
+            }
+        }).start();
+    }
+    class ExecutorTask implements Runnable {
+
+
+        private Map<String,Object> delayJob;
+
+        public ExecutorTask(Map<String,Object> delayJob) {
+            this.delayJob = delayJob;
+        }
+        @Override
+        public void run() {
+            service.execute(delayJob);
+        }
+    }
+}