当前位置：首页 > wzjs >正文

cms进行网站开发宁波seo网络推广软件系统

wzjs 2025/8/28 20:17:37

cms进行网站开发,宁波seo网络推广软件系统,杭州市建设部门网站,双语网站怎么做在分布式系统中，确保消息的可靠传输是至关重要的。无论是处理金融交易、用户互动还是后台数据同步，丢失或重复的消息都可能导致严重的问题。为了增强系统的可靠性，我们可以依赖于生产者的重试机制和生产者确认机制。本文将探讨这两种机制如何…

在分布式系统中，确保消息的可靠传输是至关重要的。无论是处理金融交易、用户互动还是后台数据同步，丢失或重复的消息都可能导致严重的问题。为了增强系统的可靠性，我们可以依赖于生产者的重试机制和生产者确认机制。本文将探讨这两种机制如何共同作用来提高发送者的可靠性。

生产者重试机制

生产者确认机制

实现生产者确认

对应的模块添加配置

定义ReturnCallback

定义ConfirmCallback

生产者确认实现效果

生产者重试机制

问题：网络故障，导致与MQ的连接中断。

解决方案：SpringAMQP提供的消息发送时的重试机制。

即：当RabbitTemplate与MQ连接超时后，多次重试。

方案实现如下：

修改publisher模块的application.yml文件，添加如下配置：

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multipliermax-attempts: 3 # 最大重试次数

测试：（故意停掉RabbitMQ服务，验证方案可行性）

效果如下：

解决方案验证成功

生产者确认机制

问题：消息发送到MQ之后丢失的现象。

描述：

MQ内部处理消息的进程发生了异常

生产者发送消息到达MQ后未找到Exchange(往往是程序员编码出问题导致)

生产者发送消息到达MQ的Exchange后，未找到合适的Queue，因此无法路由(往往是程序员编码出问题导致)

解决方案：RabbitMQ提供了生产者消息确认机制。

即：Publisher Confirm和Publisher Return两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。

如图所示：

图像解读：

publisher --- exchange2：当消息投递到MQ，但是路由失败时，返回异常信息，同时返回ACK(投递成功)

publisher --- exchange：临时消息投递到了MQ，并且入队成功，返回ACK（投递成功）

publisher --- exchange：持久消息投递到了MQ，并且入队完成持久化，返回ACK （投递成功）

其它情况都会返回NACK（投递失败）

实现生产者确认

对应的模块添加配置

在publisher模块的application.yml中添加配置：

spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型publisher-returns: true # 开启publisher return机制

publisher-confirm-type有三种模式，推荐使用correlated（回调机制）

定义ReturnCallback

每个RabbitTemplate只能配置一个ReturnCallback。

我们在publisher模块定义MqConfig配置类,代码如下：

package com.itheima.publisher.config;import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;import javax.annotation.PostConstruct;@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {private final RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {log.error("触发return callback,");log.debug("exchange: {}", returned.getExchange());log.debug("routingKey: {}", returned.getRoutingKey());log.debug("message: {}", returned.getMessage());log.debug("replyCode: {}", returned.getReplyCode());log.debug("replyText: {}", returned.getReplyText());}});}
}

定义ConfirmCallback

由于每个消息发送时的处理逻辑不一定相同，因此ConfirmCallback需要在每次发消息时定义。具体来说，是在调用RabbitTemplate中的convertAndSend方法时，多传递一个参数,代码如下：

@Test
void testPublisherConfirm() {// 1.创建CorrelationDataCorrelationData cd = new CorrelationData();// 2.给Future添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {// 2.1.Future发生异常时的处理逻辑，基本不会触发log.error("send message fail", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执log.debug("发送消息成功，收到 ack!");}else{ // result.getReason()，String类型，返回nack时的异常描述log.error("发送消息失败，收到 nack, reason : {}", result.getReason());}}});// 3.发送消息rabbitTemplate.convertAndSend("user.direct", "q", "hello", cd);
}