Linux系统安装RabbitMQ（Centos 7）

叁歲伎倆 2022-04-10 12:47 448阅读 0赞

### 一. Linux下RabbitMQ环境搭建 ###

**1.首先装erlang语言**

Yum安装socat

# yum -y install socat

下载erlang软件包，本文使用erlang-19.0.4版本，下面给出下载链接

# wget http://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el7.centos.x86_64.rpm

安装erlang

# rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm

安装完成后执行erl命令，出现下图则代表成功

![2019010712342977.png][]

**2. 安装RabbitMQ**

下载rabbitmq软件包

#  wget  http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.10/rabbitmq-server-3.6.10-1.el7.noarch.rpm

** ** 安装rabbitmq

# rpm -ivh rabbitmq-server-3.6.10-1.el7.noarch.rpm

**  **启动命令

# systemctl start rabbitmq-server

查看rabbitmq 启动后的情况

# rabbitmqctl status

**3. 配置网页插件**

首先创建目录，否则可能报错：

# mkdir /etc/rabbitmq

然后启用插件：

# rabbitmq-plugins enable rabbitmq_management

**4.配置防火墙**

配置linux 端口 15672 网页管理 5672 AMQP端口：

# firewall-cmd --permanent --add-port=15672/tcp
    # firewall-cmd --permanent --add-port=5672/tcp
    # systemctl restart firewalld.service

**5.配置web端访问账号密码和权限**

默认网页是不允许访问的，需要增加一个用户修改一下权限，代码如下：

添加用户：（后面两个参数分别是用户名和密码）

# rabbitmqctl add_user myrabbitmq rabbitmq123

添加权限：

# rabbitmqctl set_permissions -p / myrabbitmq ".*" ".*" ".*"

修改用户角色：

# rabbitmqctl set_user_tags myrabbitmq administrator

**6. ECS云服务器添加安全组规则**

![20190107124414367.png][]

**7. 访问web端管理界面**

在浏览器输入服务器ip:15672，即可看到RabbitMQ的WEB管理页面，账号密码分别为myrabbitmq，rabbitmq123

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70][]  
**常用命令：**

**\# systemctl stop firewalld.service   关闭防火墙**

**\# service rabbitmq-server start       启动RabbitMQ**

### 二. SpringBoot使用死信队列 ###

在 RabbitMQ 3.6.x 之前我们一般采用死信队列+TTL过期时间来实现延迟队列。

在 RabbitMQ 3.6.x 开始，RabbitMQ 官方提供了延迟队列的插件。

**1. 安装延迟插件（用于死信队列）**

SpringBoot项目想要使用死信队列，首先先安装延时插件：

插件下载地址：[https://www.rabbitmq.com/community-plugins.html][https_www.rabbitmq.com_community-plugins.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 1][]

找到与rabbitmq对应的版本下载即可。

① rpm安装的rabbitmq，方法如下：

解压下载好的插件，拷贝到linux服务器如下目录：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 2][]

重启rabbitmq，执行如下命令：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

② 如果是zip安装的rabbitmq，方法如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 3][]

**2. SpringBoot使用死信队列**

**场景分析：**一些电商网站或者外卖APP，在我们提交订单后，24小时或者半小时未支付，则该笔订单作废，即会变为失效状态，如果用传统的定时任务，订单数量上去了，轻则性能非常非常低，重则服务器宕机，为了解决该场景，可以采用RabbitMQ的死信队列（也叫延时队列）。

**准备工作**：maven依赖，application.yml配置，实体类，数据传输对象：

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

spring:
      datasource:
        driver-class-name: com.mysql.jdbc.Driver
        url: jdbc:mysql://localhost:3306/rabbitmq-plugin?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC
        username: root
        password: 123456
    
      rabbitmq:
        publisher-confirms: true
        publisher-returns: true
        template:
          mandatory: true
        listener:
          simple:
            # 消费者监听并发数
            concurrency: 10
            # 最大并发数
            max-concurrency: 20
            # 签收模式，手动    
            acknowledge-mode: manual
            # 限流（预拉取量5条）
            prefetch: 5
        host: 192.168.10.10
        port: 5672
        username: myrabbit
        password: 123456
        connection-timeout: 15000
        virtual-host: /

package com.example.entity;
    import java.io.Serializable;
    import java.util.Date;
    @Data
    public class OrderEntity implements Serializable{
        private Integer id;
        private String orderNo;
        private Integer userId;
        private Integer status;
        private Date createTime;
        private Date updateTime;
    }

package com.example.entity;
    import java.io.Serializable;
    @Data
    public class OrderDto implements Serializable{
        private String orderNo;
        private Integer userId;
    }

数据表结构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 4][]

**接下来进行rabbitmq的具体配置：**

① 首先建立RabbitMq的配置文件

声明处理消息的队列，通过routingkey绑定交换机，注意交换机为CustomExchange。

package com.example.rabbitmq;
    
    import java.util.HashMap;
    import java.util.Map;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.QueueBuilder;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMqConfig {
    
        public static final String DELAY_EXCHANGE_NAME = "plugin-dead-exchange";
    	
        public static final String DELAY_QUEUE_NAME = "plugin-dead-queue";
    	
        public static final String ROUTING_KRY = "plugin-routing-key";
    	
        /**
         * 声明一个死信队列
         * @return
         */
        @Bean
        Queue delayQueue(){
            return QueueBuilder.durable(DELAY_QUEUE_NAME)
                .build();
        }
        /**
         * 声明一个交换机
         * @return
         */
        @Bean
        CustomExchange delayExchange(){
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct"); // 直连模式
            return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true,false, args);
        }
        /**
        * 绑定
        * @param delayQueue
        * @param delayExchange
        * @return
        */
        @Bean
        Binding queueBinding(Queue delayQueue, CustomExchange delayExchange){
            return BindingBuilder.bind(delayQueue)
                .to(delayExchange)
                .with(ROUTING_KRY)
                .noargs();
        }
    
    }

② 编写生产者模板

package com.example.rabbitmq;
    
    import com.example.entity.OrderEntity;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Producer {
    
        private static final Logger log = LoggerFactory.getLogger(Producer.class);
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    	
        /**
         * 监听函数：confirm确认消息投递成功
         */
        /*
        final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                String messageId = correlationData.getId();
                if (ack) {
                    log.info("消息投递成功，{}",messageId);
                    //进行消息记录的数据库更新	
                }else{
                    log.info("消息投递失败");
                }	
            }
        };
    	*/
        /**
         * 发送消息
         * @param msg 消息
         * @param expiration 延迟时间
         */
        public void sendMessage(OrderEntity msg, Long expiration){
            // 绑定异步监听回调函数
            // rabbitTemplate.setConfirmCallback(this.confirmCallback);
    
          rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE_NAME,RabbitMqConfig.ROUTING_KRY, msg,(message)->{
                // MessageProperties properties=message.getMessageProperties();
                // properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);//持久化
                // properties..setHeader("x-delay", expiration); // 设置延迟时间
                message.getMessageProperties().setHeader("x-delay", expiration); // 设置延迟时间
                return message;
            },new CorrelationData(msg.getOrderNo()));
        }
    
    }

③ 编写Dao层

package com.example.dao;
    
    import com.example.entity.OrderEntity;
    import com.example.provider.OrderProvider;
    import org.apache.ibatis.annotations.*;
    
    @Mapper
    public interface OrderDao {
    
        /**
         * 提交订单
         * @param orderEntity
         * @return
         */
        @InsertProvider(type= OrderProvider.class,method="insertOrder")
        @Options(useGeneratedKeys=true)
        Integer insertOrder(OrderEntity orderEntity);
    
        /**
         * 根据订单号查询订单
         * @param orderNo
         * @return
         */
        @Select("select * from user_order where order_no=#{orderNo} and status=1")
        OrderEntity getByOrderNo(@Param("orderNo")String orderNo);
    
        /**
         * 修改订单状态为已失效
         * @param orderNo
         * @return
         */
        @Update("update user_order set status=4 where order_no=#{orderNo}")
        int updateStatusToLoseEfficacy(@Param("orderNo")String orderNo);
    
    }

package com.example.provider;
    
    import com.example.entity.OrderEntity;
    import org.apache.ibatis.jdbc.SQL;
    
    public class OrderProvider {
    
        public String insertOrder(OrderEntity orderEntity){
            return new SQL(){
       {
                INSERT_INTO("user_order");
                if(orderEntity.getId() != null){
                    VALUES("id","#{id}");
                }
                if(orderEntity.getOrderNo() != null){
                    VALUES("order_no","#{orderNo}");
                }
                if(orderEntity.getUserId() != null){
                    VALUES("user_id","#{userId}");
                }
                if(orderEntity.getStatus() != null){
                    VALUES("status","#{status}");
                }
                if(orderEntity.getCreateTime() != null){
                    VALUES("create_time","#{createTime}");
                }
                if(orderEntity.getUpdateTime() != null){
                    VALUES("update_time","#{updateTime}");
                }
            }}.toString();
        }
    
    }

④ 编写提交订单接口（**生产者**）

package com.example.controller;
    
    import com.example.dao.OrderDao;
    import com.example.entity.OrderDto;
    import com.example.entity.OrderEntity;
    import com.example.rabbitmq.Producer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.BeanUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestBody;
    import org.springframework.web.bind.annotation.RestController;
    import org.yuyi.full.handler.exception.ResultBO;
    import org.yuyi.full.handler.exception.ResultTool;
    import java.util.Date;
    
    
    @RestController
    public class OrderController {
    
        private static final Logger log = LoggerFactory.getLogger(OrderController.class);
    
        @Autowired
        private OrderDao orderDao;
        @Autowired
        private Producer producer;
    
        private static final Long EXPIRE_TIME = 60000L; // 1分钟
    
        /**
         * 提交订单
         */
        @PostMapping("/submitOrder")
        public ResultBO<?> submitOrder(@RequestBody OrderDto orderDto){
            // 数据库操作
            OrderEntity orderEntity = new OrderEntity();
            BeanUtils.copyProperties(orderDto,orderEntity);
            orderEntity.setStatus(1);
            orderEntity.setCreateTime(new Date());
            orderDao.insertOrder(orderEntity);
    
            // RabbitMQ操作
            producer.sendMessage(orderEntity, EXPIRE_TIME);
            log.info("【RabbitMQ发送消息成功】：{}",orderEntity);
    
            return ResultTool.success(orderEntity);
        }
    
    }

编写RabbitMQ监听器（**消费者**）

package com.example.rabbitmq;
    
    import java.time.LocalDateTime;
    import java.util.Map;
    import java.util.Objects;
    import com.example.dao.OrderDao;
    import com.example.entity.OrderEntity;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    import com.rabbitmq.client.Channel;
    
    @Component
    public class Consumer {
    
        private static final Logger log = LoggerFactory.getLogger(Consumer.class);
    
        @Autowired
        private OrderDao orderDao;
    
        @RabbitListener(queues = "plugin-dead-queue")
        @RabbitHandler
        public void consumer(@Payload OrderEntity msg, @Headers Map<String, Object> headers, Channel channel) throws Exception {
            log.info("【RabbitMQ监听到消息】：{}",msg);
            String orderNo = msg.getOrderNo();
    
            // 根据订单号查询订单，如果订单状态还是未付款，则修改为已失效
            OrderEntity orderEntity = orderDao.getByOrderNo(orderNo);
            if(Objects.nonNull(orderEntity)){
                // 24小时未付款，设为失效状态
                orderDao.updateStatusToLoseEfficacy(orderNo);
            }
    
            // ack手工签收，通知RabbitMQ，消费端消费成功
            Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
            channel.basicAck(deliveryTag,false);
        }
    
    }

此时，代码编写完毕！下面用postman测试一下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 5][]  ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 6][]

![20191009143306244.png][]

可以看到提交订单后数据库会生成一条记录，默认状态为1（待付款），IDEA控制台也会打印发送者日志：

![20191009143459998.png][]

24小时太长，本人设置的延时时间为1分钟，所以耐心等待1分钟后，再观察数据库和控制台如下：

![20191009143622900.png][]

![2019100914365727.png][]

可以发现，在没付款的情况下，死信队列会修改订单状态为1（已失效）。

通过死信队列解决订单超时失效场景，异步操作，性能非常高，这是互联网电商项目必不可少的技术点 ~

[2019010712342977.png]: /images/20220330/f2113ef0186b4f11a7e7bb893bb5d7bb.png
[20190107124414367.png]: /images/20220330/708d838b74994e18aa0f73c1e5d80d39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70]: /images/20220330/0dd45c2bd90d4ef3a2bb87d7ef5dd4e6.png
[https_www.rabbitmq.com_community-plugins.html]: https://www.rabbitmq.com/community-plugins.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 1]: /images/20220330/97d64af706044756ab47b75e537371c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 2]: /images/20220330/20b79c0e6c824e5787eb6ec879a9ff51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 3]: /images/20220330/49dea21b0e4141e684c18c901d2117fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 4]: /images/20220330/4189854607d845649e8fea89bbed20e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20191009143156936.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FraXJhTmlja3k_size_16_color_FFFFFF_t_70 6]: /images/20220330/97279dc842094f5388725c3de9082e37.png
[20191009143306244.png]: /images/20220330/7e8f0acf3f0e4f33bd97369adbbcf473.png
[20191009143459998.png]: /images/20220330/f23bb313501743b2af9ba43d10fe8ca5.png
[20191009143622900.png]: /images/20220330/65f39242993843ecbb9d1111736f912b.png
[2019100914365727.png]: /images/20220330/e44e73daabd84359a103e9891911bab9.png