并发

src/main/java/com/izouma/nineth/service/OrderService.java

@@ -27,6 +27,7 @@ import com.izouma.nineth.dto.PageQuery;
 import com.izouma.nineth.enums.*;
 import com.izouma.nineth.event.CreateAssetEvent;
 import com.izouma.nineth.event.CreateOrderEvent;
+import com.izouma.nineth.event.OrderNotifyEvent;
 import com.izouma.nineth.event.TransferAssetEvent;
 import com.izouma.nineth.exception.BusinessException;
 import com.izouma.nineth.repo.*;
@@ -46,6 +47,7 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.context.event.EventListener;
 import org.springframework.core.env.Environment;
 import org.springframework.data.domain.Page;
+import org.springframework.data.redis.core.BoundSetOperations;
 import org.springframework.data.redis.core.BoundValueOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -471,16 +473,16 @@ public class OrderService {
 
     public void notifyOrder(Long orderId, PayMethod payMethod, String transactionId) {
         log.info("订单回调 orderId: {}, payMethod: {}, transactionId: {}", orderId, payMethod, transactionId);
+        BoundSetOperations<String, Object> listOps = redisTemplate.boundSetOps("orderNotify::" + orderId);
+        listOps.add(transactionId);
+        listOps.expire(7, TimeUnit.DAYS);
+
         BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps("orderLock::" + orderId);
         Boolean flag = ops.setIfAbsent(1, 1, TimeUnit.DAYS);
         if (!Boolean.TRUE.equals(flag)) {
-            log.info("订单回调失败 orderId: {} redis锁定", orderId);
-            errorOrderRepo.save(ErrorOrder.builder()
-                    .orderId(orderId)
-                    .transactionId(transactionId)
-                    .payMethod(payMethod)
-                    .errorMessage("redis锁定")
-                    .build());
+            log.info("订单回调失败 orderId: {} redis锁定, 重新发送到队列", orderId);
+            rocketMQTemplate.syncSend(generalProperties.getOrderNotifyTopic(),
+                    new OrderNotifyEvent(orderId, payMethod, transactionId, System.currentTimeMillis()));
             return;
         }
 
@@ -565,6 +567,7 @@ public class OrderService {
         redisTemplate.delete("orderLock::" + orderId);
     }
 
+
     @EventListener
     public void onCreateAsset(CreateAssetEvent event) {
         Asset asset = event.getAsset();
@@ -607,6 +610,23 @@ public class OrderService {
             log.error("订单取消失败 {}, redis锁了", order.getId());
             return;
         }
+
+        Set<Object> transactionIds = redisTemplate.opsForSet().members("orderNotify::" + order.getId());
+        if (transactionIds != null && transactionIds.size() > 0) {
+            if (transactionIds.parallelStream().anyMatch(transactionId -> {
+                try {
+                    Map<String, Object> map = Payment.query(transactionId.toString());
+                    return "succeeded".equalsIgnoreCase(MapUtils.getString(map, "status"));
+                } catch (BaseAdaPayException e) {
+                    e.printStackTrace();
+                }
+                return false;
+            })) {
+                log.info("订单已经支付成功，不能取消 {}", order.getId());
+                return;
+            }
+        }
+
         try {
             if (order.getStatus() != OrderStatus.NOT_PAID) {
                 throw new BusinessException("已支付订单无法取消");

src/main/java/com/izouma/nineth/web/OrderNotifyController.java

@@ -15,6 +15,8 @@ import com.izouma.nineth.config.AlipayProperties;
 import com.izouma.nineth.config.GeneralProperties;
 import com.izouma.nineth.enums.PayMethod;
 import com.izouma.nineth.event.OrderNotifyEvent;
+import com.izouma.nineth.repo.ErrorOrder;
+import com.izouma.nineth.repo.ErrorOrderRepo;
 import com.izouma.nineth.service.AssetService;
 import com.izouma.nineth.service.GiftOrderService;
 import com.izouma.nineth.service.MintOrderService;
@@ -23,6 +25,8 @@ import com.izouma.nineth.utils.SnowflakeIdWorker;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.MapUtils;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.*;
@@ -49,6 +53,7 @@ public class OrderNotifyController {
     private final RocketMQTemplate  rocketMQTemplate;
     private final GeneralProperties generalProperties;
     private final MintOrderService  mintOrderService;
+    private final ErrorOrderRepo    errorOrderRepo;
 
     @PostMapping("/order/alipay")
     @ResponseBody
@@ -186,9 +191,24 @@ public class OrderNotifyController {
                 JSONObject jsonObject = JSON.parseObject(data);
                 String channel = jsonObject.getString("pay_channel");
                 String id = jsonObject.getString("id");
+                PayMethod payMethod = channel.startsWith("wx") ? PayMethod.WEIXIN : PayMethod.ALIPAY;
 
-                rocketMQTemplate.syncSend(generalProperties.getOrderNotifyTopic(),
-                        new OrderNotifyEvent(orderId, channel.startsWith("wx") ? PayMethod.WEIXIN : PayMethod.ALIPAY, id, System.currentTimeMillis()));
+                SendResult result = null;
+                int count = 0;
+                do {
+                    if (++count > 10) {
+                        log.error("发送订单通知失败，订单id：{}", orderId);
+                        errorOrderRepo.save(ErrorOrder.builder()
+                                .orderId(orderId)
+                                .transactionId(id)
+                                .payMethod(payMethod)
+                                .errorMessage("发送到回调队列失败")
+                                .build());
+                        break;
+                    }
+                    result = rocketMQTemplate.syncSend(generalProperties.getOrderNotifyTopic(),
+                            new OrderNotifyEvent(orderId, payMethod, id, System.currentTimeMillis()));
+                } while (result != null && result.getSendStatus() == SendStatus.SEND_OK);
             }
         }
     }

src/test/java/com/izouma/nineth/service/AdapayTest.java

@@ -214,4 +214,10 @@ public class AdapayTest {
         Map<String, Object> refund = Refund.query(queryParams);
         System.out.println(JSON.toJSONString(refund, SerializerFeature.PrettyFormat));
     }
+
+    @Test
+    public void singleQuery() throws BaseAdaPayException {
+        Map<String, Object> map = Payment.query("002112022021815230610340647575776317440");
+        System.out.println(JSON.toJSONString(map, SerializerFeature.PrettyFormat));
+    }
 }