RabbitMQ入门与使用

你的名字 2023-02-26 13:27 27阅读 0赞

最近学习了Kafka相关的知识，打算再把RabbitMQ也了解一下，可以有个对比，有时间RocketMQ也去了解下，其实学多了MQ其实是大同小异，我们就进入今天的RabbitMQ的领域去探索一下吧。

有想要在Linux上安装RabbiMQ的可以看下这篇博客，亲测可用，一路畅通无阻:[https://blog.csdn.net/wang1988081309/article/details/97374677][https_blog.csdn.net_wang1988081309_article_details_97374677]

## 基本特性 ##

高可靠:RabbitMQ 提供了多种多样的特性让你在可靠性和性能之间做出权衡，包括持久化、发送应答、发布确认以及高可用性。

灵活的路由:通过交换机(Exchange)实现消息的灵活路由。

支持多客户端:对主流开发语言(Python、Java、Ruby、PHP、C\#、JavaScript、 Go、Elixir、Objective-C、Swift 等)都有客户端实现。

集群与扩展性:多个节点组成一个逻辑的服务器，支持负载。

高可用队列:通过镜像队列实现队列中数据的复制。

权限管理:通过用户与虚拟机实现权限管理。

插件系统:支持各种丰富的插件扩展，同时也支持自定义插件。

与 Spring 集成:Spring 对 AMQP 进行了封装

## AMQP 协议 ##

AMQP:高级消息队列协议，是一个工作于应用层的协议

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70][]

RabbitMQ的工作模型

由于 RabbitMQ 实现了 AMQP 协议，所以 RabbitMQ 的工作模型也是基于 AMQP 的。理解这张图片至关重要。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 1][]

上面的名词讲解

1、Broker

我们要使用 RabbitMQ 来收发消息，必须要安装一个 RabbitMQ 的服务，可以安装 在 Windows 上面也可以安装在 Linux 上面，默认是 5672 的端口。这台 RabbitMQ 的 服务器我们把它叫做 Broker，中文翻译是代理/中介，因为 MQ 服务器帮助我们做的事 情就是存储、转发消息。

2、Connection

无论是生产者发送消息，还是消费者接收消息，都必须要跟 Broker 之间建立一个连接，这个连接是一个 TCP 的长连接。

3、Channel

如果所有的生产者发送消息和消费者接收消息，都直接创建和释放 TCP 长连接的话， 对于 Broker 来说肯定会造成很大的性能损耗，因为 TCP 连接是非常宝贵的资源，创建和 释放也要消耗时间。

所以在 AMQP 里面引入了 Channel 的概念，它是一个虚拟的连接。我们把它翻译 成通道，或者消息信道。这样我们就可以在保持的 TCP 长连接里面去创建和释放 Channel，大大了减少了资源消耗。另外一个需要注意的是，Channel 是 RabbitMQ 原 生 API 里面的最重要的编程接口，也就是说我们定义交换机、队列、绑定关系，发送消 息消费消息，调用的都是 Channel 接口上的方法。

之前有人问过我为啥有Channel，我在网上看到过这样一个解释，挺好的，可以让大家看一看

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 2][]

4、Queue

现在我们已经连到 Broker 了，可以收发消息了。在其他一些 MQ 里面，比如 ActiveMQ 和 Kafka，我们的消息都是发送到队列上的。

队列是真正用来存储消息的，是一个独立运行的进程，有自己的数据库(Mnesia)。

消费者获取消息有两种模式，一种是 Push 模式，只要生产者发到服务器，就马上推 送给消费者。另一种是 Pull 模式，消息存放在服务端，只有消费者主动获取才能拿到消 息。消费者需要写一个 while 循环不断地从队列获取消息吗?不需要，我们可以基于事 件机制，实现消费者对队列的监听。

由于队列有 FIFO 的特性，只有确定前一条消息被消费者接收之后，才会把这条消息 从数据库删除，继续投递下一条消息。

5、Exchange

在 RabbitMQ 里面永远不会出现消息直接发送到队列的情况。因为在 AMQP 里面 引入了交换机(Exchange)的概念，用来实现消息的灵活路由。

交换机是一个绑定列表，用来查找匹配的绑定关系。  
队列使用绑定键(Binding Key)跟交换机建立绑定关系。 生产者发送的消息需要携带路由键(Routing Key)，交换机收到消息时会根据它保存的绑定列表，决定将消息路由到哪些与它绑定的队列上。 注意:交换机与队列、队列与消费者都是多对多的关系。

6、Vhost

我们每个需要实现基于 RabbitMQ 的异步通信的系统，都需要在服务器上创建自己 要用到的交换机、队列和它们的绑定关系。如果某个业务系统不想跟别人混用一个系统， 怎么办?再采购一台硬件服务器单独安装一个 RabbitMQ 服务?这种方式成本太高了。 在同一个硬件服务器上安装多个 RabbitMQ 的服务呢?比如再运行一个 5673 的端口? 没有必要，因为 RabbitMQ 提供了虚拟主机 VHOST。

VHOST 除了可以提高硬件资源的利用率之外，还可以实现资源的隔离和权限的控 制。它的作用类似于编程语言中的 namespace 和 package，不同的 VHOST 中可以有 同名的 Exchange 和 Queue，它们是完全透明的。

这个时候，我们可以为不同的业务系统创建不同的用户(User)，然后给这些用户 分配 VHOST 的权限。比如给风控系统的用户分配风控系统的 VHOST 的权限，这个用户 可以访问里面的交换机和队列。给超级管理员分配所有 VHOST 的权限。

我们说到 RabbitMQ 引入 Exchange 是为了实现消息的灵活路由，到底有哪些路由 方式?

# 可以看我之前写的博客，这里不想重复造轮子 #

[RabbitMQ面试必备][RabbitMQ]

# 使用方式 #

## JavaAPI ##

<dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.6.0</version>
            </dependency>

public class MyProducer {
    
        private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";
    
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            // 设置ip
            factory.setHost("10.206.0.14");
            // 设置端口
            factory.setPort(5672);
            // 用户密码
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 建立连接
            Connection connection = factory.newConnection();
            // 创造消息通道
            Channel channel = connection.createChannel();
            // 发送消息
            // String exchange, String routingKey, BasicProperties props, byte[] body
            channel.basicPublish(EXCHANGE_NAME, "carry.best", null, "hello mq".getBytes());
            //关闭连接
            channel.close();
            connection.close();
        }
    }

package carry.分布式服务治理专题.RabbitMQ.java;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class MyConsumer {
    
        private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";
        private final static String QUEUE_NAME = "SIMPLE_QUEUE";
    
        public static void main(String[] args) throws IOException, TimeoutException {
    
    
            ConnectionFactory factory = new ConnectionFactory();
            // 设置ip
            factory.setHost("10.206.0.14");
            // 设置端口
            factory.setPort(5672);
            // 虚拟机
            factory.setVirtualHost("/");
    
            //设置访问的用户
            factory.setUsername("guest");
            factory.setPassword("guest");
            // 建立连接
            Connection conn = factory.newConnection();
            // 创建消息通道
            Channel channel = conn.createChannel();
            // 声明交换机
            // String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", false, false, null);
    
            //声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //绑定交换机和队列
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"carry.best");
    
            //创建一个消费者
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String msg = new String(body, "UTF-8");
                    System.out.println("Received message : '" + msg + "'");
                    System.out.println("consumerTag : " + consumerTag );
                    System.out.println("deliveryTag : " + envelope.getDeliveryTag() );
                }
            };
            // 开始获取消息
            // String queue, boolean autoAck, Consumer callback
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
        }
    }

当你明白了 RabbitMQ 的 Java 原生 API 编程，可以用它来干什么? (不要离开 Spring 就不知道怎么实现功能了)

参数详解

1)声明交换机的参数  
String type:交换机的类型，direct, topic, fanout 中的一种。  
boolean durable:是否持久化，代表交换机在服务器重启后是否还存在。

2)声明队列的参数

boolean durable:是否持久化，代表队列在服务器重启后是否还存在。

boolean exclusive:是否排他性队列。排他性队列只能在声明它的 Connection 中使用(可以在同一个 Connection 的不同的 channel 中使用)，连接断开时自动删 除。

boolean autoDelete:是否自动删除。如果为 true，至少有一个消费者连接到 这个队列，之后所有与这个队列连接的消费者都断开时，队列会自动删除。

Map<String, Object> arguments:队列的其他属性，例如:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 3][]

3)消息属性 BasicProperties 以下列举了一些主要的参数:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 5][]

RabbitMQ 进阶知识

消息的过期时间

有两种设置方式:  
1) 通过队列属性设置消息过期时间 所有队列中的消息超过时间未被消费时，都会过期。

@Bean("ttlQueue") public Queue queue() {
    Map<String, Object> map = new HashMap<String, Object>();
     map.put("x-message-ttl", 11000); // 队列中的消息未被消费 11 秒后过期 return new Queue("GP_TTL_QUEUE", true, false, false, map);
    }

2) 设置单条消息的过期时间 在发送消息的时候指定消息属性。

MessageProperties messageProperties = new MessageProperties(); 
    
    messageProperties.setExpiration("4000"); // 消息的过期属性，单位 ms
    
    Message message = new Message("这条消息 4 秒后过期".getBytes(), messageProperties); 
    
    rabbitTemplate.send("GP_TTL_EXCHANGE", "gupao.ttl", message);

死信队列

消息在某些情况下会变成死信(Dead Letter)。

队列在创建的时候可以指定一个死信交换机 DLX(Dead Letter Exchange)。 死信交换机绑定的队列被称为死信队列 DLQ(Dead Letter Queue)，DLX 实际上 也是普通的交换机，DLQ 也是普通的队列(例如替补球员也是普通球员)。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 6][]

什么情况下消息会变成死信?

)消息被消费者拒绝并且未设置重回队列:(NACK || Reject ) && requeue == false

2)消息过期

3)队列达到最大长度，超过了 Max length(消息数)或者 Max length bytes (字节数)，最先入队的消息会被发送到 DLX。

死信队列如何使用?

1、声明原交换机(GP\_ORI\_USE\_EXCHANGE)、原队列(GP\_ORI\_USE\_QUEUE)，相互绑定。

队列中的消息 10 秒钟过期，因为没有消费者，会变成死信。指定原队列的死信交换

机(GP\_DEAD\_LETTER\_EXCHANGE)。

@Bean("oriUseExchange")
    public DirectExchange exchange() {
        return new DirectExchange("GP_ORI_USE_EXCHANGE", true, false, new HashMap<>()); 
    }
    
    @Bean("oriUseQueue") public Queue queue() {
    
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("x-message-ttl", 10000); // 10 秒钟后成为死信
        map.put("x-dead-letter-exchange", "GP_DEAD_LETTER_EXCHANGE"); // 队列中的消息变成死信后，                    
        进入死信交换机
    return new Queue("GP_ORI_USE_QUEUE", true, false, false, map); }
    
    @Bean
    public Binding binding(@Qualifier("oriUseQueue") Queue queue,@Qualifier("oriUseExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("gupao.ori.use");
     }

2 、 声 明 死 信 交 换 机 ( GP\_DEAD\_LETTER\_EXCHANGE ) 、 死 信 队 列 (GP\_DEAD\_LETTER\_QUEUE)，相互绑定

@Bean("deatLetterExchange")
        public TopicExchange deadLetterExchange() {
            return new TopicExchange("GP_DEAD_LETTER_EXCHANGE", true, false, new HashMap<>()); 
        }
        @Bean("deatLetterQueue")
        public Queue deadLetterQueue() {
            return new Queue("GP_DEAD_LETTER_QUEUE", true, false, false, new HashMap<>());
        }
        
        @Bean
        public Binding bindingDead(@Qualifier("deatLetterQueue") Queue         
           queue,@Qualifier("deatLetterExchange") TopicExchange exchange) {
           return BindingBuilder.bind(queue).to(exchange).with("#"); // 无条件路由
         }

3、最终消费者监听死信队列。

4.生产发送消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 7][]

延迟队列

我们在实际业务中有一些需要延时发送消息的场景，例如: 1、 家里有一台智能热水器，需要在 30 分钟后启动  
2、 未付款的订单，15 分钟后关闭

RabbitMQ 本身不支持延迟队列，总的来说有三种实现方案:

1、 先存储到数据库，用定时任务扫描

2、 利用 RabbitMQ 的死信队列(Dead Letter Queue)实现

3、 利用 rabbitmq-delayed-message-exchange 插件

[https_blog.csdn.net_wang1988081309_article_details_97374677]: https://blog.csdn.net/wang1988081309/article/details/97374677
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70]: /images/20230209/b533193b3cb24fb98bfa135686a65b66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200715221021151.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200715222222958.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[RabbitMQ]: https://blog.csdn.net/oldshaui/article/details/88027554
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2020071523154963.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200715231617868.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200715231631147.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200715232203333.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200715232819471.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L29sZHNoYXVp,size_16,color_FFFFFF,t_70