当前位置：首页 > news >正文

Centos安装RabbitMQ，JavaSpring发送RabbitMQ延迟延时消息，JavaSpring消费RabbitMQ消息

news 2025/9/6 8:15:53

1，版本说明

erlang 和 rabbitmq 版本说明
https://www.rabbitmq.com/which-erlang.html
确认需要安装的mq版本以及对应的erlang版本。

2，下载安装文件

RabbitMQ下载地址：
https://packagecloud.io/rabbitmq/rabbitmq-server

Erlang下载地址：
https://packagecloud.io/rabbitmq/erlang

RabbitMQ延迟消息插件下载
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

下载文件如图

在这里插入图片描述

3，安装步骤

3.1，查询是否有安装过erlang、rabbitmq，查询到有的话需要删除。

	rpm -qa | grep rabbitmq-serverrpm -qa | grep erlang# 删除yum -y remove rabbitmq-server.noarch

3.2，本地安装erlang

	yum localinstall erlang-23.2.7-2.el7.x86_64.rpm# 查询安装的版本erl -version# Erlang (SMP,ASYNC_THREADS,HIPE) (BEAM) emulator version xxx

3.3，本地安装rabbitmq

	yum localinstall rabbitmq-server-3.9.0-1.el7.noarch.rpm

	# 启动rabbitmqsystemctl start rabbitmq-server# 查看rabbitmq状态systemctl status rabbitmq-server# 设置rabbitmq服务开机自启动systemctl enable rabbitmq-server# 关闭rabbitmq服务systemctl stop rabbitmq-server# 重启rabbitmq服务systemctl restart rabbitmq-server

3.4, mq 端口开放：

	firewall-cmd --zone=public --add-port=5672/tcp --permanentfirewall-cmd --zone=public --add-port=15672/tcp --permanentfirewall-cmd --reloadfirewall-cmd --zone=public --list-ports

3.5, 安装mq管理界面

	# 启用管理界面插件rabbitmq-plugins enable rabbitmq_managementcurl http://localhost:15672 就可以打开web管理页面# rabbitmq有一个默认的账号密码guest，但该情况仅限于本机localhost进行访问，所以需要添加一个远程登录的用户# 添加用户rabbitmqctl add_user 用户名 密码rabbitmqctl add_user admin 123456# 设置用户角色,分配操作权限rabbitmqctl set_user_tags 用户名 角色rabbitmqctl set_user_tags admin administrator# 为用户添加资源权限(授予访问虚拟机根节点的所有权限)rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"# 角色有四种：# administrator：可以登录控制台、查看所有信息、并对rabbitmq进行管理# monToring：监控者；登录控制台，查看所有信息# policymaker：策略制定者；登录控制台指定策略# managment：普通管理员；登录控制# 修改密码rabbitmqctl change_ password 用户名 新密码# 删除用户rabbitmqctl delete_user 用户名# 查看用户清单rabbitmqctl list_users

3.6, 延迟消息插件安装：

    # 把插件包先复制到	 /usr/lib/rabbitmq/lib/rabbitmq_server-3.9.0/pluginscp rabbitmq_delayed_message_exchange-3.9.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.9.0/plugins/rabbitmq-plugins enable rabbitmq_delayed_message_exchange#重启mq		systemctl restart rabbitmq-serverrabbitmq-plugins list

3.7，登录测试

访问地址： ip:15672 账号密码： admin 123456
登录界面

找到交换机 exchange，看看类型是否有延迟消息类型的
在这里插入图片描述

然后就可以写代码去连接发消息了。

4， Java代码

4.1, pom 引入：

         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

4.2, 配置类：

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {@Beanpublic RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);rabbitAdmin.setAutoStartup(true);return rabbitAdmin;}}

4.3, 消息定义配置类：


import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class OrderRabbitMQConfig {@Autowiredprivate RabbitAdmin rabbitAdmin;//================================订单延时=================================@BeanCustomExchange order_pay_delay_exchange() {HashMap<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange("order_pay_delay_exchange", "x-delayed-message", true, false, args);}@Beanpublic Queue order_pay_delay_queue() {Queue queue = new Queue("order_pay_delay_queue", true, false, false);rabbitAdmin.declareQueue(queue);return queue;}@Beanpublic Binding order_pay_delay_binding() {return BindingBuilder.bind(order_pay_delay_queue()).to(order_pay_delay_exchange()).with("order_pay_delay_routing").noargs();}//================================订单支付通知======================================@Beanpublic DirectExchange order_pay_notify_exchange() {return new DirectExchange("order_pay_notify_exchange", true, false);}@Beanpublic Queue order_pay_notify_direct_queue() {Map<String, Object> argsMap = new HashMap<>();argsMap.put("x-max-priority", 5);Queue queue = new Queue("order_pay_notify_queue", true, false, false, argsMap);rabbitAdmin.declareQueue(queue);return queue;}@Beanpublic Binding ctc_bidding_auction_pay_notify_binding() {return BindingBuilder.bind(order_pay_notify_direct_queue()).to(order_pay_notify_exchange()).with("order_pay_notify_routing");}
}

4.4, 消息发送类：


import cn.hutool.json.JSONUtil;
import com.xxx.rabbitmq.dto.PayOrderNotifyDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class RabbitMQSendUtils {private static RabbitTemplate rabbitTemplate;@Autowiredpublic RabbitMQSendUtils(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}/*** 订单支付延时通知、发送MQ消息*/public static void sendPayDelayMessage(PayOrderNotifyDto dto, final Integer delayTimes) {//给延迟队列发送消息String msg = JSONUtil.toJsonStr(dto);log.info("订单支付延时通知、发送MQ消息: {}, delayTimes={}", msg, delayTimes);rabbitTemplate.convertAndSend("order_pay_delay_exchange", "order_pay_delay_routing", msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//给消息设置延迟毫秒值message.getMessageProperties().setDelay(delayTimes);return message;}});}/*** 订单支付通知，发送MQ消息*/public static void sendPayNotifyMsg(PayOrderNotifyDto dto) {log.info("订单支付通知，发送MQ消息: {}", dto);rabbitTemplate.convertAndSend("order_pay_notify_exchange", "order_pay_notify_routing", JSONUtil.toJsonStr(dto));}
}

4.5, 消息监听消费类：


import cn.hutool.json.JSONUtil;
import com.xxx.rabbitmq.dto.PayOrderNotifyDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** MQ消费监听*/
@Slf4j
@Component
public class OrderMQListener {/*** 订单延时通知 消息*/@RabbitListener(queues = {"order_pay_delay_queue"})public void payDelayNotify(Message message) {try {String msg = new String(message.getBody());log.info("【消费】订单延时通知 MQ 消息内容: {}, Message={}", msg, message);//支付订单改成超时未支付》取消PayOrderNotifyDto dto = JSONUtil.toBean(msg, PayOrderNotifyDto.class);} catch (Exception e) {log.error("订单延时通知 消息消费失败：", e);}}/*** 订单支付通知 消息*/@RabbitListener(queues = {"order_pay_notify_queue"})public void payNotify(Message message) {try {String msg = new String(message.getBody());log.info("订单支付通知 MQ 消息内容:{}, {}", msg, message);PayOrderNotifyDto payOrderNotifyDto = JSONUtil.toBean(msg, PayOrderNotifyDto.class);} catch (Exception e) {log.error("订单支付通知 消息消费失败：", e);}}}