网站导航
> 文章列表 > RabbitMQ防止消息丢失

RabbitMQ防止消息丢失

网友: 文章列表 2024-03-22 04:49:36

RabbitMQ防止消息丢失

生产者没有成功把消息发送到MQ
丢失的原因 ：因为网络传输的不稳定性，当生产者在向MQ发送消息的过程中，MQ没有成功接收到消息，但是生产者却以为MQ成功接收到了消息，不会再次重复发送该消息，从而导致消息的丢失。

解决办法 ： 有两个解决办法：事务机制和confirm机制，最常用的是confirm机制（发布确认机制）。

注意：

          RabbitMQ的事务机制是同步的，很耗型能，会降低RabbitMQ的吞吐量。

          confirm机制是异步的，生成者发送完一个消息之后，不需要等待RabbitMQ的回调，就可以发送下一个消息，当RabbitMQ成功接收到消息之后会自动异步的回调生产者的一个接口返回成功与否的消息。

两个机制说明如下:

confirm（发布确认）机制
解释：RabbitMQ可以开启 confirm 模式，在生产者那里设置开启 confirm 模式之后，生产者每次写的消息都会分配一个唯一的 id，如果消息成功写入 RabbitMQ 中，RabbitMQ 会给生产者回传一个 ack 消息，告诉你说这个消息 ok 了。如果 RabbitMQ 没能处理这个消息，会回调你的一个 nack 接口，告诉你这个消息接收失败，生产者可以重新发送。而且你可以结合这个机制自己在内存里维护每个消息 id 的状态，如果超过一定时间还没接收到这个消息的回调，那么可以重发。

代码

yml配置

----------------------------------------------------------------------------------------------------

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
/
* 交换机回滚
*/
@Component
@Slf4j
public class ExchangeCallback implements RabbitTemplate.ConfirmCallback{/* correlationData 内含消息内容* ack 交换机接受成功或者失败。 true表示交换机接受消息成功， false表示交换机接受失败* cause 表示失败原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("hello world");String id = correlationData.getId();String message = new String(correlationData.getReturnedMessage().getBody());if (ack){log.info("交换机收到消息id为{}, 消息内容为{}", id, message);}else {log.info("交换机未收到消息id为{}, 消息内容为{}, 原因为{}", id, message, cause);}}
}

----------------------------------------队列防止消息丢失----------------------------------------------------------------

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/* 队列防止消息丢失*/
@Slf4j
@Component
public class QueueCallback implements RabbitTemplate.ReturnCallback{@Overridepublic void returnedMessage(Message message,int replyCode, String replyText, String exchange, String routingKey) {log.info("消息 {} 经交换机 {} 通过routingKey={} 路由到队列失败，失败code为：{}， 失败原因为：{}",new String(message.getBody()), exchange, routingKey, replyCode, replyText);}
}

--------------------------引用->controller-----------------------------------------------

//交换机回滚
@Autowired
private ExchangeCallback exchangeCallback;
//队列回滚
@Autowired
private QueueCallback queueCallback;
/* 初始化交换机监听*/
@PostConstruct
public void init(){  
//交换机
rabbitTemplate.setConfirmCallback(exchangeCallback);
/* true：交换机无法将消息进行路由时，会将该消息返回给生产者* false：如果发现消息无法进行路由，则直接丢弃*/
rabbitTemplate.setMandatory(true);
//队列
rabbitTemplate.setReturnCallback(queueCallback);
}
/* 发送消息* 结果:"这是一条消息"*/@GetMapping("/sendMessageTest")public String sendMessageTest(){// 消息类型为object 发送对象也是可以的String msg = "这是一条消息";// 第一个参数为发送消息到那个交换机上，第二个是发送的路由键（交换机进行需要符合绑定的队列），第三个参数为发送的消息
//CommonUtils.dirExchange--自己的交换机名称
//CommonUtils.routingKey --路由Key值 rabbitTemplate.convertAndSend("1235",CommonUtils.routingKey,msg);System.out.println("消息发送成功:"+msg);return "发送成功;发送内容为:"+msg;}

运行结果:

消息接收确认

消息通过 ACK 确认是否被正确接收，每个 Message 都要被确认（acknowledged），可以手动去 ACK 或自动 ACK

消息确认模式有：

  • AcknowledgeMode.NONE：自动确认
  • AcknowledgeMode.AUTO：根据情况确认（默认值）
  • AcknowledgeMode.MANUAL：手动确认

手动确认消息

1、basicAck

表示成功确认，使用此回执方法后，消息会被rabbitmq broker 删除。

void basicAck(long deliveryTag, boolean multiple)

  • deliveryTag：表示消息投递序号，每次消费消息或者消息重新投递后，deliveryTag都会增加。deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，RabbitMQ 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag， 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery tag 的范围仅限于 Channel

  • multiple：是否批量确认，值为 true 则会一次性 ack所有小于当前消息 deliveryTag 的消息。

  • 举个栗子： 假设我先发送三条消息deliveryTag分别是5、6、7，可它们都没有被确认，当我发第四条消息此时deliveryTag为8，multiple设置为 true，会将5、6、7、8的消息全部进行确认。

2、basicNack

表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列。

void basicNack(long deliveryTag, boolean multiple, boolean requeue)

  • deliveryTag：表示消息投递序号。
  • multiple：是否批量确认。
  • requeue：值为 true 消息将重新入队列。

3、basicReject

拒绝消息，与basicNack区别在于不能进行批量操作，其他用法很相似。

void basicReject(long deliveryTag, boolean requeue)

  • deliveryTag：表示消息投递序号。
  • requeue：值为 true 消息将重新入队列。
网络标签：

相关问题