[大数据]rabbitMQ死信队列、延迟队列、消息积压、幂等性、集群

死信队列 英文缩写 DLX 全称Dead Letter Exchange（读音 带的 来特 一可死陈只 死信交换机） 别的mq没有交换机这个概念 所以叫做死信队列 在tabbitMq中可以叫做死信交换机

根据名字可以看出 这个队列用来保存 被丢弃或者被删除的信息等 被丢弃被删除的信息跟死了一样 简称死信 把这些数据放到一个队列中 称为死信队列
当一个消息被删除变为Dead message（读音 带的 吗c只 这个单词就是专业术语 就是这个消息死了死了不就是被删了）后 可以被重新发送到另一个交换机 这个交换机就是dlx 然后这个交换机可以再发给队列

称为死信的条件

队列存满了 无法再存消息 之后来的消息都会变成死信

消费者拒绝接收并且 不把消息重新发给队列

队列设置了存活时间（TTL） 当消息超过存活时间被删除了 就会变成死信

死信队列的大体思想

发送者把数据发送给正常的交换机 正常的交换机把数据给 队列 然后监听者接收到消息 拒收并把消息丢弃 然后这时队列就会把这个消息发送给死信交换机 然后死信交换机发送给队列

步骤
发送者 -》 正常的交换机 -》正常的队列 -消息拒收并丢弃-》死信交换机 -》新一个正常的队列

用到了两个交换机 两个队列 一个交换机为死信交换机 一个交换机就是普通正常的交换机 第一个队列绑定一个死信交换机 第二个队列就一个普通的队列

经过上面分析 可以说 从队列到交换机的交换机称为死信交换机 由交换机到队列叫做普通交换机

设置死信交换机的参数
x-dead-letter-exchange：这里写死信交换机名称

x-dead-letter-routing-key：这里写发送给死信交换机的routing key

代码实现

spring方式

声明正常的队列（test_queue_dlx）和交换机（test_exchange_dlx）

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
    &lt;!&ndash;3. 正常队列绑定死信交换机&ndash;&gt;
    <rabbit:queue-arguments>
        &lt;!&ndash;3.1 x-dead-letter-exchange：死信交换机名称&ndash;&gt;
        <entry key="x-dead-letter-exchange" value="exchange_dlx" />

        &lt;!&ndash;3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey&ndash;&gt;
        <entry key="x-dead-letter-routing-key" value="dlx.hehe" />

        &lt;!&ndash;4.1 设置队列的过期时间 ttl&ndash;&gt;
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
        &lt;!&ndash;4.2 设置队列的长度限制 max-length &ndash;&gt;
        <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

声明死信队列(queue_dlx) 死信交换机(exchange_dlx)

<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

这样配置好 然后我们使用发送者给test_queue_dlx这个队列发消息 然后监听者监听这个队列 把消息全部拒收 test_queue_dlx这个队列就会把消息发送给exchange_dlx 然后exchange_dlx再发给queue_dlx

springboot方式

在配置类中 使用QueueBuilder.durable(QUEUE_NAME).withArguments() // withArguments这个方法绑定参数这个要一个map所以
@Bean("bootQueue3")
public Queue bootQueue3(){

    Map<String,Object> arguments = new HashMap<>();
    arguments.put("x-dead-letter-exchange", "这里写死信交换机名字 比如bootExchange");
    arguments.put("x-dead-letter-routing-key", "这里写死信交换机的路由key");

    return QueueBuilder.durable(QUEUE_NAME).withArguments(arguments).build();
}
    这样写 也可以使用 Queue的构造方法  Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) 这个构造也可以实现 各个参数的意思

* @param name队列名-不能为空;设置为""以让代理生成名称。
* @param durable true如果我们声明了一个持久队列(队列将在服务器重启后继续存在)
* @param exclusive true如果我们声明了一个独占队列(该队列只会被声明者使用)
*连接)
* @param autoDelete true表示当队列不再被使用时，服务器应该删除该队列
* @param arguments用于声明队列的参数

死信交换机

    @Bean("bootExchange")
    public Exchange bootExchange(){

    return ExchangeBuilder.topicExchange("这里写交换机名字).durable(true持久false不持久 不持久mq重启交换机就会删除).build();}

然后就配置两个交换机跟两个队列的关联就可以了

延迟队列 在rabbitMq中是不支持延迟消费的 实现延迟只能使用TTL加死信队列

 就是给 第一个队列设置存活时间 并且不给它监听者 让它里面的消息到时间自动删除
  然后发给死信交换机 
   然后死信交换机就把消息发送给 死信队列 
   然后我们监听死信队列 就可以打到延迟消费（延迟队列）
    根据上面写的死信队列 把上面第一个队列加上存活时间 
    并且把它的监听者删了 就可以实现了
spring方式 设置过存活时间<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" /> 

springboot方式设置存活时间
	 Map<String,Object> arguments = new HashMap<>();
	  	  arguments.put("x-message-ttl", "1000");

    	 return QueueBuilder.durable(QUEUE_NAME).withArguments(arguments).build();

分析总结
当消息成为死信如果配置了死信交换机 就会发送给死信交换机 然后被死信交换机发送给死信队列

死信交换机、死信队列跟普通的交换机、队列没有区别 如果非要说区别 队列给交换机发消息这个交换机称为死信交换机 被称为死信交换机的交换机给队列发送消息这个队列称为死信队列

消息成为死信的三种情况：
队列消息长度到达限制；
消费者拒接消费消息，并且不重回队列；
原队列存在消息过期设置，消息到达超时时间未被消费；

幂等性问题

幂等性同一个消息被多次消费 但是结果是一样的

使用乐观锁实现 在消息中加入一个版本 执行这个消息之后会让这个版本加1 同一个消息再一次被消费版本对不上就不执行

消息积压

   比如监听者宕机、监听者处理消息的能力不足、发送者流量过大导致的消息积压

可以上线更多的消费者，进行正常消费或者上线专门的队列消费服务（比如把消息全部存到硬盘中或者说存到数据库中 确保数据不能丢失） 将消息慢慢处理

集群 这里写的rabbit镜像集群
先把下面用到的配置文件备份！！！
先把下面用到的配置文件备份！！！
先把下面用到的配置文件备份！！！
以防配置不成功重新配置的时候无法还原配置文件

首先配置linux我的linux是圣斗7（centos 7） 这里配置了三台机器做集群
vim /etc/hosts 编辑这个文件 三台机器都要执行这个并追下面的三条数据

rabbitmq所在机器的ip mq1
rabbitmq所在机器的ip mq2
rabbitmq所在机器的ip mq3
添加这三条数据

然后在这三台机器分别执行下面的命令 一个机器执行一个命令 改个名字 不知道为什么要改名 知道可以跟我说说吗

	hostnamectl set-hostname mq1
	hostnamectl set-hostname mq2
	hostnamectl set-hostname mq3

上面的命令可能报错

vim /etc/hostname  可以执行这个命令 在这个文件中 根据当前的机器 写rabbitmq1、rabbitmq2、rabbitmq3 
比如
 第一台机器 就写rabbitmq1然后退出保存
 第二台就写rabbitmq2退出保存
 第三台就写rabbitmq3退出保存以此类推
 重启之后就可以看到改名成功root@mq1这样的一个东西

rabbitMQ做集群要erlang.cookie文件中的内容一样这里面就一个key

scp  /var/lib/rabbitmq/.erlang.cookie  mq2（这里也可以写ip地址）:/var/lib/rabbitmq/.erlang.cookie
scp  /var/lib/rabbitmq/.erlang.cookie  mq3:/var/lib/rabbitmq/.erlang.cookie

配置玩完之后启动三台机器

除了了mq1机器其它的机器全部执行下面的命令

rabbitmqctl stop_app

rabbitmqctl join_cluster --ram rabbit@mq1 // 可以看作为吧mq1这台机器当做老大 根据它组集群mq1这台机器中不用做其它的都做
							   rabbit@mq1 好像也可以写成ip

rabbitmqctl start_app

rabbitmq-plugins enable rabbitmq_management

systemctl restart rabbitmq-server.service

查看集群状态

rabbitmqctl cluster_status  // 执行完就可以看到集群名字 如果看不到那就配置失败了

在rabbitMQ主页也可以看到

然后使用haproxy做负载均衡实现 rabbitMQ高可用
配置haproxy挺简单 我这系统有问题一直配不成功我就不写了 随便百度一个haproxy配置的文章就行