RabbitMQ入门到实战(2)-防止消息丢失

忘是亡心i 2022-05-27 03:17 203阅读 0赞

转载于：http://www.cnblogs.com/4----/p/6526033.html

# 1.简介 #

RabbitMQ中，消息丢失可以简单的分为两种：客户端丢失和服务端丢失。针对这两种消息丢失，RabbitMQ都给出了相应的解决方案。

# 2.防止客户端丢失消息 #

![image][]

如图，生产者P向队列中生产消息，C1和C2消费队列中的消息，默认情况下，RabbitMQ会平均的分发消费给C1C2（Round-robin dispatching），假设一个任务的执行时间非常长，在执行过程中，客户端挂了（连接断开），那么，该客户端正在处理且未完成的消息，以及分配给它还没来得及执行的消息，都将丢失。因为默认情况下，RabbitMQ分发完消息后，就会从内存中把消息删除掉。

# 3.消息确认（Message acknowledgment） #

为了解决上述问题，RabbitMQ引入了消息确认机制，当消息处理完成后，给Server端发送一个确认消息，来告诉服务端可以删除该消息了，如果连接断开的时候，Server端没有收到消费者发出的确认信息，则会把消息转发给其他保持在线的消费者。

## 验证上述问题 ##

首先，我们验证上述问题（客户端丢失消息）是否真的存在，对Consumer进行如下改造。

![image][image 1]

先生产两条消息

![image][image 2]

启动消费者，在消费者接收到消息，还没处理完成的时候，强制关掉

![image][image 3]

这时，观察控制台，发现两条消息都没有了，1条是在执行中丢失的，还有1条，已经分配给这个Consumer，还没来得及处理，也丢失了

![image][image 4]

这证明了上述问题是真的存在的，如果发生在生产环境，将产生难以预料的后果

## 引入消息确认机制 ##

为了方便观察，我们用CMD来运行Consumer，要通过maven打成可执行的JAR包，需要在pom.xml中增加如下配置

![复制代码][copycode.gif]

<build>
            <finalName>Consumer</finalName>
            <plugins>
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <appendAssemblyId>false</appendAssemblyId>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                        <archive>
                            <manifest>
                                <mainClass>com.liyang.ticktock.rabbitmq.App</mainClass>
                            </manifest>
                        </archive>
                    </configuration>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>assembly</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
    
            </plugins>
        </build>

![复制代码][copycode.gif]

上述配置描述了最终打包名字、入口类路径、带上依赖包、使用1.8版本的JDK进行打包，配置完后，就可以通过maven的install方法，在target目录生成可执行的jar包，如果包大小很小，应检查配置，是不是没有带上依赖包

![image][image 5]

再次改造Consummer类

![image][image 6]

install成可执行jar包，通过cmd开启两个consumer

![image][image 7]

通过Sender发送一条消息，然后用Ctrl+C结束先收到消息的Consumer，发现另外一个Consumer接收到了未处理完的消息

![image][image 8]

问题得到了解决，现在消费者在执行过程中死掉也不会丢失消息了

看一下发送确认的方法

![复制代码][copycode.gif]

1 /**
     2      * Acknowledge one or several received
     3      * messages. Supply the deliveryTag from the {
        @link com.rabbitmq.client.AMQP.Basic.GetOk}
     4      * or {
        @link com.rabbitmq.client.AMQP.Basic.Deliver} method
     5      * containing the received message being acknowledged.
     6      * @see com.rabbitmq.client.AMQP.Basic.Ack
     7      * @param deliveryTag the tag from the received 这个是RabbitMQ用来区分消息的，文档在这 8      * @param multiple true to acknowledge all messages up to and 为true的话，确认所有消息，为false只确认当前消息
     9      * including the supplied delivery tag; false to acknowledge just
    10      * the supplied delivery tag.
    11      * @throws java.io.IOException if an error is encountered
    12      */
    13     void basicAck(long deliveryTag, boolean multiple) throws IOException;

![复制代码][copycode.gif]

在官方文档中，这样描述deliveryTag

![image][image 9]

简单来说，就是RabbitMQ内部用来区分消息的一个标签，从envelope中获取就行了

## 忘记确认将引起内存泄漏 ##

RabbitMQ只有在收到消费者确认后，才会从内存中删除消息，如果消费者忘了确认（更多情况是因为代码问题没有执行到确认的代码），将会导致内存泄漏

验证一下

注释掉Consumer中的确认代码

![image][image 10]

![image][image 11]

运行Sender和Consumer，不停的生产消费消息，发现消费者在正常的消费消息

![image][image 12]

查看控制台，发现已经被吃掉了43KB的内存，所以，在试用过程中，一定要保证消息确认在任何情况下都可以发出，否则即使消费者处理完成，RabbitMQ也不会把消息在内存中清除，在该消费者断开连接之后，还会把消息转发给其他消费者重新处理，将引发难以预计的问题

![image][image 13]

![image][image 14]

# 4.消息的持久化 #

现在，消费者宕机已经无法影响到我们的消息了，但如果RabbitMQ重启了，消息依然会丢失。所幸的是，RabbitMQ提供了持久化的机制，将内存中的消息持久化到硬盘上，即使重启RabbitMQ，消息也不会丢失。但是，仍然有一个非常短暂的时间窗口（RabbitMQ收到消息还没来得及存到硬盘上）会导致消息丢失，如果需要严格的控制，可以参考[官方文档][Link 1]

要使用RabbitMQ的消息持久化，在声明队列时设置一个参数即可

![image][image 15]

注意，RabbitMQ不允许对一个已经存在的队列用不同的参数重新声明，对于试图这么做的程序，会报错，所以，改动之前代码之前，要在控制台中把原来的队列删除

![image][image 16]

重新声明队列后，发现Durable为true

![image][image 17]

重启RabbitMQ

![image][image 18]

队列的消息没有丢失

![image][image 19]

# 5.结束语 #

这一章介绍了RabbitMQ消息的确认和持久化，后面将会继续深入介绍RabbitMQ的其他特性

[image]: /images/20220527/59c9eb1d23c64d1fb9f5bc3c45c84336.png
[image 1]: /images/20220527/8dc31bcb2d5a4420a6e8660e0cd0b49f.png
[image 2]: /images/20220527/1a71603cc45e40efb24ee0265753722a.png
[image 3]: /images/20220527/c3910b7fed224ef69e907d1862cf99f0.png
[image 4]: /images/20220527/2e2a8c7434a242f29fcf03bf17c1ad53.png
[copycode.gif]: /images/20220527/ff79dfb96f9e4c388d0ecaea56344219.png
[image 5]: /images/20220527/7a8208f0e52f44a1b09edbc9c7f063ee.png
[image 6]: /images/20220527/f30ae51772344165a87440318daa28d8.png
[image 7]: /images/20220527/65e44fbe01424ffc8752bc07343ff5df.png
[image 8]: /images/20220527/b5c6d20eed994ba5b92d59018daacb95.png
[image 9]: /images/20220527/4aa88186043a4f28871f384ee815ff4d.png
[image 10]: /images/20220527/6db768cb21104bdfb70c3ced78b72169.png
[image 11]: /images/20220527/d648fbeb791a4a199b7943e111502052.png
[image 12]: /images/20220527/11b5727fa38143b2b683dd649cf92e7e.png
[image 13]: /images/20220527/04bbead43a6b4e39a302866c931b80db.png
[image 14]: /images/20220527/1d4c704530584dc88b0cd72a9d736ebd.png
[Link 1]: https://www.rabbitmq.com/confirms.html
[image 15]: /images/20220527/de2fdb0a8e1a47a9bb33abbcddc4538d.png
[image 16]: /images/20220527/7116937b94744a0f86dcff3c9abe7171.png
[image 17]: /images/20220527/b4eed0a037e242d0a639042c4c8d0788.png
[image 18]: /images/20220527/1eacb3d58d4a42a3b3219efe06a2f8c1.png
[image 19]: /images/20220527/66956f8642ae4f39ab253e3e199b9a74.png