前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >消息队列:第五章:RabbitMQ的使用

消息队列:第五章:RabbitMQ的使用

作者头像
Java廖志伟
发布2022-09-28 14:56:14
2130
发布2022-09-28 14:56:14
举报
文章被收录于专栏:高级开发进阶高级开发进阶

第一步:使用之前先安装好RabbitMQ,建议安装在linux系统下

安装配置RabbitMQ:https://blog.csdn.net/qq_33450681/article/details/85339315

第二步:在配置文件下配置

代码语言:javascript
复制
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/#/

第三步:业务中使用发送消息

代码语言:javascript
复制
    @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枚举

代码语言:javascript
复制
    /**
     * 支付通知队列
     */
    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.队列名称:可以为不同的消费者指定不同的队列,可以对消息进行分类到不同的队列进行转发。

配置类

代码语言:javascript
复制
/**
 * 消息队列配置
 * 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());
    }

第六步:处理支付信息

代码语言:javascript
复制
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&amp;&amp;(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) {

        }
    }
}
本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2021-04-28,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 第一步:使用之前先安装好RabbitMQ,建议安装在linux系统下
  • 第二步:在配置文件下配置
  • 第三步:业务中使用发送消息
  • 第四步:定义QueueEnum枚举
  • 第五步:配置
  • 配置类
  • 第六步:处理支付信息
相关产品与服务
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档