devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

学成在线day13 支付通知，2024年最新讲的明明白白

作者：羊村懒王 | 2024-05-12 19:01:16

赞

踩

学成在线day13 支付通知，2024年最新讲的明明白白
            result -> {
                if(result.isAck()){
                    // 3.1.ack，消息成功
                    log.debug("通知支付结果消息发送成功, ID:{}", correlationData.getId());
                    //删除消息表中的记录
                    mqMessageService.completed(message.getId());
                }else{
                    // 3.2.nack，消息失败
                    log.error("通知支付结果消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                }
            },
            ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 发送消息 因为是广播模式所以路由键为空
    rabbitTemplate.convertAndSend(PayNotifyConfig.PAYNOTIFY_EXCHANGE_FANOUT, "", msgObj,correlationData);
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16


* **confirm callback（确认回调）**：当消息成功到达 Exchange 时，Broker 会发送一个确认给生产者。如果消息发送失败，则会触发 confirm callback 中的相应逻辑。确认回调通常用于确保消息被成功发送到 Exchange，以便生产者可以知道是否需要重新发送消息。
* **return callback（返回回调）**：当消息被路由到 Exchange 但无法路由到 Queue 时，Broker 会发送一个返回消息给生产者。返回回调通常用于处理无法路由到 Queue 的消息，生产者可以在返回回调中执行相应的逻辑，比如记录日志或者采取其他措施。


在本项目中，生产者消息投递成功到交换机时，执行confirmCallBack删除mq\_message表中的记录。如果消息从交换机投递到队列失败时，则把消息再次添加回mq\_message表，并且进行重试（重试次数在配置文件中配置）。




  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
publisher-confirm-type: correlated #correlated 异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
publisher-returns: true #开启publish-return功能，同样是基于callback机制，需要定义ReturnCallback
template:
mandatory: true #定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息
listener:
simple:
prefetch: 1 #每次只能获取一条消息，处理完成才能获取下一个消息
acknowledge-mode: manual #auto:出现异常时返回unack，消息回滚到mq；没有异常，返回ack ,manual:手动控制,none:丢弃消息，不回滚到mq
retry:
enabled: true #开启消费者失败重试
initial-interval: 1000ms #初识的失败等待时长为1秒
multiplier: 1 #失败的等待时长倍数，下次等待时长 = multiplier * last-interval
max-attempts: 3 #最大重试次数
stateless: true #true无状态；false有状态。如果业务中包含事务，这里改为false
direct:
acknowledge-mode: manual



> 
> Q1：如果消费者未收到消息或者出现系统错误时，是否会触发回调函数？
> 
> 
> 


在 RabbitMQ 中，当消费者未收到消息或者出现系统错误时，不会触发回调函数。回调函数是在消息生产者发送消息到 RabbitMQ 服务器后触发的，用于确认消息是否成功发送到 Exchange 或者是否被正确路由到 Queue。


如果消费者未收到消息，可能有以下几种原因：


* **消息丢失**：在消息生产者发送消息到 RabbitMQ 服务器后，由于网络问题或者其他原因导致消息丢失，这种情况下消费者是不会收到消息的。解决这个问题需要确保消息生产者发送消息时能够成功连接到 RabbitMQ 服务器，并且消息能够正确发送。
* **消费者无法处理消息**：如果消费者无法正确处理消息，可能是因为消费者代码出现了错误或者处理消息的逻辑不完善，这种情况下 RabbitMQ 不会触发回调函数。解决这个问题需要检查消费者代码逻辑是否正确，并确保能够处理各种异常情况。
* **队列配置问题**：如果队列配置不正确，比如消费者没有正确绑定到队列，或者队列设置了错误的参数，可能会导致消费者无法接收到消息。解决这个问题需要检查队列的配置，并确保消费者能够正确订阅到消息。


#### 2、接口定义


生产者业务层：




  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
/**
 * 请求支付宝查询支付结果
 *
 * @param payNo 支付记录id
 * @return 支付记录信息
 */
@Override
public PayRecordDto queryPayResult(String payNo) {
    //获得初始化的AlipayClient
    AlipayClient alipayClient = new DefaultAlipayClient(AlipayConfig.URL,
            APP_ID,APP_PRIVATE_KEY,
            "json",
            AlipayConfig.CHARSET,
            ALIPAY_PUBLIC_KEY,
            AlipayConfig.SIGNTYPE);
    AlipayTradeQueryRequest request = new AlipayTradeQueryRequest();
    JSONObject bizContent = new JSONObject();
    bizContent.put("out_trade_no", payNo);
    request.setBizContent(bizContent.toString());
    AlipayTradeQueryResponse response = null;
    try {
        response = alipayClient.execute(request);
    } catch (AlipayApiException e) {
        log.error("{}：查询支付宝支付结果错误！",payNo);
        return null;
    }
    if (!response.isSuccess()) {
        log.error("{}：查询支付宝支付结果失败！",payNo);
        return null;
    }
    String resultJson = response.getBody();
    //转map
    Map resultMap = JSON.parseObject(resultJson, Map.class);
    Map alipay_trade_query_response = (Map) resultMap.get("alipay_trade_query_response");
    //交易状态
    String tradeStatus = (String) alipay_trade_query_response.get("trade_status");
    //支付宝交易号
    String tradeNo = (String) alipay_trade_query_response.get("trade_no");
    PayStatusDto payStatusDto = new PayStatusDto();
    payStatusDto.setTrade_status(tradeStatus);
    payStatusDto.setTrade_no(tradeNo);
    payStatusDto.setOut_trade_no(payNo);
    payStatusDto.setApp_id(APP_ID);
    //处理订单状态
    return this.handlePayStatus(payStatusDto);
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46

/**
* 处理订单状态，更新xc_pay_record表
* @return
*/
public PayRecordDto handlePayStatus(PayStatusDto dto) {
PayRecordDto payRecordDto = new PayRecordDto();
String payNo = dto.getOut_trade_no();
String tradeNo = dto.getTrade_no();
String tradeStatus = dto.getTrade_status();

    XcPayRecord xcPayRecord = xcPayRecordMapper.selectOne(new LambdaQueryWrapper<XcPayRecord>().eq(XcPayRecord::getPayNo, payNo));
    if (null == xcPayRecord ){
        log.error("{}:查询订单记录不存在！",tradeNo);
        XueChengPlusException.cast("查询订单记录不存在！");
    }
    if (xcPayRecord.getStatus().equals("601002")){
        BeanUtils.copyProperties(xcPayRecord,payRecordDto);
        return payRecordDto;
    }

    //修改xc_pay_record和xc_orders 交易状态
    switch (tradeStatus) {
        case "TRADE_CLOSED":
            xcPayRecord.setStatus("601003");
            break;
        case "TRADE_SUCCESS":
        case "TRADE_FINISHED":
            xcPayRecord.setStatus("601002");
            xcPayRecord.setPaySuccessTime(LocalDateTime.now());
            break;
    }
    xcPayRecord.setOutPayNo(tradeNo);
    xcPayRecord.setOutPayChannel("alipay");
    xcPayRecordMapper.updateById(xcPayRecord);
    BeanUtils.copyProperties(xcPayRecord,payRecordDto);
    //通过orderId查询订单表信息
    Long orderId = xcPayRecord.getOrderId();
    XcOrders xcOrders = ordersMapper.selectOne(new LambdaQueryWrapper<XcOrders>().eq(XcOrders::getId, orderId));
    if (null == xcOrders ){
        log.error("{}:查询订单不存在！",tradeNo);
        XueChengPlusException.cast("查询订单不存在！");
    }
    //存入mq_message表
    MqMessage message = mqMessageService.addMessage("payresult_notify", xcOrders.getOutBusinessId(), xcOrders.getOrderType(), null);
    //将消息发送至队列，通知learning服务
    this.notifyPayResult(message);
    return payRecordDto;
}

/**
 * 发送通知结果
 *
 * @param message
 */
@Override
public void notifyPayResult(MqMessage message) {
    //1、消息体，转json
    String msg = JSON.toJSONString(message);
    //设置消息持久化
    Message msgObj = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .build();
    // 2.全局唯一的消息ID，需要封装到CorrelationData中
    CorrelationData correlationData = new CorrelationData(message.getId().toString());
    // 3.添加callback 回调函数，接收方接受成功或失败后回调
    correlationData.getFuture().addCallback(
            result -> {
                if(result.isAck()){
                    // 3.1.ack，消息成功
                    log.debug("通知支付结果消息发送成功, ID:{}", correlationData.getId());
                    //删除消息表中的记录
                    mqMessageService.completed(message.getId());
                }else{
                    // 3.2.nack，消息失败
                    log.error("通知支付结果消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());
                }
            },
            ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage())
    );
    // 发送消息 因为是广播模式所以路由键为空
    rabbitTemplate.convertAndSend(PayNotifyConfig.PAYNOTIFY_EXCHANGE_FANOUT, "", msgObj,correlationData);
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72

支付宝异步回调接口同理也调用了handlePayStatus方法


消费者业务层：




  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

@Service
@Slf4j
@Transactional
public class ReceviceOrderMessageServiceImpl implements ReceviceOrderMessageService {

/*
选课记录表
 */
@Autowired
private XcChooseCourseMapper chooseCourseMapper;

/*
课程表
 */
@Autowired
private XcCourseTablesMapper courseTablesMapper;

@Autowired
private MyCourseTables myCourseTables;

@Autowired
private ReceviceOrderMessageService receviceOrderMessageService;

@Resource
private RedisTemplate<String,Integer> redisTemplate;

@Resource
private RedissonClient redissonClient;

private static final String QUEUE_KEY = "learning:";

private static final String REDISSON_KEY = "redisson:";

@Override
@RabbitListener(queues = PayNotifyConfig.PAYNOTIFY_QUEUE)
public void receiveMessage(Message message, Channel channel)  {
    byte[] body = message.getBody();
    long deliverTag = message.getMessageProperties().getDeliveryTag();
    String jsonStr = new String(body);
    MqMessage mqMessage = JSON.parseObject(jsonStr, MqMessage.class);
    //得到选课id
    String businessKey1 = mqMessage.getBusinessKey1();
    //得到课程类型
    String businessKey2 = mqMessage.getBusinessKey2();
    //处理课程信息
    Integer count = 0;
    try {
        count = redisTemplate.opsForValue().get(QUEUE_KEY + businessKey1);
        if (!ObjectUtils.isEmpty(count) && count == 3){
            log.error("消息已达到最大重试次数:{},作丢弃处理",count);
            channel.basicNack(deliverTag,false,false);
            return;
        }
        this.receviceOrderMessageService.handleCourseDBData(businessKey1,businessKey2);
        int i = 1/0;
        channel.basicAck(deliverTag,true);
        log.info("消息手动确认成功！");
    } catch (Exception e) {
        try {
            channel.basicNack(deliverTag,false,true);
            RLock lock = redissonClient.getLock(REDISSON_KEY + businessKey1);
            if (lock.tryLock()){
                try {
                    redisTemplate.opsForValue().set(QUEUE_KEY + businessKey1, count == null ? 1 : count + 1,60, TimeUnit.SECONDS);
                } finally {
                    lock.unlock();
                }
            }
        } catch (IOException ex) {
            log.error("重新放入队列失败，失败原因:{}",e.getMessage(),e);
        }
        log.error("消费者出错,mq参数:{}，错误信息：{}",message,e.getMessage(),e);
    }
}

/**
 * 处理课程表，选课记录表
 * @param businessKey1 选课表id
 * @param businessKey2 课程类型
 * @return
 */
public boolean handleCourseDBData(String businessKey1, String businessKey2) {
    //根据课程id查询选课记录表
    XcChooseCourse xcChooseCourse = chooseCourseMapper.selectOne(new LambdaQueryWrapper<XcChooseCourse>().
            eq(XcChooseCourse::getId, businessKey1));
    if (ObjectUtils.isEmpty(xcChooseCourse)){
        log.error("根据课程ID：{}查询到的选课记录为空！",businessKey1);
        return false;
    }
    if (!businessKey2.equals("60201")){
        return false;
    }
    //修改状态为701001选课成功
    xcChooseCourse.setStatus("701001");
    int i = chooseCourseMapper.updateById(xcChooseCourse);
    if (i<1){
        log.error("更新选课记录表失败！选课ID：{}",businessKey1);
        XueChengPlusException.cast("更新选课记录表失败！");
    }
    //向课程表插入记录
    this.myCourseTables.initCourseTableData(xcChooseCourse);
    return true;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98

}


**此处消费者使用手动确认消息+异常重试机制。异常重试超过3次就丢弃消息，也可以使用手动确认消息+死信队列模式。**



> 
> Q2：在配置文件开启了消费者手动确认，如果消费者拒绝签收，那么生产者的ConfirmCallback和ReturnCallBack回调方法会触发吗，这条消息会怎么样？
> 
> 
> 


如果消费者拒绝签收消息（通过调用basicNack或basicReject方法），生产者的confirmCallback和returnCallback回调方法不会触发。因为消息已经被消费者处理完毕，不会再返回给生产者。


**自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。**

**深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！**

**因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。**
![img](https://img-blog.csdnimg.cn/img_convert/adb0ac98913f268384058587affc9f40.png)
![img](https://img-blog.csdnimg.cn/img_convert/a836410fb7cd54a42c91fe46691672ca.png)
![img](https://img-blog.csdnimg.cn/img_convert/d9a459624b237bd1595462960b56eefb.png)
![img](https://img-blog.csdnimg.cn/img_convert/538b858bdb4198af8162eb01a4350754.png)
![img](https://img-blog.csdnimg.cn/img_convert/47204f4740621aef1195718eca11a290.png)

**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！**

**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**

**如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）**
![img](https://img-blog.csdnimg.cn/img_convert/40a94a8dfc989b20878405ac8061293d.png)

**一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**

3027504513)]

**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！**

**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**

**如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）**
[外链图片转存中...(img-mz6hKZFB-1713027504513)]

**一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/560290
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号