RabbitMQ介绍与延时队列

朱雀 2022-10-10 12:35 157阅读 0赞

# RabbitMQ特性 #

*  **消息可靠性**：典型的生产者-消费者模型，发送端有消息发送确认机制，服务端有消息持久化方案，消费端有消息Ack机制
 *  **灵活的消息路由分发**：多种多样的交换机
 *  **多语言客户端开发AMQP**：只要实现了AMQP，就能和RabbitMQ对接
 *  **友好管理界面**：通过插件可以在浏览器通过界面来管理
 *  **灵活插件配置**：

### **RabbitMQ缺点：** ###

*  RabbitMQ不提供高可用的集群方案，需要手工实现[HAProxy + keepalived][HAProxy _ keepalived]方式进行集群搭建
 *  性能方面、处理能力没有RocketMQ和Kafka高

## RabbitMQ架构与核心组件 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70][]

RabbitMQ是一个典型的CS架构，是一个典型的生产者-消费者模型。

**Broker**：每一台单独运行的RabbitMQ进程就是一个Broker。多个Broker可以通过技术方案组成集群环境。

**virtual host**：虚拟主机，一个broker中有多个vh，用来环境隔离，区分不同环境（比如用户系统和订单系统划分在两个vh上），同一个vh中每个队列对象是唯一的，而不同vh之间，队列对象是可以相同的

**Exchange**：消息路由的交换机，是消息接收的通道。一条消息由producer产生之后，发送给broker，必须要到达交换机，然后由交换机转交给队列

**routing key**：消息路由分发的关键字，与binding中的key进行匹配从而分发到交换机绑定的指定队列queue。

**注：**发送一条消息，必须指定ExchangeName，告诉发送给哪个交换机；其次需要指定routing key，消息路由分发的关键字；

**connection**：客户端producer/consumer与服务端broker连接的连接对象，通过指定Host和Port去连接到对应的broker，还需要指定连接到哪个virtual host。但是最终实际操作对象是channel对象，一个connect可以创建多个channel，可以把connection理解成是JDBC中的连接池。

**channel**：producer中的channel是负责真正操作的对象，和交换机进行交互；consumer中的channel直接监听在对应的queue上，从而获取消息。

## RabbitMQ中提供的Exchange ##

*  direct-exchange：直连类型交换机，根据routing key和binding key进行全匹配找到对应的queue
 *  topic-exchange：主题类型交换机，根据通配符模糊匹配
 *  fanout-exchange：广播类型交换机，和交换机绑定的队列都会收到消息
 *  headers-exchange：头交换机，根据header中的值匹配

### direct-exchange直连交换机 ###

根据routing key和binding key进行全匹配，匹配到几个队列，那么消息就会发送到几个队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1][]

### topic-exchange主题交换机 ###

提供两个通配符\*和\#，根据通配符来进行匹配对应的队列。

**\* **代表英文一个单词

**\# **代表0个或多个单词

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2][]

### fanout-exchange广播交换机 ###

发送消息时不需要指定路由关键字routing key，交换机与队列之间的绑定关系也不需要binding key，因为是广播，需要将消息发送给交换机所绑定的所有队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3][]

### headers-exchange头交换机 ###

与header头信息进行配置，匹配其中一个则符合条件

**订单超时（30分钟）未支付，怎么办？**

定时任务+扫表（低级）：定时（比如3分钟）去扫描一次表，把未支付的订单退回到仓库

延时队列（高级）：30分钟后去判断当前订单是否支付，如果未支付退回到仓库

## RabbitMQ高可用方案 ##

*  Cluster普通模式：只同步元数据

Cluster单机多节点集群

Cluster多机多节点集群

*  镜像模式：同步内容

**有状态和无状态**

如果一个应用，复制多份，分别访问，可以得到一样的结果，请求一个应用修改了值，其他应用同时也会修改，则此应用是无状态的。

容器化水平扩容，需要用无状态的应用。

RabbitMQ消息可靠性

## RabbitMQ中的死信问题及处理方案 ##

**死信 dead-letter如何产生？**

*  RabbitMQ consumer主动拒绝消费
 *  消息的TTL（Time To Live）过期，设置队列中的消息5秒钟过期，那么TTL就是5秒
 *  消息在加入队列时超过队列的长度

**解决方案**

当消息成为死信之后，若该队列配置了**死信交换机**，则将消息按规则转发给该死信交换机（必须为主题交换机），如果没有配置死信交换机，则消息丢弃不再处理，消息消亡。

### 延时队列实现 ###

通过死信来实现延时队列。

![20210427223437604.png][]

### 延时策略回调 ###

**方案一：定时任务 + 扫表**

> 1.  创建任务表 \[callTime, status\]
> 2.  定时Job发起数据库查询SQL \[callTime < now()  and status = 1\]
> 3.  拿到任务若回调成功 \[update status = 0\]
> 4.  拿到任务若回调失败 \[update callTime = callTime + 间隔时间\]

**方案二：redis的zset数据类型**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4][]

**方案三：延时队列**

执行一个任务，如果执行成功则任务完成，如果执行失败，会延时10秒后再次回调执行，如果还是失败，延时更长时间再次回调执行，如果依然失败，延时时间再次延长......

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5][]

### RabbitMQ延时队列 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 6][]

> 1.  某条消息交给某个交换机exchange1
> 2.  路由分发到对应的消费队列consume queue1
> 3.  消费者Consumer1拿到消息后调用远方服务Remote Server
> 4.  如果RemoteServer返回正确结果，则消费者采用Ack机制，代表消息被成功消费
> 5.  如果RemoteServer返回错误结果，则消费者在确认消费过程中执行Nack拒绝消费
> 6.  拒绝消费的消息则会成为当前队列的死信
> 7.  将死信转发给死信交换机，exchange2就是consume queue1的死信交换机
> 8.  exchange2根据通配符 **\# **将消息发送给delay queue2延时队列
> 9.  delay queue2设置当前队列消息过期时间TTL=10s，那么10秒后消息会自动过期，变成死信
> 10. 将delay queue2的死信转发给对应的死信交换机exchange3
> 11. exchange3继续执行exchange1的相同流程，如此往复执行

## RabbitMQ基本使用 ##

生产者发送消息

public class Producer {
        public static void main(String[] args) {
            ConnectionFactory factory = new ConnectionFactory();
            // 设置连接Broker实例的属性
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = null;
            Channel channel = null;
            try {
                // 创建一个Producer与Broker实例之间的连接
                connection = factory.newConnection("producer");
                // 在连接中创建一个信道
                channel = connection.createChannel();
                // 声明交换机
                channel.exchangeDeclare("t-exchange1", "topic");
                // 声明队列
                channel.queueDeclare("t-queue1", true, false, false, null);
                // 绑定队列给交换机
                channel.queueBind("t-queue1", "t-exchange1", "t-rk*");
                channel.queueDeclare("t-queue2", true, false, false, null);
                channel.queueBind("t-queue2", "t-exchange1", "t-rk*");
                channel.queueDeclare("t-queue3", true, false, false, null);
                channel.queueBind("t-queue3", "t-exchange1", "t-rk#");
                String message = "Hello, this is message from producer";
                // 生产者发送消息
                // mandatory：如果根据消息routingKey无法找到与exchange消息类型匹配的queue，是否将消息返回给生产者
                // immediate：如果exchange将消息路由到对应的queue上时发现queue中没有消费者，是否将消息返回给生产者
                channel.basicPublish("t-exchange1", "t-rk-01", true, false, null, message.getBytes());
                System.out.println("消息已发送!");
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != channel && channel.isOpen()) {
                        channel.close();
                    }
                    if (null != connection && connection.isOpen()) {
                        connection.close();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }
    }

消费者订阅消息

public class Consumer {
        public static void main(String[] args) {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = null;
            Channel channel = null;
            try {
                connection = factory.newConnection("consumer");
                channel = connection.createChannel();
                channel.basicConsume("f-queue1", true, new DeliverCallback() {
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        System.out.println("消费者" + consumerTag + "收到消息：" + new String(message.getBody(), "UTF-8"));
                    }
                }, new CancelCallback() {
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消费者" + consumerTag + "取消消费!");
                    }
                });
                System.out.println("开始接收消息...");
                System.in.read();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != channel && channel.isOpen()) {
                        channel.close();
                    }
                    if (null != connection && connection.isOpen()) {
                        connection.close();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }
    }

### RabbitMQ在Spring中的使用 ###

pom.xml中引入rabbitmq依赖

<dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.3.5.RELEASE</version>
    </dependency>

applicationContext.xml相关配置

rabbitMQ.xml相关配置

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/rabbit
         http://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd">
    
        
        <rabbit:connection-factory id="connectionFactory" virtual-host="/" username="guest" password="guest" host="127.0.0.1" port="5672" />
    
        
        <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" />
    
        
        
        <rabbit:queue name="MY_FIRST_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />
    
        
        <rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin">
            <rabbit:bindings>
                <rabbit:binding queue="MY_FIRST_QUEUE" key="FirstKey">
                </rabbit:binding>
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        
        <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="MY_DIRECT_EXCHANGE" />
    
        
        <bean id="messageReceiver" class="com.gupaoedu.consumer.FirstConsumer"></bean>
    
        
        <rabbit:listener-container connection-factory="connectionFactory">
            <rabbit:listener queues="MY_FIRST_QUEUE" ref="messageReceiver" />
        </rabbit:listener-container>
    
        
        <rabbit:queue name="MY_THIRD_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />
    
        
        <rabbit:topic-exchange name="MY_TOPIC_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin">
            <rabbit:bindings>
                <rabbit:binding queue="MY_THIRD_QUEUE" pattern="#.Third.#"></rabbit:binding>
            </rabbit:bindings>
        </rabbit:topic-exchange>
    
        
        <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory" exchange="MY_TOPIC_EXCHANGE" />
    
        
        <bean id="receiverThird" class="com.gupaoedu.consumer.ThirdConsumer"></bean>
    
        
        <rabbit:listener-container connection-factory="connectionFactory">
            <rabbit:listener queues="MY_THIRD_QUEUE" ref="receiverThird" />
        </rabbit:listener-container>
    </beans>

消息生产者

@Service
    public class MessageProducer {
        private Logger logger = LoggerFactory.getLogger(MessageProducer.class);
    
        @Autowired
        @Qualifier("amqpTemplate")
        private AmqpTemplate amqpTemplate; // 直连交换机
    
        @Autowired
        @Qualifier("amqpTemplate2")
        private AmqpTemplate amqpTemplate2; // 主题交换机
    
        /**
         * 演示三种交换机的使用
         *
         * @param message
         */
        public void sendMessage(Object message) {
            // Exchange 为 direct 模式，直接指定routingKey
            amqpTemplate.convertAndSend("FirstKey", "[Direct,FirstKey] "+message);
            amqpTemplate.convertAndSend("SecondKey", "[Direct,SecondKey] "+message);
    
            // Exchange模式为topic，通过topic匹配关心该主题的队列
            amqpTemplate2.convertAndSend("msg.Third.send","[Topic,msg.Third.send] "+message);
    
            // 广播消息，与Exchange绑定的所有队列都会收到消息，routingKey为空
            amqpTemplate2.convertAndSend("MY_FANOUT_EXCHANGE",null,"[Fanout] "+message);
        }
    }

消息订阅者/消费者

public class FirstConsumer implements MessageListener {
        private Logger logger = LoggerFactory.getLogger(FirstConsumer.class);
    
        public void onMessage(Message message) {
            logger.info("The first consumer received message : " + message.getBody());
        }
    }

## RabbitMQ常见问题 ##

**1、如何保证消息可靠性？**

*  用事务保证消息正确传递（但是会有性能问题，不推荐）
 *  生产端：开启confirm监听器（推荐）
 *  Broker实例：开启持久化
 *  消费端：使用手动ACK机制（默认是自动ACK）

从以下三个方面考虑可靠性：

生产端发送可靠性：confirm机制确保消息成功发送到Broker

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 7][]

Broker实例存储可靠性：Broker对消息持久化，确保消息不会丢失

消费端消费可靠性：消费者在消费的同时，将autoAck设置为false，然后通过手动确认的方式去确认消息被正确消费，以免引起不必要的消息丢失。

**2、如何保证消息不重复消费？**

消息确认机制和幂等机制：每个消息生成一个全局唯一MessageID，每次消费前判断下当前消息是否已经被消费，如果消费过了，返回ack，则会把消息从队列中删除。

**3、如何保证顺序消费？**

同一个交换机，同一个队列，同一个消费者，单线程消费

**4、消息重试机制？**

消费端在处理消息过程中可能会出错，此时需要设置消息重试机制，解决方案有以下两种：

*  在redis或者数据库中记录重试次数，达到最大重试次数以后消息进入死信队列或者其他队列，再单独针对这些消息进行处理；
 *  使用spring-rabbit中自带的retry功能；

spring:
      rabbitmq:
        listener:
          simple:
            retry:
              enabled: true #是否开启重试
              initial-interval: 3000ms #重试时间间隔
              max-attempts: 3 #重试次数
              max-interval: 15000ms #重试最大时间间隔
              multiplier: 2 #倍数，间隔时间*倍数=下一次的间隔时间，最大不能超过设置的最大间隔时间

**5、死信队列和延时队列**

**死信消息：**

1.  消息被拒绝消费（Basic.Reject或Basic.Nack），并且设置 requeue 参数的值为 false
2.  消息过期了，超过了队列中设置的TTL时间仍然没有被消费
3.  消息加入队列时，队列达到了最大的长度

**死信队列：**当消息变成死信后，如果这个消息所在的队列存在x-dead-letter-exchange参数，就会被发送到x-dead-letter-exchange对应值的交换机上，这个交换机就称为死信交换机，与这个死信交换机绑定的队列就是死信队列。

**延时队列：**实际上RabbitMQ中不存在延时队列，但是我们可以通过设置消息过期时间TTL和死信队列来构建延时队列。

**6、消息积压，如果保证消息不丢失？**

如果是consumer消费速度落后于producer生产的速度，可以扩容消费者群组。

如果消息积压比较严重，上百万条，那么写个临时consumer程序处理积压消息。

**7、分布式事务**

**8、订单超时如何处理？**

移到死信队列

**9、监控**

自研（通过rest api）：适合大厂有能力做监控平台的

prometheus：适合中小企业，节约成本

小项目用rabbitmq-management

**10、mandatory和immediate**

mandatory和immediate是AMQP协议中basic.publish方法中的两个标识位，它们都有当消息传递过程中不可达目的地时是否将消息返回给生产者的功能。

mandatory：为true时，如果exchange根据自身类型与消息routingKey无法找到与之匹配的queue，那么会调用basic.return方法将消息返回给生产者（Basic.Return + Content-Header + Content-Body）；当mandatory设置为false时，出现上述情形broker会直接将消息扔掉。

immediate：为true时，如果exchange将消息路由到对应的queue上时发现queue中没有消费者，那么这条消息不会放入到队列中，该消息会通过basic.return返回给生产者，不用消息入队列等待消费者消费。

[HAProxy _ keepalived]: https://www.cnblogs.com/dkblog/archive/2011/07/06/2098949.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70]: /images/20221005/93282a5b7885477280b0764164d737a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1]: /images/20221005/de8ec951bbe6407b919d5f0444fc2619.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2]: /images/20221005/c958ea7c91b34b3eaffe3bf656346c64.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3]: /images/20221005/9aeaf1b42ec647f181e0d1a4ad6d8210.png
[20210427223437604.png]: /images/20221005/c80c0968203d4f4288ff817e8dc63fc4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4]: /images/20221005/863ec04a53354f50884d9b435b6d9774.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5]: /images/20221005/e8757086ec4f4417976735ee67955534.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 6]: /images/20221005/0e0b66b057544df1a4851e789ecac7c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 7]: /images/20221005/43ad3bd3c2fd4f1088e4c72ee5ac2e6b.png