那么这就带来一个问题#xff0c;我们在程序中添加一个中间件#xff0c;把我们的消息依托给MQ#xff0c;那么我们如何保证我们的消息在这个过程不会丢失。

保证它的的可靠性#xff0c;目…上一篇我介绍了RabbitMQ的基本交换机以及队列不了解可以再回顾一下。

那么这就带来一个问题我们在程序中添加一个中间件把我们的消息依托给MQ那么我们如何保证我们的消息在这个过程不会丢失。

保证它的的可靠性目前主流从三个角度保证生产者可靠性MQ可靠性消费者可靠性。

所谓生产者可靠性就是我们如何保证将我们的消息从生产者将消息发送到队列或者消息的可靠性

重试机制

针对网络异常连接中断卡顿的自动重发。

(萌芽阶段与客户端的连接还没有到交换机/d队列)

spring:rabbitmq:/....connection-timeout:

template:retry:initial-interval:

1000msmultiplier:

connetion-timeout连接时长如果在这个时长内依旧没有连接成功就认为连接失败initial-interval等待时长第一次连接失败后多久重新重试mutiplier等待倍数结合initial-interval计算下次等待时长,initial-intervalinitial-interval*mutipliermax-attempts最大重连次数

重连机制似乎看起来我们每个程序都需要配置一下但实际上我们真是企业还是要慎重因为我们的不断重连是阻塞式的。

所以说在追求极度的高并发情况下需要慎重使用如果必须使用我们应该优化参数值比如连接次数连接时长等参数值调整。

回调机制

确认消息是否到投递到交换机以及是都正确路由(成长阶段到达交换机/队列)

Confirm

spring:rabbitmq:/...publisher-confirm-type:

参数解释

退回消息的机制none关闭confirm机制simple同步阻塞等待MQ回执消息correlatedMQ异步回调方式返回回执消息

//创建关联数据

CorrelationData(UUID.randomUUID().toString());

使用Lambda设置确认回调

rabbitTemplate.setMandatory(true);

rabbitTemplate.setConfirmCallback((correlationData1,

ack,

spring:rabbitmq:/...publisher-returns:

truerabbitTemplate.setMandatory(true);

rabbitTemplate.setReturnsCallback(returnedMessage

接受一个参数

在{}体内写方法体System.err.println(消息投递失败);//...处理逻辑/});

交换机持久化

MessageBuilder.withBody(hello.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)

消息的持久化影响了我的MQ的吞吐量同时保证了消息的可靠性。

为了平衡可靠性性和性能

RabbitMQ从

如果我们的消息是持久化的消息依旧会先存放在内存中当我们的信息量达到一定程度后服务器page

out将消息扇出到磁盘中。

这个时刻队列暂时无法接受消息处于阻塞状态。

queue先将消息写入磁盘然后根据消费者的读取速速将消息扇出到内存中(最多2048条)。

在RabbitMQ3.8版本及以下如若使用多队列需要手动配置。

RabbitMQ3.12