rabbitmq connection lost之后重发消息

拼搏现实的明天。 2021-09-21 08:10 449阅读 0赞

在debug一个使用了rabbit的应用程序时，遇到这样一个问题。应用程序收到了rabbitmq的消息，正在处理，由于是debug，应用程序在编辑器里长时间的停留在某个节点，导致rabbitmq的连接丢失，随后继续执行程序时rabbitmq又重连，由于重连rabbitmq不确定之前的消息被成功的消费，所以又再次发送了之前的消息，然后service在处理完message之后，又重新收到了一样的event且再此处理。

rabbit连接的配置为：

10:50:59.233 [abc.abc-changed.exchange.group-1] INFO  o.s.a.r.l.SimpleMessageListenerContainer - Restarting Consumer@fba5072: tags=[[amq.ctag-ceBsEnpsb-Rqk7TSFFjExQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,18), conn: Proxy@554d4856 Shared Rabbit Connection: SimpleConnection@388a6cb2 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 63108], acknowledgeMode=AUTO local queue size=0

链接断开的log为

10:50:55.246 [abc.abc-changed.exchange.group-1] WARN  c.s.n.b.c.l.RabbitmqConnectionListener - Connection to RabbitMQ was lost. Emptying cache ...
    10:50:55.256 [abc.abc-changed.exchange.group-1] INFO  c.s.n.b.c.s.impl.BCCacheEvictor - clear all caches
    10:50:55.264 [abc.abc-changed.exchange.group-1] INFO  o.s.a.r.c.CachingConnectionFactory - Attempting to connect to: [localhost:5672]
    10:50:55.387 [abc.abc-changed.exchange.group-1] INFO  o.s.a.r.c.CachingConnectionFactory - Created new connection: rabbitConnectionFactory.publisher#3a98aa39:1/SimpleConnection@430bfa20 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 63146]
    10:50:55.413 [abc.abc-changed.exchange.group-1] INFO  o.s.amqp.rabbit.core.RabbitAdmin - Auto-declaring a non-durable, auto-delete, or exclusive Queue (spring.gen-Jt7qx2wXTeeqOeUOIsQpWg) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
    10:50:55.881 [abc.abc-changed.exchange.group-1] INFO  c.s.n.b.c.l.RabbitmqConnectionListener - Connection to RabbitMQ established. Creating bindings ...
    10:50:59.233 [abc.abc-changed.exchange.group-1] INFO  o.s.a.r.l.SimpleMessageListenerContainer - Restarting Consumer@fba5072: tags=[[amq.ctag-ceBsEnpsb-Rqk7TSFFjExQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,18), conn: Proxy@554d4856 Shared Rabbit Connection: SimpleConnection@388a6cb2 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 63108], acknowledgeMode=AUTO local queue size=0

这和rabbit的消息确认机制有关系：

消息确认即Message acknowledgment。这是一种机制。其目的是用于保证消息不会丢失。

在上面的例子程序中，当rabbitmq向消费者发送消息之后，它就会将这条消息从rabbitmq当中删除。这里的问题在于虽然向消费者发送了这条消息，但是我们却无从知晓消费者是否真的收到了这条消息。

为了确保消息永不丢失，RabbitMQ支持消息确认机制。 消费者在在收到，处理了消息之后发回ack（nowledgement）告诉RabbitMQ，之后RabbitMQ就可以删除它。

如果消费者因为某种原因挂掉（通道关闭，连接关闭或TCP连接丢失）而不发送确认，RabbitMQ将理解为消息未完全处理并会将消息重新排队。 如果同时有其他在线消费者，则会迅速将其重新发送给其他消费者。 这样就可以确保没有消息丢失，即使在消息消费者挂掉的情况下。

在rabbitmq当中没有任何消息超时， 即使处理消息需要非常长的时间，也没关系.

上面配置的消息确认配置为：

acknowledgeMode=AUTO

当节点将消息传递给消费者时，它必须决定是否应将消息视为由消费者处理（或至少已接收）。 由于多种情况（客户端连接，消费者应用程序等）可能会失败，因此此决定是数据安全问题。 消息传递协议通常提供一种确认机制，该机制允许使用者确认向其连接的节点的传递。 是否使用该机制由用户订阅时决定。

rabbitmq提供下面三种消息确认机制：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RvcGRldmVsb3BlcnI_size_16_color_FFFFFF_t_70][]

在auto模式下，容器将根据侦听器是否正常返回或引发异常来发出确认/否定。因此，当连接断开时，消息不会得到确认.这也导致了之后的重新发送.

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RvcGRldmVsb3BlcnI_size_16_color_FFFFFF_t_70]: /images/20210920/7ce899893fe94604a950b274e5b2c744.png