【消息队列】消息中间件RabbitMQ急速入门3-阿里云开发者社区

【消息队列】消息中间件RabbitMQ急速入门3

2023-05-23 220

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 【消息队列】消息中间件RabbitMQ急速入门

6.RabbitMQ死信队列

6.1.什么是TTL

time to live消息存活时间
如果消息在存活时间内未被消费，就会清除

RabbitMQ支持两种TTL

单独消息进行配置ttl
个队列进行配置ttl

6.2.什么是RabbitMQ的死信队列

没有被及时消费的消息存放的队列

6.3.什么是RabbitMQ的死信交换机

Dead Letter Exchange（死信交换机，缩写：DLX）当消息过期未被消费后，会通过死信交换机，转发到死信队列，这个交换机就是DLX死信交换机

6.4.消息有哪几种情况成为死信

消费者拒收消息（basicNack或者basicReject），并且没有重新入队requeue=false
消息在队列中未被消费，且到了过期时间的消息（TTL）
队列长度达到极限后，如果绑定死信交换机和死信队列就会被投放到死信队列

6.5.什么是延迟队列

一种带有延迟功能的消息队列，Producer将消息发送到消息队列的服务端，但并不希望该条消息立马投递，而是推迟到当前时间的之后的一个时间Consumer进行消费，也可以叫定时消息

使用场景

通过消息触发一些定时任务，比如在某一固定时间点向用户发送提醒消息
用户登录之后5分钟给用户做分类推送、用户多少天未登录给用户做召回推送
订单超时未支付关闭订单场景

业界一些实现方式

定时任务高精度轮回
采用RocketMQ自带延迟消息功能
RabbitMQ结合死信交换机死信队列做到延迟消息

6.6.死信队列+死信交换机实现延迟消息

/**
 * mq配置类
 * 新商家上架->new_merchant_queue->死信交换机->死信队列
 */
@Configuration
public class RabbitMQConfig {
    /**
     * 死信队列
     */
    public static final String LOCK_MERCHANT_DEAD_QUEUE = "lock_merchant_dead_queue";
    /**
     * 死信交换机
     */
    public static final String LOCK_MERCHANT_DEAD_EXCHANGE = "lock_merchant_dead_exchange";
    /**
     * 进入死信队列的路由key
     */
    public static final String LOCK_MERCHANT_ROUTING_KEY = "lock_merchant_routing_key";
    /**
     * 创建死信交换机
     * @return
     */
    @Bean
    public Exchange lockMerchantDeadExchange(){
        return new TopicExchange(LOCK_MERCHANT_DEAD_EXCHANGE,true,false);
    }
    /**
     * 创建死信队列
     * @return
     */
    @Bean
    public Queue lockMerchantDeadQueue(){
        return QueueBuilder.durable(LOCK_MERCHANT_DEAD_QUEUE).build();
    }
    /**
     * 绑定死信交换机和死信队列
     * @return
     */
    @Bean
    public Binding lockMerchantBinding(){
        return new Binding(LOCK_MERCHANT_DEAD_QUEUE,Binding.DestinationType.QUEUE,
                LOCK_MERCHANT_DEAD_EXCHANGE,LOCK_MERCHANT_ROUTING_KEY,null);
    }
    /**
     * 普通队列，绑定的个死信交换机
     */
    public static final String NEW_MERCHANT_QUEUE = "new_merchant_queue";
    /**
     * 普通的topic交换机
     */
    public static final String NEW_MERCHANT_EXCHANGE = "new_merchant_exchange";
    /**
     * 路由key
     */
    public static final String NEW_MERCHANT_ROUTIING_KEY = "new_merchant_routing_key";
    /**
     * 创建普通交换机
     * @return
     */
    @Bean
    public Exchange newMerchantExchange(){
        return new TopicExchange(NEW_MERCHANT_EXCHANGE,true,false);
    }
    /**
     * 创建普通队列
     * @return
     */
    @Bean
    public Queue newMerchantQueue(){
        Map<String,Object> args = new HashMap<>(3);
        //消息过期后，进入到死信交换机
        args.put("x-dead-letter-exchange",LOCK_MERCHANT_DEAD_EXCHANGE);
        //消息过期后，进入到死信交换机的路由key
        args.put("x-dead-letter-routing-key",LOCK_MERCHANT_ROUTING_KEY);
        //过期时间，单位毫秒
        args.put("x-message-ttl",20000);
        return QueueBuilder.durable(NEW_MERCHANT_QUEUE).withArguments(args).build();
    }
    /**
     * 绑定交换机和队列
     * @return
     */
    @Bean
    public Binding newMerchantBinding(){
        return new Binding(NEW_MERCHANT_QUEUE,Binding.DestinationType.QUEUE,
                NEW_MERCHANT_EXCHANGE,NEW_MERCHANT_ROUTIING_KEY,null);
    }
}

@Component
@RabbitListener(queues = RabbitMQConfig.LOCK_MERCHANT_DEAD_QUEUE)
public class OrderMQListener {
    /**
     * 处理器，适配器，加上@RabbitHandler注解
     */
    @RabbitHandler
    public void messageHandler(String body, Message message, Channel channel) throws IOException {
        long tag = message.getMessageProperties().getDeliveryTag();
        System.out.println("message:"+message.toString());
        System.out.println("==============");
        System.out.println("消息标识tag:"+tag);
        System.out.println("消息体body:"+body);
        //如果发生异常就重新入队
        try{
            channel.basicAck(tag,false);
        }catch (Exception e){
            //拒收消息，重新入队
            channel.basicNack(tag,false,true);
        }
    }
}

7.RabbitMQ高可用集群模式

7.1.RabbitMQ集群模式介绍

普通集群

默认的集群模式，普通集群模式下同步交换机、队列、虚拟主机元数据，不同步消息信息，当消费者去B节点访问数据，数据在A节点，这时A节点先把数据转发给B节点，消费者在从B节点中pull消息。
存在问题：假如存在消息的节点宕机了，那么消费者想要消费这个消息，就要等当前节点恢复后才可以恢复正常，如果没有做消息持久化，则消息会丢失

镜像集群

队列做成镜像队列，让队列存在于各个节点中，和普通集群比较大的区别就是queue的message在各个节点之间同步，并不是在consumer获取时拉去，转发。

存在问题：由于镜像队列模式下，消息数量过去，大量的消息同步也会加大网络带宽的开销，适合高可用的项目，多节点性能会收到影响。

注意：集群需要保证各个节点有相同的token令牌，集群内各个节点的erlang.cookie需要相同，才可以相互通信

7.2.RabbitMQ搭建普通集群

准备三个mq节点

#节点一，主节点,创建-v映射目录
docker run -d --hostname rabbit_host1 --name rabbitmq1 -p 15672:15672 -p 5672:5672 -e RABBITMQ_NODENAME=rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=xdclass.net168  -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie_xdclass' --privileged=true -v /usr/local/rabbitmq/1/lib:/var/lib/rabbitmq -v /usr/local/rabbitmq/1/log:/var/log/rabbitmq rabbitmq:management

#节点二,创建-v映射目录
docker run -d --hostname rabbit_host2 --name rabbitmq2  -p 15673:15672 -p 5673:5672 --link rabbitmq1:rabbit_host1 -e RABBITMQ_NODENAME=rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=xdclass.net168 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie_xdclass' --privileged=true -v /usr/local/rabbitmq/2/lib:/var/lib/rabbitmq -v /usr/local/rabbitmq/2/log:/var/log/rabbitmq rabbitmq:management

#节点三,创建-v映射目录
docker run -d --hostname rabbit_host3 --name rabbitmq3 -p 15674:15672 -p 5674:5672 --link rabbitmq1:rabbit_host1 --link rabbitmq2:rabbit_host2 -e RABBITMQ_NODENAME=rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=xdclass.net168 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie_xdclass' --privileged=true -v /usr/local/rabbitmq/3/lib:/var/lib/rabbitmq -v /usr/local/rabbitmq/3/log:/var/log/rabbitmq rabbitmq:management

参数说明：

--hostname：自定义Docker容器的hostname
--link：容器之间连接，link不可缺，使得三个容器相互通信
--privileged=true：使用该参数，container内的root拥有真正的root权限，否则容器出现perission denied
-v：宿主机和容器路径映射
RABBITMQ_DEFAULT_USER=admin：配置用户名
RABBITMQ_DEFAULT_PASS=123456：配置密码
Erlang Cookie值必须相同，相当于不同节点间通信的密钥。

配置集群：

#节点一配置集群
docker exec -it rabbitmq1 bash
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
exit
#节点二假如集群，--ram是以内存方式加入，忽略该参数默认未磁盘节点
docker exce -it rabbitmq2 bash
rabbitmqctl stop_app
rabbitmqctl join_cluster --ram rabbit@rabbit_host1
exit
#节点三假如集群，--ram是以内存方式加入，忽略该参数默认未磁盘节点
docker exce -it rabbitmq3 bash
rabbitmqctl stop_app
rabbitmqctl join_cluster --ram rabbit@rabbit_host1
exit
#查看集群节点状态，配置启动三个节点，1个磁盘节点和2个内存节点
rabbitmqctl cluster_status

访问节点1的web控制台，可以看到多个节点

测试：node1节点创建队列，发送消息（可以选择消息的持久化，Spring-AMPQ中默认就是持久化），node2和node3都可以看到消息同步，kill掉node1节点，发现node2和node3为NaN模式，如果是非主节点创建队列和发送消息，则其他队列也可以显示。

7.3.RabbitMQ高可用mirror镜像集群配置策略

背景：解决普通集群主节点宕机出现数据丢失的现象

RabbitMQ的策略policy是用来控制和修改集群的bhost队列和Exchange复制行为，就是要设置那些Exchange或者queue的数据需要复制、同步，以及如何复制同步

创建一个策略来匹配队列

路径：rabbitmq管理界面->Admin->Policies->Add/update a policy

参数：策略会同步一个VirtualHost中的交换机和队列数据

name：自定义策略名称
Pattern：^匹配符，代表匹配所有
Definition：ha-mode=all为匹配类型，分为3中模式

ha-model：指明镜像队列的模式，可选下面一种
all：表示在集群中所有的节点上进行镜像同步
exactly：表示在指定个数的节点上进行镜像同步，节点的个数由ha-params指定
nodes：表示在指定的节点上进行镜像同步，界定啊名称由ha-params指定
ha-sync-mode：镜像消息同步方式automatic（自动），manually（手动）

配置好后，+2的意思就是由三个节点，一个节点本身和两个镜像节点

集群重启顺序

集群重启的顺序是固定的，并且是相反的
启动顺序：磁盘节点=》内存节点
关闭顺序：内存节点=》磁盘节点

7.4.SpringBoot配置RabbitMQ集群

把host和port节点去掉换成addresses :10.211.55.13:5672,10.211.55.13:5673,10.211.55.13:5674