淘宝开店铺网站怎么做,wordpress yoast seo,公司备案查询网站,iphone app wordpress消息队列发送支付通知消息
需求分析
订单服务作为通用服务#xff0c;在订单支付成功后需要将支付结果异步通知给其他对接的微服务#xff0c;微服务收到支付结果根据订单的类型去更新自己的业务数据 技术方案
使用消息队列进行异步通知需要保证消息的可靠性即生产端将消息…消息队列发送支付通知消息
需求分析
订单服务作为通用服务在订单支付成功后需要将支付结果异步通知给其他对接的微服务微服务收到支付结果根据订单的类型去更新自己的业务数据 技术方案
使用消息队列进行异步通知需要保证消息的可靠性即生产端将消息成功通知到服务端 消息发送到交换机 -- 由交换机发送到队列 -- 消费者监听队列收到消息进行处理参考文章02- 使用Docker安装RabbitMQ-CSDN博客 生产者确认机制: 发送消息前使用数据库事务将消息保证到数据库表中成功发送到交换机将消息从数据库中删除 配置MQ持久化(交换机、队列、发送消息)MQ收到消息持久化当MQ重启时即使消息没有消费完也不会丢失 消费者确认机制: 消费者消费成功自动发送ACK负责重试消费
发布订阅模式: 订单服务接收支付成功结果通知后创建一条消息发送给Fanout广播类型的交换机,学习中心服务绑定队列到交换机接收消息参考文章04- 基于SpringAMQP封装RabbitMQ,消息队列的Work模型和发布订阅模型-CSDN博客
环境搭建
第一步 在订单服务和学习中心服务中添加消息队列依赖
!--AMQP依赖包含RabbitMQ--
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency第二步在Nacos的dev环境下添加RabbitMQ的配置信息rabbitmq-dev.yaml设置group为xuecheng-plus-common
spring:rabbitmq:host: 192.168.101.128 # 主机port: 5672 # 端口名username: root # 用户名password: root # 密码virtual-host: / # 虚拟主机publisher-confirm-type: correlated # 异步回调定义ConfirmCallbackMQ返回结果时会回调这个ConfirmCallbackpublisher-returns: true # 开启publish-return功能同样是基于callback机制调用回调函数ReturnCallbacktemplate:mandatory: true # 定义消息路由失败时的策略true表示调用ReturnCallbackfalse表示直接丢弃消息listener:simple:# 每次只能获取一条消息处理完成才能获取下一个消息prefetch: 1 # auto:出现异常时返回unack且消息回滚到mq,如果没有异常直接返回ack# manual:手动控制# none:丢弃消息不回滚到mqacknowledge-mode: auto retry:enabled: false # 开启消费者失败重试initial-interval: 5000ms # 初始的失败等待时长为几秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态如果业务中包含事务需要改为false第三步在订单服务和学习中心服务的接口工程中引入rabbitmq-dev.yaml配置文件
- data-id: rabbitmq-${spring.profiles.active}.yamlgroup: xuecheng-plus-commonrefresh: true第四步: 在订单服务的service工程编写MQ配置类PayNotifyConfig创建交换机和队列
Slf4j
Configuration
public class PayNotifyConfig implements ApplicationContextAware {// 交换机public static final String PAYNOTIFY_EXCHANGE_FANOUT paynotify_exchange_fanout;// 支付结果通知消息类型public static final String MESSAGE_TYPE payresult_notify;// 支付通知队列public static final String PAYNOTIFY_QUEUE paynotify_queue;// 声明交换机且持久化Bean(PAYNOTIFY_EXCHANGE_FANOUT)public FanoutExchange paynotify_exchange_fanout() {// 三个参数交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new FanoutExchange(PAYNOTIFY_EXCHANGE_FANOUT, true, false);}//支付通知队列且持久化Bean(PAYNOTIFY_QUEUE)public Queue course_publish_queue() {return QueueBuilder.durable(PAYNOTIFY_QUEUE).build();}// 交换机和支付通知队列绑定Beanpublic Binding binding_course_publish_queue(Qualifier(PAYNOTIFY_QUEUE) Queue queue, Qualifier(PAYNOTIFY_EXCHANGE_FANOUT) FanoutExchange exchange) {return BindingBuilder.bind(queue).to(exchange);}// 交换机路由消息到队列的时候如果失败执行回调函数Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplateRabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 消息处理serviceMqMessageService mqMessageService applicationContext.getBean(MqMessageService.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - {// 消息发送失败记录日志log.info(消息发送失败应答码{}原因{}交换机{}路由键{},消息{},replyCode, replyText, exchange, routingKey, message.toString());// 解析消息内容将消息再添加到消息表MqMessage mqMessage JSON.parseObject(message.toString(), MqMessage.class);mqMessageService.addMessage(mqMessage.getMessageType(), mqMessage.getBusinessKey1(), mqMessage.getBusinessKey2(), mqMessage.getBusinessKey3());});}
}第五步: 在学习中心服务编写MQ配置类PayNotifyConfig创建交换机和队列避免学习中心服务启动的时候监听的队列还没有创建如果生产端已经创建就不再创建
Slf4j
Configuration
public class PayNotifyConfig implements ApplicationContextAware {// 声明交换机,支付通知队列,交换机和支付通知队列绑定关系// 不用设置回调函数只有生产者才需要确认
}重启订单服务登录rabbitmq查看交换机自动创建成功 生产者发送信息
在订单服务的OrderService中定义接口接收支付宝响应的通知消息结果并发送给学习中心服务
public interface OrderService {/*** 接收通知结果并发送给学习中心服务* param mq Message 消息*/void notifyPayResult(MqMessage mqMessage);
}Slf4j
Service
public class OrderServiceImpl implements OrderService {AutowiredMqMessageService mqMessageService;AutowiredRabbitTemplate rabbitTemplate;Overridepublic void notifyPayResult(MqMessage mqMessage) {// 1. 将消息体转为JsonString jsonMsg JSON.toJSONString(mqMessage);// 2. 设置消息的持久化方式为PERSISTENT即消息会被持久化到磁盘上确保即使在RabbitMQ服务器重启后也能够恢复消息Message msgObj MessageBuilder.withBody(jsonMsg.getBytes()).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();// 3. 封装CorrelationData用于跟踪指定Id消息的相关信息CorrelationData correlationData new CorrelationData(mqMessage.getId().toString());// 3.1 使用CorrelationData添加一个Callback对象指定回调方法该对象用于在消息确认时处理消息的结果correlationData.getFuture().addCallback(result - {if (result.isAck()) {// 3.2 消息成功发送到交换机删除消息表中的记录log.debug(消息发送成功{}, jsonMsg);mqMessageService.completed(mqMessage.getId());} else {// 3.3 消息发送失败log.error(消息发送失败id{}原因{}, mqMessage.getId(), result.getReason());}}, ex - {// 3.4 消息异常可能是网络问题log.error(消息发送异常id{}原因{}, mqMessage.getId(), ex.getMessage());});// 4. 发送消息rabbitTemplate.convertAndSend(PayNotifyConfig.PAYNOTIFY_EXCHANGE_FANOUT, , msgObj, correlationData);}
}订单服务收到第三方平台的支付结果时在saveAliPayStatus方法中除了保存支付宝响应的结果信息还需要向数据库消息表添加消息记录并将消息封装好后发送给消费端 /*** 保存支付结果信息向数据库中的消息表添加消息并发送给消费端* param payStatusDto 支付结果信息*/
Transactional
Override
public void saveAlipayStatus(PayStatusDto payStatusDto) {// 1. 获取支付流水号String payNo payStatusDto.getOut_trade_no();// 2. 查询数据库订单状态XcPayRecord payRecord getPayRecordByPayNo(payNo);if (payRecord null) {XueChengPlusException.cast(未找到支付记录);}XcOrders order xcOrdersMapper.selectById(payRecord.getOrderId());if (order null) {XueChengPlusException.cast(找不到相关联的订单);}String statusFromDB payRecord.getStatus();// 2.1 已支付直接返回if (600002.equals(statusFromDB)) {return;}// 3. 查询支付宝交易状态String tradeStatus payStatusDto.getTrade_status();// 3.1 支付宝交易已成功保存订单表和交易记录表更新交易状态if (TRADE_SUCCESS.equals(tradeStatus)) {// 更新支付交易表payRecord.setStatus(601002);payRecord.setOutPayNo(payStatusDto.getTrade_no());payRecord.setOutPayChannel(Alipay);payRecord.setPaySuccessTime(LocalDateTime.now());int updateRecord xcPayRecordMapper.updateById(payRecord);if (updateRecord 0) {XueChengPlusException.cast(更新支付交易表失败);}// 更新订单表order.setStatus(600002);int updateOrder xcOrdersMapper.updateById(order);if (updateOrder 0) {log.debug(更新订单表失败);XueChengPlusException.cast(更新订单表失败);}}// 4. 创建消息记录并保存到消息表中参数1支付结果类型通知参数2业务id参数3业务类型MqMessage mqMessage mqMessageService.addMessage(payresult_notify, order.getOutBusinessId(), order.getOrderType(), null);// 5. 封装消息记录并发送给消费端notifyPayResult(mqMessage);
}消费者接收消息
在学习中心服务定义impl/ReceivePayNotifyService类
监听消息队列接收支付结果, 当接收到消息后更新选课记录表的选课状态为选课成功,同时向我的课程表中插入一条课程记录
Slf4j
Service
public class ReceivePayNotifyService {AutowiredMyCourseTablesService tablesService;RabbitListener(queues PayNotifyConfig.PAYNOTIFY_QUEUE)public void receive(Message message) {// 1. 获取消息MqMessage mqMessage JSON.parseObject(message.getBody(), MqMessage.class);// 2. 根据消息内容,更新选课记录,向我的课程表插入记录// 2.1 消息类型学习中心只处理支付结果的通知String messageType mqMessage.getMessageType();// 2.2 选课idString chooseCourseId mqMessage.getBusinessKey1();// 2.3 订单类型60201表示购买课程String orderType mqMessage.getBusinessKey2();// 3. 学习中心只负责处理支付结果的通知if (PayNotifyConfig.MESSAGE_TYPE.equals(messageType)){// 3.1 学习中心只负责购买课程类订单的结果if (60201.equals(orderType)){// 3.2 保存选课记录boolean flag tablesService.saveChooseCourseStatus(chooseCourseId);if (!flag){XueChengPlusException.cast(保存选课记录失败);}}}}
}在MyCourseTablesService接口中定义方法更新选课记录的选课状态同时向我的课程表添加选课记录(之前添加免费课程的时候已经实现过了
public interface MyCourseTablesService {/*** 保存选课成功状态* param chooseCourseId* return*/public boolean saveChooseCourseSuccess(String chooseCourseId);
}
Slf4j
Service
public class MyCourseTablesServiceImpl implements MyCourseTablesService {OverrideTransactionalpublic boolean saveChooseCourseStatus(String chooseCourseId) {// 1. 根据选课id查询对应的选课记录XcChooseCourse chooseCourse chooseCourseMapper.selectById(chooseCourseId);if (chooseCourse null) {log.error(接收到购买课程的消息根据选课id未查询到课程选课id{}, chooseCourseId);return false;}// 2. 选课状态为未支付时更新选课状态为选课成功if (701002.equals(chooseCourse.getStatus())) {chooseCourse.setStatus(701001);int update chooseCourseMapper.updateById(chooseCourse);if (update 0) {log.error(更新选课记录失败{}, chooseCourse);}}// 3. 向我的课程表添加记录addCourseTables(chooseCourse);return true;}
}public XcCourseTables addCourseTabls(XcChooseCourse xcChooseCourse){//选课成功了才可以向我的课程表添加String status xcChooseCourse.getStatus();if(!701001.equals(status)){XueChengPlusException.cast(选课没有成功无法添加到课程表);}XcCourseTables xcCourseTables getXcCourseTables(xcChooseCourse.getUserId(), xcChooseCourse.getCourseId());if(xcCourseTables!null){return xcCourseTables;}xcCourseTables new XcCourseTables();BeanUtils.copyProperties(xcChooseCourse,xcCourseTables);xcCourseTables.setChooseCourseId(xcChooseCourse.getId());//记录选课表的逐渐xcCourseTables.setCourseType(xcChooseCourse.getOrderType());//选课类型xcCourseTables.setUpdateDate(LocalDateTime.now());int insert courseTablesMapper.insert(xcCourseTables);if(insert0){XueChengPlusException.cast(添加我的课程表失败);}return xcCourseTables;
}通知支付结果测试
选择一门已发布的收费课程如果在我的课程表存储则删除记录及其相关的选课记录及订单记录信息
进入课程详细页面点击马上学习生成二维码进行支付支付完成点击“支付完成”观察订单服务控制台是否发送消息使用内网穿透工具观察学习中心服务控制台是否接收到消息观察数据库中的消息表的相应记录是否已删除我的选课表中是否有对应的选课记录
