异步分佣任务

mp-admin/src/main/java/com/qs/mp/web/controller/api/admin/ChannelOrderMgrController.java

@@ -200,8 +200,7 @@ public class ChannelOrderMgrController extends BaseApiController {
 		if(null != shipParam && null != shipParam.getPkgIds()
 				&& shipParam.getPkgIds().size() > 0) {
 			ChannelOrder channelOrder = new ChannelOrder();
-
-
+			channelOrder.setType(order.getType());
 
 			if (DeliverTypeEnum.NO_DELIVER.getValue().equals(deliveryType)) {
 				// 无需物流，设置发货时间为当前时间

mp-admin/src/main/java/com/qs/mp/web/controller/api/channel/mall/ChannelOrderController.java

@@ -381,13 +381,8 @@ public class ChannelOrderController extends BaseApiController {
     if (!channelId.equals(channelOrder.getChannelId())) {
       return AjaxResult.error("非法操作他人订单");
     }
-
-    // 更新用户提货订单状态为已完成,并且设置收货时间
-    ChannelOrder param = new ChannelOrder();
-    param.setOrderId(order.getOrderId());
-    param.setStatus(ChannelOrderStatusEnum.FINISHED);
-    param.setConfirmTime(new Date());
-    channelOrderService.updateById(param);
+    // 确认收货
+    channelOrderService.confirm(order);
 
     return AjaxResult.success();
   }

mp-common/src/main/java/com/qs/mp/common/enums/AsyncTaskTypeEnum.java

@@ -11,7 +11,7 @@ import io.swagger.annotations.ApiModel;
  */
 @ApiModel("异步任务类型枚举")
 public enum AsyncTaskTypeEnum implements IEnum<Integer> {
-    PRE_COMMISSION(1,"预付票分佣");
+    CHANNEL_CONFIRM_RECEIPT(1,"经销商确认收货任务");
 
     private Integer value;
     private String desc;

mp-quartz/src/main/java/com/qs/mp/quartz/task/AsyncConsumeTask.java

@@ -0,0 +1,84 @@
+package com.qs.mp.quartz.task;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Assert;
+import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
+import com.qs.mp.common.core.redis.DistributedLocker;
+import com.qs.mp.common.enums.AsyncTaskTypeEnum;
+import com.qs.mp.common.utils.LogUtil;
+import com.qs.mp.framework.domain.AsyncTask;
+import com.qs.mp.framework.redis.RedisKey;
+import com.qs.mp.framework.redis.RedisLockKey;
+import com.qs.mp.framework.service.IAsyncTaskService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * 异步消息消费任务
+ * @author Cup
+ * @date 2022/6/7
+ */
+@Component("asyncConsumeTask")
+public class AsyncConsumeTask {
+
+    protected final Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName());
+
+    @Autowired
+    private IAsyncTaskService asyncTaskService;
+
+    @Autowired
+    private DistributedLocker distributedLocker;
+
+    public void processingTask(){
+
+        List<AsyncTask> asyncTaskList = asyncTaskService.list(new LambdaQueryWrapper<AsyncTask>().eq(AsyncTask::getStatus, 0));
+        if (CollectionUtils.isEmpty(asyncTaskList)) {
+            return;
+        }
+
+        for (AsyncTask asyncTask : asyncTaskList) {
+            String lockKey = RedisLockKey.build(RedisLockKey.ASYNC_TASK_KEY, asyncTask.getId());
+            // 加锁,自动续期
+            if (!distributedLocker.tryLock(lockKey,0,-1, TimeUnit.SECONDS)) {
+                continue;
+            }
+            try {
+                this.execute(asyncTask);
+            } catch (Exception e) {
+                // 更新异步任务执行次数
+                asyncTaskService.update(new LambdaUpdateWrapper<AsyncTask>()
+                        .set(AsyncTask::getOperateNum, asyncTask.getOperateNum() + 1)
+                        .eq(AsyncTask::getId, asyncTask.getId()));
+                LogUtil.error(logger, e, "异步任务处理异常。任务id:{0}", asyncTask.getId());
+            } finally {
+                // 释放锁
+                distributedLocker.unlock(lockKey);
+            }
+        }
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    public void execute(AsyncTask asyncTask){
+        if (AsyncTaskTypeEnum.CHANNEL_CONFIRM_RECEIPT.getValue().equals(asyncTask.getType())) {
+            // 确认收货后相关任务处理
+            asyncTaskService.channelConfirmReceipt(asyncTask);
+        }
+
+        // 更新异步任务状态
+        boolean rtn = asyncTaskService.update(new LambdaUpdateWrapper<AsyncTask>()
+                .set(AsyncTask::getOperateNum, asyncTask.getOperateNum() + 1)
+                .set(AsyncTask::getStatus, 1)
+                .eq(AsyncTask::getId, asyncTask.getId()));
+
+        Assert.isFalse(rtn,"异步任务状态更新失败, id:{0}", asyncTask.getId());
+    }
+
+}

mp-service/src/main/java/com/qs/mp/channel/service/IChannelOrderService.java

@@ -81,4 +81,10 @@ public interface IChannelOrderService extends IService<ChannelOrder> {
 	 * @return
 	 */
     boolean cancelOrderByAdmin(ChannelOrderCancelParam param);
+
+	/**
+	 * 订单确认收货
+	 * @param order
+	 */
+    void confirm(ChannelOrder order);
 }

mp-service/src/main/java/com/qs/mp/channel/service/impl/ChannelOrderServiceImpl.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.Wrapper;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Assert;
 import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
 import com.baomidou.mybatisplus.core.toolkit.StringUtils;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -14,6 +15,8 @@ import com.qs.mp.admin.service.ITicketBoxService;
 import com.qs.mp.admin.service.ITicketPackageService;
 import com.qs.mp.common.enums.*;
 import com.qs.mp.common.utils.DateUtils;
+import com.qs.mp.framework.domain.AsyncTask;
+import com.qs.mp.framework.service.IAsyncTaskService;
 import com.qs.mp.pay.service.IPayOrderService;
 import com.qs.mp.pay.service.IWalletService;
 import com.qs.mp.user.domain.UserAddr;
@@ -86,6 +89,9 @@ public class ChannelOrderServiceImpl extends
   @Autowired
   private IPayOrderService payOrderService;
 
+  @Autowired
+  private IAsyncTaskService asyncTaskService;
+
   @Override
   @Transactional
   public String submitOrder(Long channelId, String memo, ChannelOrderSettleVO orderSettleVO) {
@@ -236,6 +242,24 @@ public class ChannelOrderServiceImpl extends
         return true;
     }
 
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public void confirm(ChannelOrder order) {
+        // 更新用户提货订单状态为已完成,并且设置收货时间
+        ChannelOrder param = new ChannelOrder();
+        param.setOrderId(order.getOrderId());
+        param.setStatus(ChannelOrderStatusEnum.FINISHED);
+        param.setConfirmTime(new Date());
+        boolean rtn = channelOrderService.updateById(param);
+        Assert.isFalse(rtn,"更新提货订单状态失败,orderId:{0}", order.getOrderId());
+
+        // 提交异步任务
+        AsyncTask asyncTask = new AsyncTask();
+        asyncTask.setType(AsyncTaskTypeEnum.CHANNEL_CONFIRM_RECEIPT.getValue());
+        asyncTask.setBizId(order.getOrderId());
+        Assert.isFalse(asyncTaskService.save(asyncTask),"提交经销商提货订单确认收货任务失败,orderId:{0}", order.getOrderId());
+    }
+
     /**
      * 回滚盲票组销量和状态
      * @param orderItem

mp-service/src/main/java/com/qs/mp/framework/redis/RedisLockKey.java

@@ -9,7 +9,7 @@ import com.qs.mp.common.utils.MessageHelper;
  *
  */
 public enum RedisLockKey {
-
+    ASYNC_TASK_KEY("async_task_key_{0}", "异步任务key"),
     MARKETING_REAL_NUM_LOCK("marketing_real_num_lock_{0}", "活动真实人数增加锁"),
     MARKETING_LOTTERY_KEY("marketing_lottery_key_{0}","免费抽奖活动开奖"),
     USER_TICKET_CASH_LOCK("user_ticket_cash_lock_{0}", "盲票兑奖锁");

mp-service/src/main/java/com/qs/mp/framework/service/IAsyncTaskService.java

@@ -13,4 +13,9 @@ import com.baomidou.mybatisplus.extension.service.IService;
  */
 public interface IAsyncTaskService extends IService<AsyncTask> {
 
+    /**
+     * 经销商确认收货后相关任务处理
+     * @param asyncTask
+     */
+    void channelConfirmReceipt(AsyncTask asyncTask);
 }

mp-service/src/main/java/com/qs/mp/framework/service/impl/AsyncTaskServiceImpl.java

@@ -1,10 +1,22 @@
 package com.qs.mp.framework.service.impl;
 
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.qs.mp.channel.domain.ChannelOrder;
+import com.qs.mp.channel.service.IChannelOrderService;
+import com.qs.mp.channel.service.IChannelUserRelService;
+import com.qs.mp.common.enums.ChannelCommissionResourceEnum;
+import com.qs.mp.common.enums.ChannelOrderTypeEnum;
+import com.qs.mp.common.exception.ServiceException;
 import com.qs.mp.framework.domain.AsyncTask;
 import com.qs.mp.framework.mapper.AsyncTaskMapper;
 import com.qs.mp.framework.service.IAsyncTaskService;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.qs.mp.user.service.IUserTicketOrderService;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Objects;
 
 /**
  * <p>
@@ -17,4 +29,29 @@ import org.springframework.stereotype.Service;
 @Service
 public class AsyncTaskServiceImpl extends ServiceImpl<AsyncTaskMapper, AsyncTask> implements IAsyncTaskService {
 
+    @Autowired
+    private IChannelOrderService channelOrderService;
+
+    @Autowired
+    private IUserTicketOrderService userTicketService;;
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public void channelConfirmReceipt(AsyncTask asyncTask) {
+        // 获取供应商订单编号
+        String orderId = asyncTask.getBizId();
+
+        // 查询订单
+        ChannelOrder channelOrder = channelOrderService.getById(orderId);
+        if (Objects.isNull(channelOrder)) {
+            throw new ServiceException("经销商订单不存在,OrderId:" + orderId);
+        }
+
+        if (ChannelOrderTypeEnum.RETAIL.equals(channelOrder.getType())) {
+            return;
+        }
+
+        // 分佣
+        userTicketService.commToChannel(orderId, ChannelCommissionResourceEnum.CHANNEL.getValue());
+    }
 }