Rabbitmq从入门到精通

1. RabbitMQ简介

1.2 组件介绍

1. 交换机和队列的名称是不能相同的。相互隔离，不同用户之间创建的交换机是不可见的。
2. 生产者通过tcp连接到broker，broker是接收和分发消息的实体，vhost是一个虚拟主机，不同用户之间的隔离。
3. 每次连接占用一个管道，每一次连接可以多次发送消息。
4. exchange：交换机，
5. queue：存放消息的队列

rabbitMQ四大核心：生产者，交换机，队列，消费者
AMQP协议四大核心：消息，没交换，队列，绑定。

1.1 消息队列的选择

1.kafka：大量数据选择，日志采集功能
2.rokcetmq 天生金融，稳定很好
3.rabbitmq 时效性好，数据量不大使用

1.2 Docker 安装rabbitmq

1. docker安装rabbitmq：docker pull rabbitmq
2. docker run -d --name my-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:latest
3. docker exec -it <container_id> /bin/bash
4. 安装web管理插件：rabbitmq-plugins enable rabbitmq_management

1.4 消息应答机制

消息应答机制：再消费者应答之后进行应答，保证消息不丢失，
自动应答需要在消息很多就不出问题自动应答在高效率能够处理的情况下进行应答。
使用手动应答，肯定确认，否定确认，否定确认批量处理参数。手动应答表示减少网络拥堵

消息重新入队：
保证消息不丢失：消费者没有进行ack确认的话，mq将自动重新入队
消息手动确认： //接收消息回调函数

  DeliverCallback deliverCallback =(consumerTage,message) -> {
          System.out.println("接受到的消息是："+new String(message.getBody()));
           //开启手动应答模式
           channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
    };
1
2
3
4
5

队列持久化 ： channel.queueDeclare(queueName,true,false,false,null); 第二个参数
消息持久化：将消息持久化在生产者中，通过发布消息的时候添加参数实现消息持久化机制。但是刚存储在磁盘的时候，还没有存储完成，消息还有一个缓存的时间点，并没有真正写入，持久性不强
不公平分发：根据消费者消费的能力不同需要加权发送消息。可以设置参数：channel.basicQos（1），在消费端进行设置：0代表轮询，1代表不公平分发
预取值：指定一个消费者取多少数据，可以设置参数：channel.basicQos（2），在消费端进行设置：0代表轮询，1代表不公平分发，多的值就域取值

2.1 交换机类型

交换机的类型：direct，fanout，topic，header

2.1 direct

交换机和路由键进行绑定
一个key可以绑定多个队列，当完全匹配到key1时候，队列1和队列2都可以接收到消息

2.2 fanout

扇出类型交换机，将消息分发给所有绑定此交换机的队列。路由键就无关。消费者消费的这个队列只要是绑定的这个队列，就会接受到消息

2.3 topic：

主体类型交换机，和Dieect类似，也是通过路由键进行分发，tiopic进行模糊匹配 "*"代表一个部分 "#"代表0个或者多个部分。如果绑定的路由键是#的时，则接收所有的信息

2.4header

匹配amqp消息的头部而不是路由键，hader和direct是一致的，但是性能差很多，几乎不使用

3 RabbitMQ集群搭建

rabbitMQ集群搭建：
要在 RabbitMQ 中搭建集群，可以遵循以下步骤：

安装 RabbitMQ，在每个节点上安装 RabbitMQ，并确保它们都使用相同的版本和配置。配置节点

在每个节点上编辑 /etc/hostname 文件中的主机名，并编辑 /etc/hosts 文件以包括所有节点的 IP 地址和主机名。例如：

192.168.1.1 rabbitmq-1
192.168.1.2 rabbitmq-2
192.168.1.3 rabbitmq-3
1
2
3

确保所有节点都能够互相解析。

启用 Clustering 插件，执行以下命令启用 RabbitMQ 的 rabbitmq_clusterer 插件：
rabbitmq-plugins enable rabbitmq_clusterer
配置集群

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster [--ram] <disc | ram> <node_name> [--via-rabbitmq-node <rabbit>] [--[no-]migrate-data] [--timeout <timeout>]
rabbitmqctl start_app
1
2
3
4

其中，

stop_app 和 start_app 命令用于停止和重新启动 RabbitMQ 应用程序；
reset 命令用于删除旧的数据并创建一个新的空节点；
join_cluster 命令将当前节点加入到集群中；
–ram 选项用于将节点设置为仅内存模式；
<disc | ram> 参数表示节点的类型，可以是磁盘节点或仅内存节点；
<node_name> 参数表示节点的名称；
–via-rabbitmq-node 选项用于指定用于加入集群的 RabbitMQ 节点的主机名或 IP 地址；
–[no-]migrate-data 选项用于指定是否迁移节点上的数据，默认为 false；
–timeout 选项用于指定操作超时时间。
验证集群状态

在其中一个节点上执行以下命令验证集群状态：

rabbitmqctl cluster_status

3.1docker 搭建 rabbitmq集群

docker network create rabbitmq-net
这将创建一个名为 rabbitmq-net 的 Docker 网络。

启动 RabbitMQ 节点

在三个不同的终端中分别执行以下命令：

docker run --name rabbitmq-node1 --hostname rabbitmq-node1 --net rabbitmq-net -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -d rabbitmq:3-management
docker run --name rabbitmq-node2 --hostname rabbitmq-node2 --net rabbitmq-net -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -d rabbitmq:3-management
docker run --name rabbitmq-node3 --hostname rabbitmq-node3 --net rabbitmq-net -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -d rabbitmq:3-management
1
2
3

这将在三个 Docker 容器中启动三个 RabbitMQ 节点。其中，rabbitmq-node1、rabbitmq-node2 和 rabbitmq-node3 分别表示节点的名称，
rabbitcookie 是用于加密 Erlang Cookie 的密码，rabbitmq:3-management 是 RabbitMQ 的 Docker 镜像名称（包含管理界面）。

使用 docker inspect 命令获取节点 IP 地址运行以下命令获取每个节点的 IP 地址：
docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' rabbitmq-node1
docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' rabbitmq-node2
docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' rabbitmq-node3
这将返回每个节点的 IP 地址。在后续步骤中，你需要使用这些地址来配置集群。

配置集群以具有 "rabbitmq-node1" 名称的节点为主节点，执行以下命令配置集群：

docker exec -it rabbitmq-node1 rabbitmqctl stop_app
docker exec -it rabbitmq-node1 rabbitmqctl reset
docker exec -it rabbitmq-node1 rabbitmqctl join_cluster rabbit@rabbitmq-node1 rabbit@rabbitmq-node2 rabbit@rabbitmq-node3
docker exec -it rabbitmq-node1 rabbitmqctl start_app
这将将节点 rabbitmq-node2 和 rabbitmq-node3 加入到名为 rabbit 的 RabbitMQ 集群中。
1
2
3
4
5
6
7
8
9
10
11
12
13

在管理界面中验证集群状态打开一个 Web 浏览器，输入地址 http://<任意一个节点的IP地址>:15672/ 即可查看 RabbitMQ 的管理界面。
输入用户名和密码（默认为 guest 和 guest），进入管理界面。在 “Cluster Status” 页面上可以验证集群状态。应该可以看到三个节点，且所有节点都处于运行状态。
现在，你已经成功地在 Docker 中搭建了 RabbitMQ 集群。你可以使用多个节点部署应用程序，以实现高可用性和扩展性。

4 死信队列

死信队列：无法被消费的消息，在某些时候由于特定的原因，队列中的消息无法消费，就成了死信消息。当消息出现异常的时候，将消息投递到死信队列

死信的来源：消息TTL过期，队列达到最大长度，消息被拒绝了

4.1 ttl消息过期：生产者添加延迟生产

        AMQP.BasicProperties properties = new AMQP.BasicProperties()
                .builder().expiration("10000").build();

        /***发送消息
         * 1.交换机名和曾
         * 2.队列名称
         * 3.其他参数信息
         * 4.发送消息的消息体
         * **/
        for (int i = 1; i < 1000; i++) {
            String message = "info"+i;
            System.out.println("发送消息： "+message);
            channel.basicPublish(NORMAL_EXCHANGE,"normal", properties,message.getBytes());
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

4.2 队列最大长度

通过在普通队列中设置队列的长度参数：注册队列携带参数

        //设置死信队列的参数，存储到hashmap中
        Map<String,Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        arguments.put("x-dead-letter-routing-key","sixin");
        //死信时间
        arguments.put("x-message-ttl",10000);
        //队列长度(超过长度就是死信队列)
//        arguments.put("x-max-length",6);
       channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
1
2
3
4
5
6
7
8
9

4.3

消息被拒绝：使用拒绝的方法，最后一个参数是不重新入队列
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);

5. 延迟队列

队列内部是有序的，体现在延时属性上希望在指定时间到了之后取出，延时队列就是存放需要在指定时间被处理的元素的队列还可以使用

延迟队列的使用常见

1. 订单在10分钟内未支付就取消（延迟队列过期时间10分钟）
2. 新创建的店铺，十天之内没有上传过商品就会发送消息提醒（）
3. 用户注册之后三天之内没有登录就进行短信提醒

基于死信的问题：rabitmq只会检查第一个消息是否过期，过期就去死信队列，如果第一个消息延迟的时间很长，第二个消息延迟的时间很短，第二个消息不会优先执行。
消息的可靠投递，通过死信保证消息至少被消费一次。通过

基于插件的延迟消息：可以在交换机中进行延迟，不是在队列中进行延迟。只需要一个交换机

6. 发布确认机制

导致rabbitmq服务器重启，期间会导致消息丢失，需要手动确认恢复。

确认机制方案：发送消息备份（缓存），当交换机收到消息就清除缓存。交换机和队列之间。交换机存在队列不存在的时候，消息就会丢失。

在配置文件中添加： publisher-confirm-type: correlated 实现confirmCallback接口

消息回退机制：在消息不可达的时候，配置文件添加 publish-returns=true 实现ReturnCallback接口。消息不能达到目的地的时候就进行消息回退

1. 备份交换机：交换机出现问题无法接受到消息，要求交换机发布确认，如果队列收不到也是，还有一种就是备份交换机。将消息发送给备份交换机。（消息检测和报警）
   两个队列，备份队列和报警队列

2. 幂等性：同一操作或者多次请求的结果是一致的。重复提交的问题；消息被被重复消费。消息幂等性是用全局id，每次消费的时候进行判断
   唯一id+指纹码机制：一些规则加时间戳，不是系统生成的，基本是业务规则拼接过来的，但是一定保证唯一性，使用查询语句判断
   Redis原子性：setnx命令

3. 优先级队列：订单催付的场景，大客户和小客户进行推送。订单优先处理。曾今是redis存放定时轮询，redis只能做list简单的消息队列，并不能实现优先队列。使用rabbitmq实现优先级
   对消息进行排序，队列中设置参数，params.put("x-max-priority",10) 在发送消息的时候确定消息的优先级 官方允许0-255之间 AMQP.BasicProperties.build().priority(5),build()

4. 惰性队列：消费者下线或者消息存放地，消息存储在内存中还是磁盘中。正常是在内存中的，惰性队列是在磁盘中的。队列有两种模式，default和lazy。可以在 注册队列的时候参数中设置。
   params.put("x-queue-mode",lazy)
   将消息存放到磁盘中。内存消耗小，惰性队列消费慢

6.1 发布确认原理

1.生产者发送给队列，队列持久化，消息设置持久化。发布确认才可以保证生产者消息不丢失
channel.confirmSelect();
2.单个确认发布：同步发布 ，速度很慢
2.批量确认发布：如果有错误的话不知道是哪一条出现错误的
3.异步确认发布：性价比是最高的，可靠性和效率是最高的，利用回调函数确认。添加chaner.addconfirmLisen(ack,noack)
将没有确定的消息，存储到基于内存，可以发布线程访问的队列，，并发链式调用，可以在确认 和回调之间进行消息传递。
new ConcurrentSkipListMap() 线程有序的hash表 在发送的时候将消息投递进去，当消息投递成功将消息清除