RabbitMQ学习笔记04--ttl--死信队列--集群搭建

2023-10-22 15:12:28

RabbitMQ学习笔记04

01 过期时间TTL
- 队列ttl
- 消息ttl
02 死信队列
03 RabbitMQ内存磁盘的监控
RabbitMQ集群搭建
- CentOS搭建rabbitmq集群
- docker搭建rabbitmq集群

01 过期时间TTL

过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过了之后消息将自动被删除。RabbitMQ可以对消息和队列设置TTL。目前有两种方法可以设置。

第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间。
第二种方法是对消息进行单独设置，每条消息TTL可以不同。

如果上述两种方法同时使用，则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值，就称为dead message被投递到死信队列，消费者将无法再收到该消息。

队列ttl

配置ttl队列

@Configuration
public class TTLRabbitMQConfiguration {
    
    //交换机
    @Bean
    public DirectExchange ttlDirectExchange(){
        return new DirectExchange("ttl_direct_exchange", true, false);
    }
    
    //队列
    @Bean
    public Queue directTtlQueue(){
        //设置过期时间
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000); //注意这里一定是一个int类型
        return new Queue("ttl.direct.queue", true, false, false, args);
    }
    
    //绑定
    @Bean
    public Binding directTTLSmsBinding(){
        return BindingBuilder.bind(directTtlQueue()).to(ttlDirectExchange()).with("ttl");
    }
    
}

生产者方法

public void makeOrderTtl(int userId, int productId, int num) {
        //1.根据商品id查询库存是否充足
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功" + orderId);
        //3.通过MQ来完成消息的分发
        String exchangeName = "ttl_direct_exchange";
        String routingKey = "ttl";
        /*分发方法
         * param1 交换机
         * param2 Routing key/队列名称
         * param3 消息内容
         * */
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);

运行生产者模块的测试发送消息给ttl队列
从Web端可以看到5秒后消息过期消失

消息ttl

新增普通队列

@Bean
    public Queue directTtlMessageQueue(){

        return new Queue("ttl.message.direct.queue", true);
    }
    @Bean
    public Binding directTTLMessageBinding(){
        return BindingBuilder.bind(directTtlMessageQueue()).to(ttlDirectExchange()).with("ttlMessage");
    }

生产方法

public void makeOrderTtlMessage(int userId, int productId, int num) {
        //1.根据商品id查询库存是否充足
        //2.保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功" + orderId);
        //3.通过MQ来完成消息的分发
        String exchangeName = "ttl_direct_exchange";
        String routingKey = "ttlMessage";

        //给消息设置过期时间
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");//这里是String类型
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };

        /*分发方法
         * param1 交换机
         * param2 Routing key/队列名称
         * param3 消息内容
         * */
        rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId, messagePostProcessor);
    }

注：设置过期的消息，过期后不会移入死信队列，但设置了ttl的队列中过期的消息可以集中转移到死信队列

02 死信队列

DLX，全称为Dead-Letter-Exchange , 可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message)之后，它能被重新发送到另一个交换机中，这个交换机就是DLX ，绑定DLX的队列就称之为死信队列。
消息变成死信，可能是由于以下的原因：

消息被拒绝
消息过期
队列达到最大长度

DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。
要想使用死信队列，只需要在定义队列的时候设置队列参数x-dead-letter-exchange指定交换机即可。

死信队列和交换机配置

@Configuration
public class DeadRabbitMQConfiguration {

    //交换机
    @Bean
    public DirectExchange deadDirectExchange(){
        return new DirectExchange("dead_direct_exchange", true, false);
    }

    //队列
    @Bean
    public Queue deadQueue(){
        return new Queue("dead.direct.queue", true);
    }

    //绑定
    @Bean
    public Binding deadBinding(){
        return BindingBuilder.bind(deadQueue()).to(deadDirectExchange()).with("dead");
    }

}

只是普通交换机和普通队列，将其指定为死信队列和死信交换机需要在ttl队列配置中指定

配置这两个参数

//队列
    @Bean
    public Queue directTtlQueue(){
        //设置过期时间
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 5000); //注意这里一定是一个int类型
        args.put("x-dead-letter-exchange", "dead_direct_exchange");
        args.put("x-dead-letter-routing-key", "dead");
        return new Queue("ttl.direct.queue", true, false, false, args);
    }

报错原因：队列一旦创建过，修改其参数不会覆盖，所以需要删掉队列重新创建

过期时间后，ttl.direct.queue中的消息就会被投放到死信队列中

设置ttl.direct.queue最大消息长度，超过长度后拒绝消息，从而投放到私信队列中

//队列
    @Bean
    public Queue directTtlQueue(){
        //设置过期时间
        Map<String, Object> args = new HashMap<>();
        //args.put("x-message-ttl", 5000); //注意这里一定是一个int类型
        args.put("x-max-length", 5);
        args.put("x-dead-letter-exchange", "dead_direct_exchange");
        args.put("x-dead-letter-routing-key", "dead");
        return new Queue("ttl.direct.queue", true, false, false, args);
    }

循环发十一条消息

 @Test
    public void testTtlDirect(){
        for (int i = 0; i < 11; i++)
            orderService.makeOrderTtl(1, 1, 12);
    }

结果：

被拒绝的六条消息被转发到了死信队列

03 RabbitMQ内存磁盘的监控

RabbitMQ的内存警告
当内存使用超过配置的阈值或者磁盘空间剩余空间对于配置的阈值时，RabbitMQ会暂时阻塞客户端的连接，并且停止接收从客户端发来的消息，以此避免服务器的崩溃，客户端与服务端的心态检测机制也会失效。

OverView中，显示内存137MB，若达到735M，就会警告。
持久化：将内存中的数据同步到磁盘的过程。
爆红之后消息队列无法再接收消息。

RabbitMQ的内存控制
参考帮助文档：https://www.rabbitmq.com/configure.html
当出现警告的时候，可以通过配置去修改和调整
命令的方式：

rabbitmqctl set_vm_memory_high_watermark <fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

fraction/value 为内存阈值。默认情况是：0.4/2GB，代表的含义是：当RabbitMQ的内存超过40%时，就会产生警告并且阻塞所有生产者的连接。通过此命令修改阈值在Broker重启以后将会失效，通过修改配置文件方式设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启broker才会生效。

配置文件的方式：
配置文件：/etc/rabbitmq/rabbitmq.conf

#默认
#vm_memory_high_watermark.relative = 0.4
# 使用relative相对值进行设置fraction,建议取值在04~0.7之间，不建议超过0.7.
vm_memory_high_watermark.relative = 0.6
# 使用absolute的绝对值的方式，但是是KB,MB,GB对应的命令如下
vm_memory_high_watermark.absolute = 2GB

RabbitMQ的内存换页
在某个Broker节点及内存阻塞生产者之前，它会尝试将队列中的消息换页到磁盘以释放内存空间，持久化和非持久化的消息都会写入磁盘中，其中持久化的消息本身就在磁盘中有一个副本，所以在转移的过程中持久化的消息会先从内存中清除掉。
默认情况下，内存到达的阈值是50%时就会换页处理。
也就是说，在默认情况下该内存的阈值是0.4的情况下，当内存超过0.40.5=0.2时，会进行换页动作。
默认情况下，内存到达的阈值是50%时就会换页处理。
也就是说，在默认情况下该内存的阈值是0.4的情况下，当内存超过0.40.5=0.2时，会进行换页动作。
可以通过设置 vm_memory_high_watermark_paging_ratio 来进行调整。

vm_memory_high_watermark.relative = 0.4
vm_memory_high_watermark_paging_ratio = 0.7（设置小于1的值）

RabbitMQ的磁盘预警
当磁盘的剩余空间低于确定的阈值时，RabbitMQ同样会阻塞生产者，这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务器崩溃。
默认情况下：磁盘预警为50MB的时候会进行预警。表示当前磁盘空间第50MB的时候会阻塞生产者并且停止内存消息换页到磁盘的过程。
这个阈值可以减小，但是不能完全的消除因磁盘耗尽而导致崩溃的可能性。比如在两次磁盘空间的检查空隙内，第一次检查是：60MB ，第二检查可能就是1MB,就会出现警告。
通过命令修改：

rabbitmqctl set_disk_free_limit  <disk_limit>
rabbitmqctl set_disk_free_limit memory_limit  <fraction>
disk_limit：固定单位 KB MB GB
fraction ：是相对阈值，建议范围在:1.0~2.0之间。（相对于内存）

通过配置文件修改：

disk_free_limit.relative = 3.0
disk_free_limit.absolute = 50mb

RabbitMQ集群搭建

RabbitMQ这款消息队列中间件产品本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。因此，RabbitMQ天然支持Clustering。这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。
在实际使用过程中多采取多机多实例部署方式，为了便于同学们练习搭建，有时候你不得不在一台机器上去搭建一个rabbitmq集群，本章主要针对单机多实例这种方式来进行开展。

主要参考官方文档：https://www.rabbitmq.com/clustering.html

CentOS搭建rabbitmq集群

安装RabbitMQ

新建路径：

[root@iZ8vbhcwtixrzhsn023sghZ ~]# mkdir /usr/rabbitmq -p
[root@iZ8vbhcwtixrzhsn023sghZ ~]# cd /usr/rabbitmq/

Erlang安装

wget https://packages.erlang-solutions.com/erlang-solutions-2.0-1.noarch.rpm
rpm -Uvh erlang-solutions-2.0-1.noarch.rpm
yum install -y erlang

安装scoat

yum install -y socat

安装rabbitmq

 wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.13/rabbitmq-server-3.8.13-1.el8.noarch.rpm
 rpm -Uvh rabbitmq-server-3.8.13-1.el8.noarch.rpm

启动rabbitmq

# 启动服务
> systemctl start rabbitmq-server
# 查看服务状态
> systemctl status rabbitmq-server
# 停止服务
> systemctl stop rabbitmq-server
# 开机启动服务
> systemctl enable rabbitmq-server

查看rabbitmq是否正常运行

ps aux|grep rabbitmq

停掉rabbitmq

假设有两个rabbitmq节点，分别为rabbit-1, rabbit-2，rabbit-1作为主节点，rabbit-2作为从节点。

启动第一个节点rabbit-1

sudo RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &

启动第二个节点rabbit-2

sudo RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &

再次查看rabbitmq进程

将rabbit-1作为主节点

#停止应用
> sudo rabbitmqctl -n rabbit-1 stop_app
#目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
> sudo rabbitmqctl -n rabbit-1 reset
#启动应用
> sudo rabbitmqctl -n rabbit-1 start_app

将rabbit-2作为从节点

# 停止应用
> sudo rabbitmqctl -n rabbit-2 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
> sudo rabbitmqctl -n rabbit-2 reset
# 将rabbit2节点加入到rabbit1（主节点）集群当中【Server-node服务器的主机名】
> sudo rabbitmqctl -n rabbit-2 join_cluster rabbit-1@'Server-node'
# 启动应用
> sudo rabbitmqctl -n rabbit-2 start_app

对rabbitmq操作用rabbitmqctl指令，-n后面跟节点名字

验证集群状态

sudo rabbitmqctl cluster_status -n rabbit-1

Web查看
默认是关闭Web图形界面的，打开

rabbitmq-plugins enable rabbitmq_management

给两个集群节点设置账号密码

rabbitmqctl -n rabbit-1 add_user admin admin
rabbitmqctl -n rabbit-1 set_user_tags admin administrator
rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*"

rabbitmqctl -n rabbit-2 add_user admin admin
rabbitmqctl -n rabbit-2 set_user_tags admin administrator
rabbitmqctl -n rabbit-2 set_permissions -p / admin ".*" ".*" ".*"

docker搭建rabbitmq集群

创建映射数据卷目录

[root@iZ8vbhcwtixrzhsn023sghZ ~]# cd /
[root@iZ8vbhcwtixrzhsn023sghZ /]# mkdir rabbitmqcluster
[root@iZ8vbhcwtixrzhsn023sghZ /]# cd rabbitmqcluster/
[root@iZ8vbhcwtixrzhsn023sghZ rabbitmqcluster]# mkdir rabbitmq01 rabbitmq02 rabbitmq03

创建并运行三个rabbitmq节点容器

docker run -d --hostname rabbitmq01 --name rabbitmqcluster01 -p 15672:15672 -p 5672:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

docker run -d --hostname rabbitmq02 --name rabbitmqcluster02 -p 5673:5672 --link rabbitmqcluster01:rabbitmq01 -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

docker run -d --hostname rabbitmq03 --name rabbitmqcluster03 -p 5674:5672 --link rabbitmqcluster01:rabbitmq01 --link rabbitmqcluster02:rabbitmq02  -e RABBITMQ_ERLANG_COOKIE='rabbitmq_cookie' rabbitmq:management

启动容器成功后，可以访问
http://(服务器ip):15672
http://(服务器ip):15673
http://(服务器ip):15674
Web管理界面

容器节点加入集群

进入第一个rabbitmq节点容器

docker exec -it rabbitmqCluster01 bash

完成重启清除历史数据操作

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
exit

进入第二个节点容器执行如下命令：

[root@iZ8vbhcwtixrzhsn023sghZ rabbitmqcluster]# docker exec -it rabbitmqCluster02 bash
root@rabbitmq02:/# rabbitmqctl stop_app
Stopping rabbit application on node rabbit@rabbitmq02 ...
root@rabbitmq02:/# rabbitmqctl reset
Resetting node rabbit@rabbitmq02 ...
root@rabbitmq02:/# rabbitmqctl join_cluster rabbit@rabbitmq01
Clustering node rabbit@rabbitmq02 with rabbit@rabbitmq01
root@rabbitmq02:/# rabbitmqctl start_app
Starting node rabbit@rabbitmq02 ...
 completed with 3 plugins.
root@rabbitmq02:/# exit
exit

然后是第三个节点：

[root@iZ8vbhcwtixrzhsn023sghZ ~]# docker exec -it rabbitmqCluster03 bash
root@rabbitmq03:/# rabbitmqctl stop_app
Stopping rabbit application on node rabbit@rabbitmq03 ...
root@rabbitmq03:/# rabbitmqctl reset
Resetting node rabbit@rabbitmq03 ...
root@rabbitmq03:/# rabbitmqctl join_cluster --ram rabbit@rabbitmq01
Clustering node rabbit@rabbitmq03 with rabbit@rabbitmq01
root@rabbitmq03:/# rabbitmqctl start_app
Starting node rabbit@rabbitmq03 ...
 completed with 3 plugins.
root@rabbitmq03:/# exit
exit

码农公寓