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

编程入门 行业动态 更新时间:2024-10-23 04:39:11

Centos安装RabbitMQ,JavaSpring发送RabbitMQ延迟延时<a href=https://www.elefans.com/category/jswz/34/1771421.html style=消息,JavaSpring消费RabbitMQ消息"/>

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

1,版本说明

erlang 和 rabbitmq 版本说明
.html
确认需要安装的mq版本以及对应的erlang版本。

2,下载安装文件

RabbitMQ下载地址:

Erlang下载地址:

RabbitMQ延迟消息插件下载

下载文件如图

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);}}}

更多推荐

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

本文发布于:2023-12-03 07:09:35,感谢您对本站的认可!
本文链接:https://www.elefans.com/category/jswz/34/1652499.html
版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,我们将在24小时内删除。
本文标签:消息   Centos   RabbitMQ   JavaSpring

发布评论

评论列表 (有 0 条评论)
草根站长

>www.elefans.com

编程频道|电子爱好者 - 技术资讯及电子产品介绍!