首页分享消息队列：第五章：RabbitMQ的使用

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

来源：萌宠菠菠乐园时间：2025-04-01 04:55

2023-01-13 201 发布于黑龙江

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介：消息队列：第五章：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) { } } }

相关实践学习

消息队列RocketMQ版：基础消息收发功能体验

本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。

消息队列 MNS 入门课程

1、消息队列MNS简介本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client

解决方案 | 云消息队列RabbitMQ实践

在实际业务中，网站因消息堆积和高流量脉冲导致系统故障。为解决这些问题，云消息队列 RabbitMQ 版提供高性能的消息处理和海量消息堆积能力，确保系统在流量高峰时仍能稳定运行。迁移前需进行技术能力和成本效益评估，包括功能、性能、限制值及费用等方面。迁移步骤包括元数据迁移、创建用户、网络打通和数据迁移。

云消息队列RabbitMQ实践 - 评测

根据反馈，对本解决方案的实践原理已有一定理解，描述整体清晰但需在消息队列配置与使用上增加更多示例和说明以助理解。部署体验中获得了一定的引导和文档支持，尽管文档仍有待完善；期间出现的配置文件错误及依赖库缺失等问题已通过查阅资料解决。设计验证展示了云消息队列RabbitMQ的核心优势，包括高可用性和灵活性，未来可通过增加自动化测试来提高系统稳定性。实践后，用户对方案解决问题的能力及适用场景有了明确认识，认为其具有实际生产价值，不过仍需在性能优化、安全性增强及监控功能上进行改进以适应高并发和大数据量环境。

消息队列 MQ操作报错合集之出现"Connection reset by peer"的错误，该如何处理

消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。

消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump，该怎么办