From 6cd47649fc179a379dc84e61b9a943935cb5d4ae Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Wed, 17 Dec 2025 15:52:41 +0800
Subject: [PATCH] =?UTF-8?q?fix(wxpay):=20=E4=BF=AE=E5=A4=8D=20Kafka=20?=
 =?UTF-8?q?=E7=94=9F=E4=BA=A7=E8=80=85=E7=A9=BA=E6=8C=87=E9=92=88=E5=BC=82?=
 =?UTF-8?q?=E5=B8=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 添加对 profitShareKafkaProducer 的空值检查
- 在发送消息前确保 Kafka 生产者已注入
- 使用 CompletableFuture 处理异步退款消息发送
- 设置 Kafka 生产者的注入模式为 required=false
- 避免因 Kafka 生产者缺失导致的服务启动失败
---
 .../service/mobile/impl/WxPayServiceImpl.java | 49 ++++++++++++-------
 1 file changed, 30 insertions(+), 19 deletions(-)

diff --git a/src/main/java/com/ycwl/basic/service/mobile/impl/WxPayServiceImpl.java b/src/main/java/com/ycwl/basic/service/mobile/impl/WxPayServiceImpl.java
index bddd9742..5992c06a 100644
--- a/src/main/java/com/ycwl/basic/service/mobile/impl/WxPayServiceImpl.java
+++ b/src/main/java/com/ycwl/basic/service/mobile/impl/WxPayServiceImpl.java
@@ -31,10 +31,12 @@ import com.ycwl.basic.repository.OrderRepository;
 import com.ycwl.basic.service.mobile.WxPayService;
 import com.ycwl.basic.service.pc.ScenicService;
 import com.ycwl.basic.utils.SnowFlakeUtil;
+import io.netty.util.concurrent.CompleteFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
 
 import jakarta.servlet.http.HttpServletRequest;
@@ -42,6 +44,7 @@ import java.io.IOException;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @Author: songmingsong
@@ -74,7 +77,7 @@ public class WxPayServiceImpl implements WxPayService {
     @Autowired
     @Lazy
     private RedisTemplate<String, String> redisTemplate;
-    @Autowired
+    @Autowired(required = false)
     private ProfitShareKafkaProducer profitShareKafkaProducer;
 
     @Override
@@ -126,15 +129,17 @@ public class WxPayServiceImpl implements WxPayService {
                     invalidateStatisticsCache(scenicId);
                 }
             });
-            profitShareKafkaProducer.sendProfitShareMessage(OrderMessage.builder()
-                    .orderId(callbackResponse.getOrderNo())
-                    .scenicId(scenicId)
-                    .totalAmount(callbackResponse.getPayPrice() / 100.0)
-                    .paymentSystem("wechat")
-                    .paymentOrderId(callbackResponse.getTransactionId())
-                    .timestamp(System.currentTimeMillis() / 1000)
-                    .build()
-            );
+            if (profitShareKafkaProducer != null) {
+                profitShareKafkaProducer.sendProfitShareMessage(OrderMessage.builder()
+                        .orderId(callbackResponse.getOrderNo())
+                        .scenicId(scenicId)
+                        .totalAmount(callbackResponse.getPayPrice() / 100.0)
+                        .paymentSystem("wechat")
+                        .paymentOrderId(callbackResponse.getTransactionId())
+                        .timestamp(System.currentTimeMillis() / 1000)
+                        .build()
+                );
+            }
         } catch (Exception e) {
             throw new AppException(BizCodeEnum.ADVANCE_PAYMENT_CALLBACK_FAILED, e.toString());
         }
@@ -157,15 +162,21 @@ public class WxPayServiceImpl implements WxPayService {
         IPayAdapter scenicPayAdapter = scenicService.getScenicPayAdapter(order.getScenicId());
         BigDecimal payPrice = order.getPayPrice();
         int priceInCents = payPrice.multiply(new BigDecimal(NumberConstant.HUNDRED)).intValue(); // 转换为分（int）
-        profitShareKafkaProducer.sendRefundMessage(RefundMessage.builder()
-                .refundOrderId(String.valueOf(order.getId()))
-                .originalOrderId(String.valueOf(order.getId()))
-                .refundAmount(payPrice.doubleValue())
-                .refundType("full")
-                .paymentSystem("wechat")
-                .timestamp(System.currentTimeMillis() / 1000)
-                .build()
-        ).whenComplete((result, throwable) -> {
+        CompletableFuture<SendResult<String, String>> future;
+        if (profitShareKafkaProducer != null) {
+            future = profitShareKafkaProducer.sendRefundMessage(RefundMessage.builder()
+                    .refundOrderId(String.valueOf(order.getId()))
+                    .originalOrderId(String.valueOf(order.getId()))
+                    .refundAmount(payPrice.doubleValue())
+                    .refundType("full")
+                    .paymentSystem("wechat")
+                    .timestamp(System.currentTimeMillis() / 1000)
+                    .build()
+            );
+        } else {
+            future = CompletableFuture.completedFuture(null);
+        }
+        future.whenComplete((result, throwable) -> {
             RefundOrderRequest request = new RefundOrderRequest()
                     .setOrderNo(orderId)
                     .setPrice(priceInCents)