商城商店步骤学习完成

order-service/src/main/java/cn/mayiming/Consumer/PaySuccessConsumer.java

@@ -0,0 +1,47 @@
+package cn.mayiming.Consumer;
+
+import cn.mayiming.Mapper.OrderMapper;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Map;
+
+@Component
+@RocketMQMessageListener(
+        topic = "pay_success_topic",
+        consumerGroup = "order-pay-success-consumer"
+)
+public class PaySuccessConsumer implements RocketMQListener<Map<String, Object>> {
+
+    @Autowired
+    private OrderMapper orderMapper;
+
+    @Transactional(rollbackFor = Exception.class)
+    @Override
+    public void onMessage(Map<String, Object> msg) {
+        try {
+            String orderNo = msg.get("orderNo").toString();
+            System.out.println("========== 订单服务接收支付成功消息 ==========");
+            System.out.println("订单号：" + orderNo);
+            System.out.println("支付金额：" + msg.get("payAmount"));
+            System.out.println("===========================================");
+
+            // 核心：更新订单状态为「已支付」（假设1=已支付）
+            int updateResult = orderMapper.updateOrderStatus(orderNo, 0, 1);
+            if (updateResult == 0) {
+                throw new RuntimeException("更新订单状态失败，订单号：" + orderNo);
+            }
+            System.out.println("订单" + orderNo + "已更新为「已支付」状态");
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.err.println("消费支付成功消息失败：" + e.getMessage());
+            // 生产环境：抛出异常触发MQ重试，确保订单状态最终一致
+            throw new RuntimeException("消费失败，触发重试", e);
+        }
+    }
+}
+

order-service/src/main/java/cn/mayiming/Mapper/OrderMapper.java

@@ -78,4 +78,8 @@ public interface OrderMapper {
         // 简单生成规则：时间戳 + 6位随机数
         return System.currentTimeMillis() + "" + (int)(Math.random() * 900000 + 100000);
     }
+
+    // OrderMapper补充更新订单状态方法
+    @Update("UPDATE t_order SET order_status = #{newStatus}, update_time = CURRENT_TIMESTAMP WHERE order_no = #{orderNo} AND order_status = #{oldStatus}")
+    int updateOrderStatus(@Param("orderNo") String orderNo, @Param("oldStatus") Integer oldStatus, @Param("newStatus") Integer newStatus);
 }

pay-service/pom.xml

@@ -82,6 +82,10 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-spring-boot-starter</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>5.3.2</version>  <!-- 这个存在于 Central -->
+        </dependency>
     </dependencies>
 </project>

pay-service/src/main/java/cn/mayiming/App.java

@@ -1,13 +1,17 @@
 package cn.mayiming;
 
+import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 /**
  * Hello world!
  *
  */
 @SpringBootApplication
+@EnableTransactionManagement
+@MapperScan("cn.mayiming.Mapper")
 public class App 
 {
     public static void main( String[] args )

pay-service/src/main/java/cn/mayiming/Consumer/payConsumer.java

@@ -1,6 +1,7 @@
 package cn.mayiming.Consumer;
 
 import cn.mayiming.Entity.PayTriggerMsgDTO;
+import cn.mayiming.Service.PayService;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -14,6 +15,9 @@ import org.springframework.stereotype.Component;
 )
 public class payConsumer implements RocketMQListener<PayTriggerMsgDTO> {
 
+    @Autowired
+    private PayService payService;
+
     @Override
     public void onMessage(PayTriggerMsgDTO msg) {
         // 消费者接收到消息，模拟支付流程（本地仅生成支付链接，记录到日志/订单）
@@ -25,6 +29,13 @@ public class payConsumer implements RocketMQListener<PayTriggerMsgDTO> {
             System.out.println("模拟支付链接：http://localhost:8080/order/mock/pay?orderNo=" + msg.getOrderNo());
             System.out.println("=======================================");
 
+            // 核心：调用PayService处理（写入幂等记录+支付记录）
+            payService.handlePayTrigger(msg);
+
+            // 生成模拟支付链接（包含订单号+幂等ID，供前端调用）
+            String mockPayUrl = "http://localhost:8080/pay/mock/success?orderNo=" + msg.getOrderNo() + "&businessId=" + msg.getBusinessId();
+            System.out.println("模拟支付链接：" + mockPayUrl);
+            System.out.println("=============================================");
 
         } catch (Exception e) {
             e.printStackTrace();

pay-service/src/main/java/cn/mayiming/Controller/payController.java

@@ -0,0 +1,33 @@
+package cn.mayiming.Controller;
+
+import cn.mayiming.Entity.checkpayDTO;
+import cn.mayiming.Service.PayService;
+import jakarta.annotation.Resource;
+import jakarta.servlet.http.HttpServletRequest;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+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 java.util.HashMap;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/pay")
+public class payController {
+    @Autowired
+    PayService payService;
+
+    @PostMapping("/paycheck")
+    public Map<String, Object> checkpay(@RequestBody checkpayDTO checkpayDTO) {
+        Map<String, Object> result = new HashMap<>();
+        boolean isSuccess = payService.isPaySuccess(checkpayDTO.getOrderno());
+        result.put("success", true);
+        result.put("orderNo", checkpayDTO.getOrderno());
+        result.put("isPaySuccess", isSuccess);
+        result.put("msg", isSuccess ? "该订单已支付成功" : "该订单未支付或不存在");
+        return result;
+    }
+}

pay-service/src/main/java/cn/mayiming/Entity/IdempotentRecord.java

@@ -0,0 +1,14 @@
+package cn.mayiming.Entity;
+
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+public class IdempotentRecord {
+    private Long id;
+    private String requestId; // 对应表的request_id（前端唯一请求ID/订单号）
+    private Integer status;   // 0-处理中 1-处理成功 2-处理失败
+    private Date createTime;
+    private Date updateTime;
+}

pay-service/src/main/java/cn/mayiming/Entity/PayRecord.java

@@ -0,0 +1,15 @@
+package cn.mayiming.Entity;
+
+import lombok.Data;
+
+import java.math.BigDecimal;
+import java.util.Date;
+
+@Data
+public class PayRecord {
+    private Long id;
+    private String orderNo;    // 关联订单号
+    private BigDecimal payAmount; // 支付金额
+    private Integer payStatus; // 0-未支付 1-支付成功
+    private Date createTime;
+}

pay-service/src/main/java/cn/mayiming/Entity/checkpayDTO.java

@@ -0,0 +1,10 @@
+package cn.mayiming.Entity;
+
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+@Data
+public class checkpayDTO {
+    @NotNull
+    private String orderno;
+}

pay-service/src/main/java/cn/mayiming/Mapper/payMapper.java

@@ -1,5 +1,82 @@
 package cn.mayiming.Mapper;
 
+import cn.mayiming.Entity.IdempotentRecord;
+import cn.mayiming.Entity.PayRecord;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
+
+import java.math.BigDecimal;
+
+@Repository
 public interface payMapper {
 
-}
+    // ==================== 幂等记录（idempotent_record）相关操作 ====================
+    /**
+     * 根据requestId查询幂等记录
+     * @param requestId 前端唯一请求ID（对应businessId）
+     * @return 幂等记录对象
+     */
+    @Select("SELECT id, request_id AS requestId, status, create_time AS createTime, update_time AS updateTime " +
+            "FROM idempotent_record WHERE request_id = #{requestId}")
+    IdempotentRecord selectIdempotentByRequestId(@Param("requestId") String requestId);
+
+    /**
+     * 插入幂等记录（初始状态：0-处理中）
+     * @param requestId 唯一请求ID
+     * @param status 处理状态
+     * @return 插入成功返回1，失败返回0
+     */
+    @Insert("INSERT INTO idempotent_record (request_id, status, create_time, update_time) " +
+            "VALUES (#{requestId}, #{status}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)")
+    int insertIdempotentRecord(@Param("requestId") String requestId, @Param("status") Integer status);
+
+    /**
+     * 更新幂等记录状态
+     * @param requestId 唯一请求ID
+     * @param status 新状态（1-成功/2-失败）
+     * @return 更新成功返回1，无匹配记录返回0
+     */
+    @Update("UPDATE idempotent_record SET status = #{status}, update_time = CURRENT_TIMESTAMP " +
+            "WHERE request_id = #{requestId}")
+    int updateIdempotentStatus(@Param("requestId") String requestId, @Param("status") Integer status);
+
+    // ==================== 支付记录（t_pay_record）相关操作 ====================
+    /**
+     * 根据订单号查询支付记录
+     * @param orderNo 订单号
+     * @return 支付记录对象
+     */
+    @Select("SELECT id, order_no AS orderNo, pay_amount AS payAmount, pay_status AS payStatus, create_time AS createTime " +
+            "FROM t_pay_record WHERE order_no = #{orderNo}")
+    PayRecord selectPayRecordByOrderNo(@Param("orderNo") String orderNo);
+
+    /**
+     * 插入支付记录（初始状态：0-未支付）
+     * @param orderNo 订单号
+     * @param payAmount 支付金额
+     * @param payStatus 支付状态
+     * @return 插入成功返回1，失败返回0
+     */
+    @Insert("INSERT INTO t_pay_record (order_no, pay_amount, pay_status, create_time) " +
+            "VALUES (#{orderNo}, #{payAmount}, #{payStatus}, CURRENT_TIMESTAMP)")
+    int insertPayRecord(@Param("orderNo") String orderNo,
+                        @Param("payAmount") BigDecimal payAmount,
+                        @Param("payStatus") Integer payStatus);
+
+    /**
+     * 更新支付记录状态（模拟支付成功时调用）
+     * @param orderNo 订单号
+     * @param payStatus 新状态（1-支付成功）
+     * @return 更新成功返回1，无匹配记录返回0
+     */
+    @Update("UPDATE t_pay_record SET pay_status = #{payStatus}, create_time = CURRENT_TIMESTAMP " +
+            "WHERE order_no = #{orderNo} AND pay_status = 0")
+    int updatePayRecordStatus(@Param("orderNo") String orderNo, @Param("payStatus") Integer payStatus);
+
+
+    @Select("SELECT COUNT(1) FROM t_pay_record WHERE order_no = #{orderNo} AND pay_status = 1")
+    int checkPaySuccessByOrderNo(@Param("orderNo") String orderNo);
+}

pay-service/src/main/java/cn/mayiming/Service/PayService.java

@@ -0,0 +1,139 @@
+package cn.mayiming.Service;
+
+import cn.mayiming.Entity.IdempotentRecord;
+import cn.mayiming.Entity.PayRecord;
+import cn.mayiming.Entity.PayTriggerMsgDTO;
+import cn.mayiming.Mapper.payMapper;
+import jakarta.annotation.Resource;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+@Service
+public class PayService {
+    // 替换为你实际的 payMapper（整合了幂等+支付记录操作）
+    @Autowired
+    private payMapper payMapper;
+
+
+    @Resource
+    private RocketMQTemplate rocketMQTemplate;
+
+    private static final String PAY_SUCCESS_TOPIC = "pay_success_topic";
+
+
+    /**
+     * 处理订单支付触发消息（核心：事务+幂等）
+     * @param msg 订单服务发送的支付触发消息
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void handlePayTrigger(PayTriggerMsgDTO msg) {
+        // 1. 幂等校验：用businessId作为requestId（唯一标识）
+        String requestId = msg.getBusinessId();
+        IdempotentRecord existRecord = payMapper.selectIdempotentByRequestId(requestId);
+        if (existRecord != null) {
+            // 已存在幂等记录，直接返回（避免重复处理）
+            System.out.println("幂等记录已存在，requestId：" + requestId + "，无需重复处理");
+            return;
+        }
+
+        try {
+            // 2. 写入幂等记录（初始状态：0-处理中）
+            int insertIdempotentResult = payMapper.insertIdempotentRecord(requestId, 0);
+            if (insertIdempotentResult == 0) {
+                throw new RuntimeException("幂等记录插入失败，requestId：" + requestId);
+            }
+
+            // 3. 校验支付记录是否已存在（双重幂等保障）
+            PayRecord existPayRecord = payMapper.selectPayRecordByOrderNo(msg.getOrderNo());
+            if (existPayRecord != null) {
+                // 更新幂等记录为成功，返回
+                payMapper.updateIdempotentStatus(requestId, 1);
+                System.out.println("支付记录已存在，订单号：" + msg.getOrderNo());
+                return;
+            }
+
+            // 4. 写入支付记录（初始状态：0-未支付）
+            BigDecimal payAmount = msg.getPayAmount() == null ? new BigDecimal("0.00") : msg.getPayAmount();
+            int insertPayResult = payMapper.insertPayRecord(msg.getOrderNo(), payAmount, 0);
+            if (insertPayResult == 0) {
+                throw new RuntimeException("支付记录插入失败，订单号：" + msg.getOrderNo());
+            }
+
+            // 5. 所有操作成功，更新幂等记录为处理成功
+            payMapper.updateIdempotentStatus(requestId, 1);
+            System.out.println("支付记录创建成功，订单号：" + msg.getOrderNo() + "，幂等ID：" + requestId);
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            // 6. 操作失败，更新幂等记录为处理失败
+            try {
+                payMapper.updateIdempotentStatus(requestId, 2);
+            } catch (Exception ex) {
+                System.err.println("更新幂等记录为失败状态失败，requestId：" + requestId);
+            }
+            throw new RuntimeException("处理支付触发消息失败，订单号：" + msg.getOrderNo(), e);
+        }
+    }
+
+    /**
+     * 模拟支付成功（更新支付记录状态）
+     * @param orderNo 订单号
+     * @param businessId 幂等ID
+     * @return 操作结果
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public boolean mockPaySuccess(String orderNo, String businessId) {
+        // 1. 原有校验逻辑
+        IdempotentRecord idempotentRecord = payMapper.selectIdempotentByRequestId(businessId);
+        if (idempotentRecord == null || idempotentRecord.getStatus() != 1) {
+            System.err.println("幂等记录异常，requestId：" + businessId);
+            return false;
+        }
+        PayRecord payRecord = payMapper.selectPayRecordByOrderNo(orderNo);
+        if (payRecord == null || payRecord.getPayStatus() != 0) {
+            System.err.println("支付记录异常，订单号：" + orderNo);
+            return false;
+        }
+
+        // 2. 更新支付记录为成功
+        int updateResult = payMapper.updatePayRecordStatus(orderNo, 1);
+        if (updateResult == 0) {
+            throw new RuntimeException("更新支付状态失败，订单号：" + orderNo);
+        }
+
+        // ========== 新增：支付成功后发送MQ消息给订单服务 ==========
+        try {
+            // 构建支付成功消息体
+            Map<String, Object> paySuccessMsg = new HashMap<>();
+            paySuccessMsg.put("orderNo", orderNo);
+            paySuccessMsg.put("payAmount", payRecord.getPayAmount());
+            paySuccessMsg.put("payTime", new Date());
+            paySuccessMsg.put("businessId", businessId);
+
+            // 发送消息（同步发送，确保订单服务能收到）
+            rocketMQTemplate.convertAndSend(PAY_SUCCESS_TOPIC, paySuccessMsg);
+            System.out.println("支付成功消息已发送，订单号：" + orderNo);
+        } catch (Exception e) {
+            e.printStackTrace();
+            // 生产环境：消息发送失败需重试/告警，避免订单状态不一致
+            throw new RuntimeException("发送支付成功消息失败，订单号：" + orderNo, e);
+        }
+
+        return true;
+    }
+
+
+    public boolean isPaySuccess(String orderNo) {
+        // 调用新增的mapper方法，返回1则表示支付成功
+        int count = payMapper.checkPaySuccessByOrderNo(orderNo);
+        return count > 0;
+    }
+
+
+}

pay-service/src/main/resources/application.yml

@@ -43,7 +43,7 @@ rocketmq:
   name-server: 127.0.0.1:9876
   consumer:
     # 消费者组名（必须唯一，建议按服务+用途命名）
-    group: order-service-consumer
+    group: pay-service-consumer
     # 消费模式：CONCURRENTLY（并发消费，默认）/ORDERLY（顺序消费）
     consume-mode: CONCURRENTLY
     # 批量消费最大条数（默认1，单条消费）