个人中心
退出
消息队列：第五章：RabbitMQ的使用
第一步：使用之前先安装好RabbitMQ，建议安装在linux系统下
安装配置RabbitMQ：https://blog.csdn.net/qq_33450681/article/details/85339315
第二步：在配置文件下配置
rabbitmq:
host: 192.168.0.100
port: 5672
virtual-host: /mall
username: mall
password: mall
publisher-confirms: true #如果对异步消息需要回调必须设置为true
浏览器访问http://192.168.0.100:15672/#/
第三步：业务中使用发送消息
@Autowired
private OmsOrderSettingMapper orderSettingMapper;
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 发送检查支付结果的消息队列
* @param orderSn
* @param count
*/
@Override
public void sendDelayPaymentCheck(String orderSn, int count) {
//获取订单超时时间
OmsOrderSetting orderSetting = orderSettingMapper.selectByPrimaryKey(1L);
long delayTimes = orderSetting.getNormalOrderOvertime() * 60 * 1000;
//将需要发送的数据封装到hashmap中
HashMap<Object, Object> hashMap = new HashMap<>();
hashMap.put("out_trade_no",orderSn);
hashMap.put("count",count);
//给延迟队列发送消息
amqpTemplate.convertAndSend(QueueEnum.QUEUE_PAY_CANCEL.getExchange(),
QueueEnum.QUEUE_PAY_CANCEL.getRouteKey(), hashMap, new
MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//给消息设置延迟毫秒值
message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
return message;
}
});
}
第四步：定义QueueEnum枚举
/**
* 支付通知队列
*/
QUEUE_PAY_CANCEL("mall.pay.direct","mall.pay.cancel","mall.pay.cancel")
/**
* 交换名称
*/
private String exchange;
/**
* 队列名称
*/
private String name;
/**
* 路由键
*/
private String routeKey;
QueueEnum(String exchange, String name, String routeKey) {
this.exchange = exchange;
this.name = name;
this.routeKey = routeKey;
}
public String getExchange() {
return exchange;
}
public String getName() {
return name;
}
public String getRouteKey() {
return routeKey;
}
第五步：配置
RabbitMQ参数配置：
使用一个RabbitMQ需要配置以下几个重要的参数
1.虚拟主机名称（Virtual host name），这个参数不是真正的IP地址或者域名，它是RabbitMQ内部的一个虚拟主机，就像是电脑安装了N台虚拟机，对外的名称一般是“/xxxx".
2.交换机名（Exchanges name）：顾名思义，就是把生产者送来的消息来进行分发给下游的多个消费者，相当一个内部软交换机。交换机的类型有fanout，direct,topic,header,fanout类型类似以太网交换机的广播模式，把送来的消息给每个下游队列。direct类似单播(使用routingkey来指定目的队列)，topic交换机类似组播，把消息传递给下面同一主题的队列，header交换机则忽略掉routingkey，使用hash数据结构来进行匹配和转发。
3.routingkey ：前面讲过了，交换机在进行消息转发时候，要使用routingkey为关键字进行转发。
4.队列名称：可以为不同的消费者指定不同的队列，可以对消息进行分类到不同的队列进行转发。
配置类
/**
* 消息队列配置
* Created by macro on 2018/9/14.
*/
@Configuration
public class RabbitMqConfig {
/**
* 支付队列
* @return
*/
@Bean
public Queue payQueue() {
return new Queue(QueueEnum.QUEUE_PAY_CANCEL.getName());
}
/**
* 绑定支付交互机
* @return
*/
@Bean
DirectExchange payDirect() {
return (DirectExchange) ExchangeBuilder
.directExchange(QueueEnum.QUEUE_PAY_CANCEL.getExchange())
.durable(true)
.build();
}
/**
* 将支付队列绑定到支付交互机
* @param payDirect
* @param payQueue
* @return
*/
@Bean
Binding payBinding(DirectExchange payDirect,Queue payQueue){
return BindingBuilder
.bind(payQueue)
.to(payDirect)
.with(QueueEnum.QUEUE_PAY_CANCEL.getRouteKey());
}
第六步：处理支付信息
package com.macro.mall.portal.component;
import com.macro.mall.model.PaymentInfo;
import com.macro.mall.portal.service.PaymentService;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.HashMap;
/**
* 支付的处理者
*/
@Component
@RabbitListener(queues = "mall.pay.cancel")
public class PayReceiver {
@Autowired
PaymentService paymentService;
@RabbitHandler
public void handle(HashMap mapMessage){
try {
String outTradeNo = mapMessage.get("out_trade_no").toString();
int count = Integer.parseInt(mapMessage.get("count").toString());
// 如果没有支付成功，再次发送延迟检查队列
if (count > 0) {
// 进行支付状态检查
System.out.println("正在进行第" + (6 - count) + "支付结果次检查");
//调用alipayClient接口，根据out_trade_no查询支付信息
PaymentInfo paymentInfo = paymentService.checkPaymentResult(outTradeNo);
Thread thread = new Thread();
thread.start();
Thread.sleep(10000);
//判断是否已经支付成功
if
(paymentInfo.getPaymentStatus()!=null&&(paymentInfo.getPaymentStatus().equals("TRADE_SUCCESS")
|| paymentInfo.getPaymentStatus().equals("TRADE_FINISHED"))) {
// 交易成功或者失败，记录交易状态
System.out.println("检查交易结果成功，记录交易状态。。。");// 修改支付的状态信息
// 修改支付信息
boolean b = paymentService.checkPaymentStatus(outTradeNo);
if(!b){
//修改为已支付
paymentService.updatePayment(paymentInfo.getCallbackContent(),outTradeNo,paymentInfo.getAlipayTradeNo());
// 发送系统消息，出发并发商品支付业务消息队列
paymentService.sendPaymentSuccess(paymentInfo.getOutTradeNo(),paymentInfo.getPaymentStatus(),paymentInfo.getAlipayTradeNo());
}
} else {//未支付
// 再次进行延迟检查
System.out.println("正在进行第" + (6 - count) + "支付结果次检查，检查用户尚未付款成功，继续巡检");
paymentService.sendDelayPaymentCheck(outTradeNo, count - 1);
}
} else {
System.out.println("支付结果次检查次数耗尽，支付未果。。。");
}
} catch (Exception e) {
}
}
}
分类导航