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

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-server
	rpm -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

	# 启动rabbitmq
	systemctl 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 --permanent
	firewall-cmd --zone=public --add-port=15672/tcp --permanent
	firewall-cmd --reload
	firewall-cmd --zone=public --list-ports

3.5, 安装mq管理界面

	
	# 启用管理界面插件
	rabbitmq-plugins enable rabbitmq_management

	curl 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/plugins
    cp 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-server
	rabbitmq-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 {

    @Bean
    public 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 {

    @Autowired
    private RabbitAdmin rabbitAdmin;
    //================================订单延时=================================
    @Bean
    CustomExchange 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);
    }
    @Bean
    public Queue order_pay_delay_queue() {
        Queue queue = new Queue("order_pay_delay_queue", true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }
    @Bean
    public Binding order_pay_delay_binding() {
        return BindingBuilder.bind(order_pay_delay_queue())
                .to(order_pay_delay_exchange()).with("order_pay_delay_routing").noargs();
    }

    //================================订单支付通知======================================
    @Bean
    public DirectExchange order_pay_notify_exchange() {
        return new DirectExchange("order_pay_notify_exchange", true, false);
    }
    @Bean
    public 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;
    }
    @Bean
    public 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;

    @Autowired
    public 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() {
            @Override
            public 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);
        }
    }

}