feat callback retry

dengwei
1 parent 5103e910
Showing 8 changed files with 177 additions and 26 deletions
cashier-mall/src/main/java/com/diligrp/cashier/mall/MallConfiguration.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/MallConstants.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/api/ManualHandlerApi.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/listener/CallBackRetryListener.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/service/biz/impl/MallBizRefundServiceImpl.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/service/sourcechannel/AbstractSourceChannel.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/service/sourcechannel/RtMallChannel.java
cashier-mall/src/main/java/com/diligrp/cashier/mall/type/RtMarkErrorCode.java
@@ -4,11 +4,18 @@ import com.diligrp.cashier.mall.client.RtMallHttpClient;
 import com.diligrp.cashier.mall.property.RtMallDynamicProperty;
 import com.diligrp.cashier.shared.mybatis.MybatisMapperSupport;
 import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.CustomExchange;
+import org.springframework.amqp.core.Queue;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import java.util.HashMap;
+import java.util.Map;
+
 @Configuration
 @ComponentScan("com.diligrp.cashier.mall")
 @MapperScan(basePackages = {"com.diligrp.cashier.mall.dao"}, markerInterface = MybatisMapperSupport.class)
@@ -24,4 +31,24 @@ public class MallConfiguration {
         RtMallDynamicProperty.AppSecretDynamicProperty config = rtMallDynamicProperty.getAppSecrets().getFirst();
         return new RtMallHttpClient(apiUrl, config.getAppKey(), config.getAppSecret());
     }
+
+    @Bean
+    public Queue queue() {
+        return new Queue(MallConstants.CASHIER_MALL_CALLBACK_RETRY_QUEUE, true, false, false);
+    }
+
+    @Bean
+    public CustomExchange customExchange() {
+        Map<String, Object> args = new HashMap<>();
+        args.put("x-delayed-type", "direct");
+        return new CustomExchange(MallConstants.CASHIER_MALL_CALLBACK_RETRY_EXCHANGE, MallConstants.EXCHANGE_TYPE, true, false, args);
+    }
+
+    @Bean
+    public Binding binding(Queue queue, CustomExchange customExchange) {
+        return BindingBuilder.bind(queue)
+                .to(customExchange)
+                .with(MallConstants.CASHIER_MALL_CALLBACK_RETRY_QUEUE)
+                .noargs();
+    }
 }
@@ -11,11 +11,17 @@ public interface MallConstants {
     String RESULT_SUCCESS = "success";
     String RESULT_FAILURE = "fail";
-    Integer RT_MALL_SOURCE = 1;
-
     String MALL_TOKEN = Constants.PRODUCT_NAME + "mall_token:";
     String MALL_USER_INFO = Constants.PRODUCT_NAME + "mall_user_info:";
     String HOURLY_PAYMENT_CALLBACK_METHOD = "hourly.payment.callback";
     String HOURLY_REFUND_CALLBACK_METHOD = "hourly.refund.callback";
+
+    // 三方回调重试队列
+    String CASHIER_MALL_CALLBACK_RETRY_QUEUE = "cashier_mall_callback_retry_queue";
+    String CASHIER_MALL_CALLBACK_RETRY_EXCHANGE = "cashier_mall_callback_retry_exchange";
+    String EXCHANGE_TYPE = "x-delayed-message";
+
+    // 延时1分钟
+    Integer WAIT_DELAY_MILLIS = 60_000;
 }
 package com.diligrp.cashier.mall.api;
+import com.diligrp.cashier.mall.context.MallInitializeContext;
+import com.diligrp.cashier.mall.dao.MallBizRefundDao;
+import com.diligrp.cashier.mall.model.MallBizPayment;
+import com.diligrp.cashier.mall.model.MallBizRefund;
+import com.diligrp.cashier.mall.service.biz.MallBizPaymentService;
 import com.diligrp.cashier.shared.domain.Message;
 import com.diligrp.cashier.shared.spi.IPaymentEventListener;
 import com.diligrp.cashier.shared.spi.domain.PaymentResultBO;
 import com.diligrp.cashier.shared.spi.domain.RefundResultBO;
 import jakarta.annotation.Resource;
 import org.springframework.validation.annotation.Validated;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
 /**
  * @ClassName ManualHandlerApi.java
@@ -23,6 +25,11 @@ import org.springframework.web.bind.annotation.RestController;
 public class ManualHandlerApi {
     @Resource
     private IPaymentEventListener payNotifyServiceImpl;
+    @Resource
+    private MallBizPaymentService mallBizPaymentService;
+    @Resource
+    private MallBizRefundDao mallBizRefundDao;
+
     @PostMapping("/payment/event")
     public Message<?> handlePaymentEvent(@RequestBody PaymentResultBO payment) {
@@ -36,4 +43,19 @@ public class ManualHandlerApi {
         return Message.success();
     }
+    @PostMapping("/payment/callback/{source}")
+    public Message<?> paymentCallback(@RequestBody PaymentResultBO payment, @PathVariable("source") Integer source) {
+        String tradeId = payment.getTradeId();
+        MallBizPayment mallBizPayment = mallBizPaymentService.getByPayTradeId(tradeId);
+        MallInitializeContext.getBySource(source).payCallBack(payment, mallBizPayment);
+        return Message.success();
+    }
+
+    @PostMapping("/refund/callback/{source}")
+    public Message<?> refundCallback(@RequestBody RefundResultBO refund, @PathVariable("source") Integer source) {
+        MallBizRefund mallBizRefund = mallBizRefundDao.getByRefundTradeId(refund.getRefundId());
+        MallInitializeContext.getBySource(source).refundCallBack(refund, mallBizRefund);
+        return Message.success();
+    }
+
 }
+package com.diligrp.cashier.mall.listener;
+
+import com.diligrp.cashier.mall.MallConstants;
+import com.diligrp.cashier.mall.context.MallInitializeContext;
+import com.diligrp.cashier.mall.model.MallBizPayment;
+import com.diligrp.cashier.mall.model.MallBizRefund;
+import com.diligrp.cashier.shared.spi.domain.PaymentResultBO;
+import com.diligrp.cashier.shared.spi.domain.RefundResultBO;
+import com.diligrp.cashier.shared.util.JsonUtils;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.rabbitmq.client.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * @ClassName CallBackRetryListener.java
+ * @author dengwei
+ * @version 1.0.0
+ * @Description CallBackRetryListener
+ */
+@Component
+public class CallBackRetryListener {
+    private static final Logger LOG = LoggerFactory.getLogger(CallBackRetryListener.class);
+
+    @RabbitListener(queues = MallConstants.CASHIER_MALL_CALLBACK_RETRY_QUEUE, ackMode = "MANUAL")
+    public void callBackRetry(Message message, Channel channel) throws IOException {
+        try {
+            LOG.info("callBackRetry message【{}】", new String(message.getBody()));
+            Map<String, Object> data = JsonUtils.fromJsonString(new String(message.getBody()), new TypeReference<>() {
+            });
+
+            switch ((String) data.get("callBackMethod")) {
+                case "pay" -> {
+                    Integer source = (Integer) data.get("source");
+                    PaymentResultBO event = JsonUtils.convertValue(data.get("event"), PaymentResultBO.class);
+                    MallBizPayment mallBizPayment = JsonUtils.convertValue(data.get("mallBizPayment"), MallBizPayment.class);
+                    MallInitializeContext.getBySource(source).payCallBack(event, mallBizPayment);
+                }
+                case "refund" -> {
+                    Integer source = (Integer) data.get("source");
+                    RefundResultBO event = JsonUtils.convertValue(data.get("event"), RefundResultBO.class);
+                    MallBizRefund mallBizRefund = JsonUtils.convertValue(data.get("mallBizRefund"), MallBizRefund.class);
+                    MallInitializeContext.getBySource(source).refundCallBack(event, mallBizRefund);
+                }
+                default -> LOG.warn("未知的回调方法【{}】", data.get("callBackMethod"));
+            }
+            // 消息处理成功，发送确认
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        } catch (Exception e) {
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        }
+    }
+}
@@ -65,7 +65,7 @@ public class MallBizRefundServiceImpl implements MallBizRefundService {
     public RefundSuccessVO refund(RefundCO refundCo) {
         MallBizRefund mallBizRefund = mallBizRefundDao.getMallBizRefund(new MallBizRefund(refundCo.getOrderId(), refundCo.getTradeId()));
         if (Objects.nonNull(mallBizRefund)) {
-            throw new RtMartMallException(RtMarkErrorCode.E50056);
+            throw new RtMartMallException(RtMarkErrorCode.E5006);
         }
         MallBizOrder mallBizOrder = mallBizOrderDao.getByOrderId(refundCo.getOrderId());
@@ -20,9 +20,13 @@ import com.diligrp.cashier.shared.util.JsonUtils;
 import jakarta.annotation.Resource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.transaction.annotation.Transactional;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -49,6 +53,8 @@ public abstract class AbstractSourceChannel {
     protected MallBizPaymentService mallBizPaymentService;
     @Resource
     protected MallBizRefundService mallBizRefundService;
+    @Resource
+    protected RabbitTemplate rabbitTemplate;
     /**
      * authLogin
@@ -155,4 +161,15 @@ public abstract class AbstractSourceChannel {
      * refundCallBack
      */
     public abstract void refundCallBack(RefundResultBO event, MallBizRefund mallBizRefund);
+
+    /**
+     * sendDelayMessage
+     */
+    protected void sendDelayMessage(String load) {
+        LOG.info("sendDelayMessage load: {}", load);
+        Message msg = MessageBuilder.withBody(load.getBytes(StandardCharsets.UTF_8))
+                .setHeader("x-delay", MallConstants.WAIT_DELAY_MILLIS)
+                .build();
+        rabbitTemplate.convertAndSend(MallConstants.CASHIER_MALL_CALLBACK_RETRY_EXCHANGE, MallConstants.CASHIER_MALL_CALLBACK_RETRY_QUEUE, msg);
+    }
 }
@@ -21,6 +21,7 @@ import com.diligrp.cashier.shared.util.DateUtils;
 import com.diligrp.cashier.shared.util.JsonUtils;
 import com.diligrp.cashier.shared.util.UrlParamParserUtils;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.collect.Maps;
 import org.apache.commons.collections4.CollectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -84,17 +85,30 @@ public class RtMallChannel extends AbstractSourceChannel {
         if (Objects.equals(mallBizOrder.getOrderType(), OrderType.MINI_PROGRAM.code)) {
             List<MallBizPaymentOrder> mallBizPaymentOrders = mallBizPaymentService.listPaymentOrderByBizPaymentId(mallBizPayment.getId());
             if (CollectionUtils.isNotEmpty(mallBizPaymentOrders)) {
-                payCallBack(mallBizOrder, mallBizPayment, mallBizPaymentOrders.getFirst());
+                payCallBack(event, mallBizOrder, mallBizPayment, mallBizPaymentOrders.getFirst());
             }
         }
     }
     /**
+     * refundCallBack
+     */
+    @Override
+    public void refundCallBack(RefundResultBO event, MallBizRefund mallBizRefund) {
+        MallBizOrder mallBizOrder = mallBizOrderDao.selectByPrimaryKey(mallBizRefund.getBizOrderId());
+        // 目前大润发只对小程序支付存在回调逻辑
+        if (Objects.equals(mallBizOrder.getOrderType(), OrderType.MINI_PROGRAM.code)) {
+            refundCallBack(event, mallBizOrder, mallBizRefund);
+        }
+    }
+
+    /**
      * payCallBack
      * @see https://shopex.yuque.com/hl0rrx/vlp0m4/nkwt5yrhdfzoy78s?singleDoc#RtKpF
      * 支付回调接口
      */
-    public void payCallBack(final MallBizOrder mallBizOrder,
+    public void payCallBack(final PaymentResultBO event,
+                            final MallBizOrder mallBizOrder,
                             final MallBizPayment mallBizPayment,
                             final MallBizPaymentOrder mallBizPaymentOrder) {
         LOG.info("orderType：{} payCallBack orderId: {} tradeId: {}", mallBizOrder.getOrderType(), mallBizPaymentOrder.getOrderId(), mallBizPaymentOrder.getTradeId());
@@ -106,37 +120,42 @@ public class RtMallChannel extends AbstractSourceChannel {
             message = HttpClientUtils.postJson(mallBizPayment.getPaymentCallback(), hourlyPaymentCallback, null, new TypeReference<>() {
             }, "rt-mall");
         } catch (Exception e) {
-            // 失败-前期未考虑重试 人工补偿重试！！
             LOG.warn("payCallBack failed: {}", JsonUtils.toJsonString(hourlyPaymentCallback), e);
+            Map<String, Object> param = Maps.newHashMapWithExpectedSize(4);
+            param.put("event", event);
+            param.put("mallBizPayment", mallBizPayment);
+            param.put("callBackMethod", "pay");
+            param.put("source", OrderSource.RT_MART.getCode());
+            sendDelayMessage(JsonUtils.toJsonString(param));
         }
         LOG.info("payCallBack message: {}", JsonUtils.toJsonString(message));
     }
     /**
      * refundCallBack
-     */
-    @Override
-    public void refundCallBack(RefundResultBO event, MallBizRefund mallBizRefund) {
-        MallBizOrder mallBizOrder = mallBizOrderDao.selectByPrimaryKey(mallBizRefund.getBizOrderId());
-        // 目前大润发只对小程序支付存在回调逻辑
-        if (Objects.equals(mallBizOrder.getOrderType(), OrderType.MINI_PROGRAM.code)) {
-            refundCallBack(mallBizOrder, mallBizRefund);
-        }
-    }
-
-    /**
-     * refundCallBack
      * @see https://shopex.yuque.com/hl0rrx/vlp0m4/nkwt5yrhdfzoy78s?singleDoc#RtKpF
      * 退款回调接口
      */
-    public void refundCallBack(final MallBizOrder mallBizOrder,
+    public void refundCallBack(final RefundResultBO event,
+                               final MallBizOrder mallBizOrder,
                                final MallBizRefund mallBizRefund) {
         LOG.info("orderType：{} refundCallBack orderId: {} tradeId: {}", mallBizOrder.getOrderType(), mallBizRefund.getOrderId(), mallBizRefund.getTradeId());
         RtMallDynamicProperty.AppSecretDynamicProperty property = getProperty(mallBizOrder);
         HourlyRefundCallbackDTO hourlyRefundCallback = HourlyRefundCallbackDTO.of(mallBizOrder, mallBizRefund, property);
         hourlyRefundCallback.setSign(sign(hourlyRefundCallback, property));
-        RtMarkMessage<?> message = HttpClientUtils.postJson(mallBizRefund.getRefundCallback(), hourlyRefundCallback, null, new TypeReference<>() {
-        }, "rt-mall");
+        RtMarkMessage<?> message = null;
+        try {
+            message = HttpClientUtils.postJson(mallBizRefund.getRefundCallback(), hourlyRefundCallback, null, new TypeReference<>() {
+            }, "rt-mall");
+        } catch (Exception e) {
+            LOG.warn("refundCallBack failed: {}", JsonUtils.toJsonString(hourlyRefundCallback), e);
+            Map<String, Object> param = Maps.newHashMapWithExpectedSize(4);
+            param.put("event", event);
+            param.put("mallBizRefund", mallBizRefund);
+            param.put("callBackMethod", "refund");
+            param.put("source", OrderSource.RT_MART.getCode());
+            sendDelayMessage(JsonUtils.toJsonString(param));
+        }
         LOG.info("refundCallBack message: {}", JsonUtils.toJsonString(message));
     }
@@ -20,7 +20,8 @@ public enum RtMarkErrorCode {
     E5003("E5003", "回调状态不匹配"),
     E5004("E5004", "数据异常"),
     E5005("E5005", "支付状态异常"),
-    E50056("E5006", "存在相同的申请"),
+    E5006("E5006", "存在相同的申请"),
+    E5007("E5007", "远程调用失败重试"),
     ;
     public final String code;