乐尚代驾十订单支付seata、rabbitmq异步消息、redisson延迟队列

@Operation(summary = "获取客户OpenId")
@GetMapping("/getCustomerOpenId/{customerId}")
public Result<String> getCustomerOpenId(@PathVariable Long customerId) {
   return Result.ok(customerInfoService.getCustomerOpenId(customerId));
}




@Override
public String getCustomerOpenId(Long customerId) {
    LambdaQueryWrapper<CustomerInfo> wrapper = new LambdaQueryWrapper<>();
    wrapper.eq(CustomerInfo::getId,customerId);
    CustomerInfo customerInfo = customerInfoMapper.selectOne(wrapper);
    return customerInfo.getWxOpenId();
}




/**
 * 获取客户OpenId
 * @param customerId
 * @return
 */
@GetMapping("/customer/info/getCustomerOpenId/{customerId}")
Result<String> getCustomerOpenId(@PathVariable("customerId") Long customerId);

获取司机openid

• 与获取乘客openid基本一致

@Operation(summary = "获取订单支付信息")
@GetMapping("/getOrderPayVo/{orderNo}/{customerId}")
public Result<OrderPayVo> getOrderPayVo(@PathVariable String orderNo, @PathVariable Long customerId) {
    return Result.ok(orderInfoService.getOrderPayVo(orderNo, customerId));
}






@Override
public OrderPayVo getOrderPayVo(String orderNo, Long customerId) {
    OrderPayVo orderPayVo = orderInfoMapper.selectOrderPayVo(orderNo,customerId);
    if(orderPayVo != null) {
        String content = orderPayVo.getStartLocation() + " 到 "+orderPayVo.getEndLocation();
        orderPayVo.setContent(content);
    }
    return orderPayVo;
}





<select id="selectOrderPayVo" resultType="com.atguigu.daijia.model.vo.order.OrderPayVo">
    select
        info.id as order_id,
        info.customer_id,
        info.driver_id,
        info.order_no,
        info.start_location,
        info.end_location,
        info.status,
        bill.pay_amount,
        bill.coupon_amount

    from order_info info inner join order_bill bill on info.id = bill.order_id
    where info.customer_id = #{customerId}
      and info.order_no = #{orderNo}
</select>









/**
 * 获取订单支付信息
 * @param orderNo
 * @param customerId
 * @return
 */
@GetMapping("/order/info/getOrderPayVo/{orderNo}/{customerId}")
Result<OrderPayVo> getOrderPayVo(@PathVariable("orderNo") String orderNo, @PathVariable("customerId") Long customerId);

申请并绑定微信支付

微信支付商户平台：https://pay.weixin.qq.com/index.php/core/home/login

对于商家来说，想要开通微信支付，必须要去“微信支付商户平台”注册，然后把需要的资料提交上去，经过审核通过，你就开通了微信支付功能。

企业申请资料：

1、营业执照：彩色扫描件或数码照片

2、组织机构代码证：彩色扫描件或数码照片，若已三证合一，则无需提供

3、对公银行账户：包含开户行省市信息，开户账号

4、法人身份证：彩色扫描件或数码照片

如果想要在网站或者小程序上面使用微信支付，还要在微信公众平台上面关联你自己的微信商户账号。前提是你的微信开发者账号必须是企业身份，个人身份的开发者账号是无法调用微信支付API的。

支付密钥和数字证书

因为调用微信支付平台的API接口，必须要用到支付密钥和数字证书，这些参数在微信支付商户平台都可以获取。

<dependencies>
    <dependency>
        <groupId>com.github.wechatpay-apiv3</groupId>
        <artifactId>wechatpay-java</artifactId>
    </dependency>
</dependencies>



wx:
  v3pay:
    #小程序微信公众平台appId
    appid: wxcc651fcbab275e33
    #商户号
    merchantId: 163*******
    #商户API私钥路径
    privateKeyPath: /root/daijia/apiclient_key.pem
    #商户证书序列号
    merchantSerialNumber: 4AE80**********
    #商户APIV3密钥
    apiV3key: 84***************
    #异步回调地址
    notifyUrl: http://139.198.127.41:8600/payment/wxPay/notify






package com.atguigu.daijia.payment.config;

import com.wechat.pay.java.core.RSAAutoCertificateConfig;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix="wx.v3pay") //读取节点
@Data
public class WxPayV3Properties {

    private String appid;
    /** 商户号 */
    public String merchantId;
    /** 商户API私钥路径 */
    public String privateKeyPath;
    /** 商户证书序列号 */
    public String merchantSerialNumber;
    /** 商户APIV3密钥 */
    public String apiV3key;
    /** 回调地址 */
    private String notifyUrl;

    @Bean
    public RSAAutoCertificateConfig getConfig(){
        return new RSAAutoCertificateConfig.Builder()
                        .merchantId(this.getMerchantId())
                        .privateKeyFromPath(this.getPrivateKeyPath())
                        .merchantSerialNumber(this.getMerchantSerialNumber())
                        .apiV3Key(this.getApiV3key())
                        .build();

    }
}

微信支付接口

@Tag(name = "微信支付接口")
@RestController
@RequestMapping("payment/wxPay")
@Slf4j
public class WxPayController {

    @Operation(summary = "创建微信支付")
    @PostMapping("/createJsapi")
    public Result<WxPrepayVo> createWxPayment(@RequestBody PaymentInfoForm paymentInfoForm) {
        return Result.ok(wxPayService.createWxPayment(paymentInfoForm));
    }
}







@Override
public WxPrepayVo createWxPayment(PaymentInfoForm paymentInfoForm) {
    try {
        //1 添加支付记录到支付表里面
        //判断：如果表存在订单支付记录，不需要添加
        LambdaQueryWrapper<PaymentInfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(PaymentInfo::getOrderNo,paymentInfoForm.getOrderNo());
        PaymentInfo paymentInfo = paymentInfoMapper.selectOne(wrapper);
        if(paymentInfo == null) {
            paymentInfo = new PaymentInfo();
            BeanUtils.copyProperties(paymentInfoForm,paymentInfo);
            paymentInfo.setPaymentStatus(0);
            paymentInfoMapper.insert(paymentInfo);
        }

        //2 创建微信支付使用对象
        JsapiServiceExtension service =
                new JsapiServiceExtension.Builder().config(rsaAutoCertificateConfig).build();

        //3 创建request对象，封装微信支付需要参数
        PrepayRequest request = new PrepayRequest();
        Amount amount = new Amount();
        amount.setTotal(paymentInfoForm.getAmount().multiply(new BigDecimal(100)).intValue());
        request.setAmount(amount);
        request.setAppid(wxPayV3Properties.getAppid());
        request.setMchid(wxPayV3Properties.getMerchantId());
        //string[1,127]
        String description = paymentInfo.getContent();
        if(description.length() > 127) {
            description = description.substring(0, 127);
        }
        request.setDescription(description);
        request.setNotifyUrl(wxPayV3Properties.getNotifyUrl());
        request.setOutTradeNo(paymentInfo.getOrderNo());

        //获取用户信息
        Payer payer = new Payer();
        payer.setOpenid(paymentInfoForm.getCustomerOpenId());
        request.setPayer(payer);

        //是否指定分账，不指定不能分账
        SettleInfo settleInfo = new SettleInfo();
        settleInfo.setProfitSharing(true);
        request.setSettleInfo(settleInfo);

        //4 调用微信支付使用对象里面方法实现微信支付调用
        PrepayWithRequestPaymentResponse response = service.prepayWithRequestPayment(request);

        //5 根据返回结果，封装到WxPrepayVo里面
        WxPrepayVo wxPrepayVo = new WxPrepayVo();
        BeanUtils.copyProperties(response,wxPrepayVo);
        wxPrepayVo.setTimeStamp(response.getTimeStamp());
        return wxPrepayVo;
    }catch (Exception e) {
        throw new GuiguException(ResultCodeEnum.DATA_ERROR);
    }
}






@FeignClient(value = "service-payment")
public interface WxPayFeignClient {

    /**
     * 创建微信支付
     * @param paymentInfoForm
     * @return
     */
    @PostMapping("/payment/wxPay/createWxPayment")
    Result<WxPrepayVo> createWxPayment(@RequestBody PaymentInfoForm paymentInfoForm);
}









@Operation(summary = "创建微信支付")
@GuiguLogin
@PostMapping("/createWxPayment")
public Result<WxPrepayVo> createWxPayment(@RequestBody CreateWxPaymentForm createWxPaymentForm) {
    Long customerId = AuthContextHolder.getUserId();
    createWxPaymentForm.setCustomerId(customerId);
    return Result.ok(orderService.createWxPayment(createWxPaymentForm));
}






@Autowired
private WxPayFeignClient wxPayFeignClient;

@Override
public WxPrepayVo createWxPayment(CreateWxPaymentForm createWxPaymentForm) {
    //获取订单支付信息
    OrderPayVo orderPayVo = orderInfoFeignClient.getOrderPayVo(createWxPaymentForm.getOrderNo(),
            createWxPaymentForm.getCustomerId()).getData();
    //判断
    if(orderPayVo.getStatus() != OrderStatus.UNPAID.getStatus()) {
        throw new GuiguException(ResultCodeEnum.ILLEGAL_REQUEST);
    }

    //获取乘客和司机openid
    String customerOpenId = customerInfoFeignClient.getCustomerOpenId(orderPayVo.getCustomerId()).getData();

    String driverOpenId = driverInfoFeignClient.getDriverOpenId(orderPayVo.getDriverId()).getData();

    //封装需要数据到实体类，远程调用发起微信支付
    PaymentInfoForm paymentInfoForm = new PaymentInfoForm();
    paymentInfoForm.setCustomerOpenId(customerOpenId);
    paymentInfoForm.setDriverOpenId(driverOpenId);
    paymentInfoForm.setOrderNo(orderPayVo.getOrderNo());
    paymentInfoForm.setAmount(orderPayVo.getPayAmount());
    paymentInfoForm.setContent(orderPayVo.getContent());
    paymentInfoForm.setPayWay(1);

    WxPrepayVo wxPrepayVo = wxPayFeignClient.createWxPayment(paymentInfoForm).getData();
    return wxPrepayVo;
}

查询支付状态

• 判断微信支付是否成功，有两种方式，就是以上两种红字
• 微信支付成功/失败就会回调我们的接作
• 收到调用之后我们就可以继续接下来的操作
• 用消息队列保证数据的最终一致性，在支付成功后我们后续还有很多的操作，如果把所有的操作全部成功再向用户进行返回，那就太慢了，如果支付成功，我们就先向用户返回结果，并向消息队列发送关键信息用于后续的操作

根据订单编号查询支付状态

@Operation(summary = "支付状态查询")
@GetMapping("/queryPayStatus/{orderNo}")
public Result queryPayStatus(@PathVariable String orderNo) {
    return Result.ok(wxPayService.queryPayStatus(orderNo));
}






//查询支付状态
@Override
public Boolean queryPayStatus(String orderNo) {
    //1 创建微信操作对象
    JsapiServiceExtension service =
            new JsapiServiceExtension.Builder().config(rsaAutoCertificateConfig).build();

    //2 封装查询支付状态需要参数
    QueryOrderByOutTradeNoRequest queryRequest = new QueryOrderByOutTradeNoRequest();
    queryRequest.setMchid(wxPayV3Properties.getMerchantId());
    queryRequest.setOutTradeNo(orderNo);

    //3 调用微信操作对象里面方法实现查询操作
    Transaction transaction = service.queryOrderByOutTradeNo(queryRequest);

    //4 查询返回结果，根据结果判断
    if(transaction != null
            && transaction.getTradeState() == Transaction.TradeStateEnum.SUCCESS) {
        //5 如果支付成功，调用其他方法实现支付后处理逻辑
        this.handlePayment(transaction);

        return true;
    }
    return false;
}








/**
 * 支付状态查询
 * @param orderNo
 * @return
 */
@GetMapping("/payment/wxPay/queryPayStatus/{orderNo}")
Result<Boolean> queryPayStatus(@PathVariable("orderNo") String orderNo);








@Operation(summary = "支付状态查询")
@GuiguLogin
@GetMapping("/queryPayStatus/{orderNo}")
public Result<Boolean> queryPayStatus(@PathVariable String orderNo) {
    return Result.ok(orderService.queryPayStatus(orderNo));
}






@Override
public Boolean queryPayStatus(String orderNo) {
    return wxPayFeignClient.queryPayStatus(orderNo).getData();
}

微信支付成功回调接口

@Operation(summary = "微信支付异步通知接口")
@PostMapping("/notify")
public Map<String,Object> notify(HttpServletRequest request) {
    try {
        wxPayService.wxnotify(request);

        //返回成功
        Map<String,Object> result = new HashMap<>();
        result.put("code", "SUCCESS");
        result.put("message", "成功");
        return result;
    } catch (Exception e) {
        e.printStackTrace();
    }

    //返回失败
    Map<String,Object> result = new HashMap<>();
    result.put("code", "FAIL");
    result.put("message", "失败");
    return result;
}







    //微信支付成功后，进行的回调
    @Override
    public void wxnotify(HttpServletRequest request) {
//1.回调通知的验签与解密
        //从request头信息获取参数
        //HTTP 头 Wechatpay-Signature
        // HTTP 头 Wechatpay-Nonce
        //HTTP 头 Wechatpay-Timestamp
        //HTTP 头 Wechatpay-Serial
        //HTTP 头 Wechatpay-Signature-Type
        //HTTP 请求体 body。切记使用原始报文，不要用 JSON 对象序列化后的字符串，避免验签的 body 和原文不一致。
        String wechatPaySerial = request.getHeader("Wechatpay-Serial");
        String nonce = request.getHeader("Wechatpay-Nonce");
        String timestamp = request.getHeader("Wechatpay-Timestamp");
        String signature = request.getHeader("Wechatpay-Signature");
        String requestBody = RequestUtils.readData(request);

        //2.构造 RequestParam
        RequestParam requestParam = new RequestParam.Builder()
                .serialNumber(wechatPaySerial)
                .nonce(nonce)
                .signature(signature)
                .timestamp(timestamp)
                .body(requestBody)
                .build();
        
        //3.初始化 NotificationParser
        NotificationParser parser = new NotificationParser(rsaAutoCertificateConfig);
        //4.以支付通知回调为例，验签、解密并转换成 Transaction
        Transaction transaction = parser.parse(requestParam, Transaction.class);

        if(null != transaction && transaction.getTradeState() == Transaction.TradeStateEnum.SUCCESS) {
            //5.处理支付业务
            this.handlePayment(transaction);
        }
    }

内网穿透配置

• 还可以把项目部署到腾讯云、华为云等来解决这个问题
• 测试的时候我们没有域名，就可以使用内网穿透的工具

这里使用的工具是ngrok.cc

@Service
public class RabbitService {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    //发送消息
    public boolean sendMessage(String exchange,
                               String routingkey,
                               Object message) {
        rabbitTemplate.convertAndSend(exchange,routingkey,message);
        return true;
    }
}

• 修改nacos配置

//如果支付成功，调用其他方法实现支付后处理逻辑
public void handlePayment(Transaction transaction) {

    //1 更新支付记录，状态修改为 已经支付
    //订单编号
    String orderNo = transaction.getOutTradeNo();
    //根据订单编号查询支付记录
    LambdaQueryWrapper<PaymentInfo> wrapper = new LambdaQueryWrapper<>();
    wrapper.eq(PaymentInfo::getOrderNo,orderNo);
    PaymentInfo paymentInfo = paymentInfoMapper.selectOne(wrapper);
    //如果已经支付，不需要更新
    if(paymentInfo.getPaymentStatus() == 1) {
        return;
    }
    paymentInfo.setPaymentStatus(1);
    paymentInfo.setOrderNo(transaction.getOutTradeNo());
    paymentInfo.setTransactionId(transaction.getTransactionId());
    paymentInfo.setCallbackTime(new Date());
    paymentInfo.setCallbackContent(JSON.toJSONString(transaction));
    paymentInfoMapper.updateById(paymentInfo);

    //2 发送端：发送mq消息，传递 订单编号
    //  接收端：获取订单编号，完成后续处理
    rabbitService.sendMessage(MqConst.EXCHANGE_ORDER,
            MqConst.ROUTING_PAY_SUCCESS,
            orderNo);
}

接收端

@Component
public class PaymentReceiver {

    @Autowired
    private WxPayService wxPayService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = MqConst.QUEUE_PAY_SUCCESS,durable = "true"),
            exchange = @Exchange(value = MqConst.EXCHANGE_ORDER),
            key = {MqConst.ROUTING_PAY_SUCCESS}
    ))
    public void paySuccess(String orderNo, Message message, Channel channel) {
        wxPayService.handleOrder(orderNo);
    }
}






//支付成功后续处理
@Override
public void handleOrder(String orderNo) {
    //1 远程调用：更新订单状态：已经支付
    orderInfoFeignClient.updateOrderPayStatus(orderNo);

    //2 远程调用：获取系统奖励，打入到司机账户
    OrderRewardVo orderRewardVo = orderInfoFeignClient.getOrderRewardFee(orderNo).getData();
    if(orderRewardVo != null && orderRewardVo.getRewardFee().doubleValue()>0) {
        TransferForm transferForm = new TransferForm();
        transferForm.setTradeNo(orderNo);
        transferForm.setTradeType(TradeType.REWARD.getType());
        transferForm.setContent(TradeType.REWARD.getContent());
        transferForm.setAmount(orderRewardVo.getRewardFee());
        transferForm.setDriverId(orderRewardVo.getDriverId());
        driverAccountFeignClient.transfer(transferForm);
    }

    //3 TODO 其他

}

订单状态更新接口

@Operation(summary = "更改订单支付状态")
@GetMapping("/updateOrderPayStatus/{orderNo}")
public Result<Boolean> updateOrderPayStatus(@PathVariable String orderNo) {
    return Result.ok(orderInfoService.updateOrderPayStatus(orderNo));
}






@Override
public Boolean updateOrderPayStatus(String orderNo) {
    //1 根据订单编号查询，判断订单状态
    LambdaQueryWrapper<OrderInfo> wrapper = new LambdaQueryWrapper<>();
    wrapper.eq(OrderInfo::getOrderNo,orderNo);
    OrderInfo orderInfo = orderInfoMapper.selectOne(wrapper);
    if(orderInfo == null || orderInfo.getStatus() == OrderStatus.PAID.getStatus()) {
        return true;
    }
    
    //2 更新状态
    LambdaQueryWrapper<OrderInfo> updateWrapper = new LambdaQueryWrapper<>();
    updateWrapper.eq(OrderInfo::getOrderNo,orderNo);
    
    OrderInfo updateOrderInfo = new OrderInfo();
    updateOrderInfo.setStatus(OrderStatus.PAID.getStatus());
    updateOrderInfo.setPayTime(new Date());

    int rows = orderInfoMapper.update(updateOrderInfo, updateWrapper);
    
    if(rows == 1) {
        return true;
    } else {
        throw new GuiguException(ResultCodeEnum.UPDATE_ERROR);
    }
}






/**
 * 更改订单支付状态
 * @param orderNo
 * @return
 */
@GetMapping("/order/info//updateOrderPayStatus/{orderNo}")
Result<Boolean> updateOrderPayStatus(@PathVariable("orderNo") String orderNo);

获取订单系统奖励

@Operation(summary = "获取订单的系统奖励")
@GetMapping("/getOrderRewardFee/{orderNo}")
public Result<OrderRewardVo> getOrderRewardFee(@PathVariable String orderNo) {
    return Result.ok(orderInfoService.getOrderRewardFee(orderNo));
}





@Override
public OrderRewardVo getOrderRewardFee(String orderNo) {
    //根据订单编号查询订单表
    OrderInfo orderInfo = 
            orderInfoMapper.selectOne(
                    new LambdaQueryWrapper<OrderInfo>()
                            .eq(OrderInfo::getOrderNo, orderNo)
                            .select(OrderInfo::getId,OrderInfo::getDriverId));
    
    //根据订单id查询系统奖励表
    OrderBill orderBill = 
            orderBillMapper.selectOne(new LambdaQueryWrapper<OrderBill>()
                    .eq(OrderBill::getOrderId, orderInfo.getId())
                    .select(OrderBill::getRewardFee));
    
    //封装到vo里面
    OrderRewardVo orderRewardVo = new OrderRewardVo();
    orderRewardVo.setOrderId(orderInfo.getId());
    orderRewardVo.setDriverId(orderInfo.getDriverId());
    orderRewardVo.setRewardFee(orderBill.getRewardFee());
    return orderRewardVo;
}





/**
 * 获取订单的系统奖励
 * @param orderNo
 * @return
 */
@GetMapping("/order/info//getOrderRewardFee/{orderNo}")
Result<OrderRewardVo> getOrderRewardFee(@PathVariable("orderNo") String orderNo);

系统奖励打入司机账户

@Tag(name = "司机账户API接口管理")
@RestController
@RequestMapping(value="/driver/account")
@SuppressWarnings({"unchecked", "rawtypes"})
public class DriverAccountController {

    @Autowired
    private DriverAccountService driverAccountService;

    @Operation(summary = "转账")
    @PostMapping("/transfer")
    public Result<Boolean> transfer(@RequestBody TransferForm transferForm) {
        return Result.ok(driverAccountService.transfer(transferForm));
    }
}






@Override
public Boolean transfer(TransferForm transferForm) {
    //1 去重
    LambdaQueryWrapper<DriverAccountDetail> wrapper = new LambdaQueryWrapper<>();
    wrapper.eq(DriverAccountDetail::getTradeNo,transferForm.getTradeNo());
    Long count = driverAccountDetailMapper.selectCount(wrapper);
    if(count > 0) {
        return true;
    }

    //2 添加奖励到司机账户表
    driverAccountMapper.add(transferForm.getDriverId(),transferForm.getAmount());

    //3 添加交易记录 
    DriverAccountDetail driverAccountDetail = new DriverAccountDetail();
    BeanUtils.copyProperties(transferForm,driverAccountDetail);
    driverAccountDetailMapper.insert(driverAccountDetail);

    return true;
}








<update id="add">
   update driver_account
   set total_amount = total_amount + #{amount}, available_amount = available_amount + #{amount}, total_income_amount = total_income_amount + #{amount}
   where driver_id = #{driverId}
</update>







@FeignClient(value = "service-driver")
public interface DriverAccountFeignClient {

    /**
     * 转账
     * @param transferForm
     * @return
     */
    @PostMapping("/driver/account/transfer")
    Result<Boolean> transfer(@RequestBody TransferForm transferForm);
}

seata实现分布式事务

订单支付成功后，订单状态更新、获取订单系统奖励、系统奖励打入司机账户都是通过远程调用来实现的，我们就在这儿使用seata分布式事务

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>
<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <version>1.7.1</version>
</dependency>





seata.tx-service-group=tingshu-tx-group
seata.service.vgroup-mapping.tingshu-tx-group=default
seata.service.grouplist.default=127.0.0.1:8091




// 在事务的入口添加@GlobalTransactional
// 其他的小事务依然使用@Transactional

乘客下单功能

• 乘客登录之后，选择代驾开始和结束地址，之后乘客呼叫代驾
• 乘客呼叫代驾之后，等待司机接单
• 但是，如果在乘客呼叫代驾之后15分钟之后，如果还是没有司机接单，订单自动取消，如果15分钟以内有司机接单，完成代驾过程
• 总结：15分钟没有司机接单，自动取消订单
• 总体思路：使用延迟队列消息实现订单到时间自动取消功能
• 有很多种方式：

第一种 使用RabbitMQ里面TTL和死信队列实现

第二种 在RabbitMQ安装延迟队列实现

• 就是装个插件，在这个队列中的消息不能被消费，只有延时时间到了才能被消费，第三种 使用Redisson实现

使用RabbitMQ里面TTL和死信队列实现

• TTL：Time To Live，消息存活时间

发送端：发送消息，设置存活时间10s

接收端：发送完成之后，10s以内如果从队列获取发送过来消息，操作结束

​ 如果超过10s消息没有被消费，消息超时了，无法被消费，成为死信

• 死信队列：无法被消费的消息

– 超过存活时间没有被消费的消息

– 消息端拒绝接收，不能放回队列里面

– 队列最大长度

使用Redisson实现

• 利用redissonClient 发送延迟消息

redissonClient.getBlockingDeque(): 创建一个阻塞队列

redissonClient.getDelayedQueue(): 获取延迟队列

delayedQueue.offer(): 向队列中存储数据

blockingDeque.take(): 从队列中获取数据

//乘客下单
@Override
public Long saveOrderInfo(OrderInfoForm orderInfoForm) {
    //order_info添加订单数据
    OrderInfo orderInfo = new OrderInfo();
    BeanUtils.copyProperties(orderInfoForm,orderInfo);
    //订单号
    String orderNo = UUID.randomUUID().toString().replaceAll("-","");
    orderInfo.setOrderNo(orderNo);
    //订单状态
    orderInfo.setStatus(OrderStatus.WAITING_ACCEPT.getStatus());
    orderInfoMapper.insert(orderInfo);
    
    //生成订单之后，发送延迟消息
    this.sendDelayMessage(orderInfo.getId());

    //记录日志
    this.log(orderInfo.getId(),orderInfo.getStatus());

    //向redis添加标识
    //接单标识，标识不存在了说明不在等待接单状态了
    redisTemplate.opsForValue().set(RedisConstant.ORDER_ACCEPT_MARK,
            "0", RedisConstant.ORDER_ACCEPT_MARK_EXPIRES_TIME, TimeUnit.MINUTES);

    return orderInfo.getId();
}

• 编写发送延迟消息的方法

//生成订单之后，发送延迟消息
private void sendDelayMessage(Long orderId) {
    try{
        //1 创建队列
        RBlockingQueue<Object> blockingDueue = redissonClient.getBlockingQueue("queue_cancel");

        //2 把创建队列放到延迟队列里面
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDueue);
        
        //3 发送消息到延迟队列里面
        //设置过期时间
        delayedQueue.offer(orderId.toString(),15,TimeUnit.MINUTES);
        
    }catch (Exception e) {
        e.printStackTrace();
        throw new GuiguException(ResultCodeEnum.DATA_ERROR);
    }
}

• 监听延迟队列消息

//监听延迟队列
@Component
public class RedisDelayHandle {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private OrderInfoService orderInfoService;

    @PostConstruct
    public void listener() {
        new Thread(()->{
        // while true 实现一直监听
            while(true) {
                //获取延迟队列里面阻塞队列
                RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue("queue_cancel");

                //从队列获取消息
                try {
                    String orderId = blockingQueue.take();

                    //取消订单
                    if(StringUtils.hasText(orderId)) {
                        //调用方法取消订单
                        orderInfoService.orderCancel(Long.parseLong(orderId));
                    }

                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

            }
        }).start();
    }
}








//调用方法取消订单
@Override
public void orderCancel(long orderId) {
    //orderId查询订单信息
    OrderInfo orderInfo = orderInfoMapper.selectById(orderId);
    //判断
    if(orderInfo.getStatus()==OrderStatus.WAITING_ACCEPT.getStatus()) {
        //修改订单状态：取消状态
        orderInfo.setStatus(OrderStatus.CANCEL_ORDER.getStatus());
        int rows = orderInfoMapper.updateById(orderInfo);
        if(rows == 1) {
            //删除接单标识

            redisTemplate.delete(RedisConstant.ORDER_ACCEPT_MARK);
        }
    }
}