做直播网站需要手续,wordpress修改管理密码,中国万网张向东去哪了,济南装饰行业网站建设1.业务分析 在业务的开发中#xff0c;我们通常将业务的非核心业务交给MQ来处理#xff0c;比如支付#xff0c;在支付过后#xff0c;我们需要扣减余额#xff0c;修改支付单状态#xff0c;修改订单状态#xff0c;发送短信提醒用户#xff0c;给用户增加积分等等我们通常将业务的非核心业务交给MQ来处理比如支付在支付过后我们需要扣减余额修改支付单状态修改订单状态发送短信提醒用户给用户增加积分等等可能真是场景并非这么简单这里举个例子在这套业务中修改订单状态发送短信提醒用户给用户增加积分这三个业务在特定场景下并非是核心业务所以把他放在MQ消息队列中进行处理在非核心业务执行的时候可能出现多个问题导致数据不一致比如用户买了东西发现自己的钱已经扣了但是页面显示的还是未支付状态这种情况就非常严重了造成这个现象的原因可能有多种比如网络丢包发布消息者挂了MQ挂了消费者挂了等等可能有硬件层面也可能有软件层面甚至是网络层面对于这一系列问题我们应该尽量的保证消息的可靠性让数据一致性得到保证接下来从三个方面进行分析以及提出解决方案不过具体还是得看业务需求是否需要数据的强一致性。 2.发送者的可靠性 2-1发送者重连 消息的发布有三个角色发布者MQ消费者在发消息的时候需要和MQ进行连接这个连接时一个网络连接如果因为网络问题消息发送失败可能会导致数据不一致产生。
解决方案在发布者的application.yaml中配置
spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数下次等待时长 initial-interval * multiplier如果是2第一此初始化1秒2秒4秒以此类推max-attempts: 3 # 最大重试次数 注意当网络不稳定的时候利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试也就是说多次重试等待的过程中当前线程是被阻塞的会影响业务性能。如果对于业务性能有要求建议禁用重试机制。如果一定要使用请合理配置等待时长和重试次数当然也可以考虑使用异步线程来执行发送消息的代码。 2-2发送者确认机制
SpringAMQP提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后当发送者发送消息给MQ后MQ会返回确认结果给发送者发送者再证实这个返回结果。返回的结果有以下几种情况:
消息投递到了MQ但是路由失败。此时会通过PublisherReturn返回路由异常原因然后返回ACK告知投递成功临时消息投递到了MQ并且入队成功返回ACK告知投递成功持久消息投递到了MQ并且入队完成持久化返回ACK告知投递成功其它情况都会返回NACK告知投递失败 返回ACK不需要重发NACK需要重发
在publisher模块的application.yaml中添加配置
spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制并设置confirm类型publisher-returns: true # 开启publisher return机制
这里 publisher-confirm-type 有三种模式可选
none : 关闭confirm机制simple:同步阻塞等待MQ的回执correlated:MQ异步回调返回回执
一般推荐使用 correlated回调机制。
每个RabbitTemplate只能配置一个ReturnCallback因此我们可以在配置类中统一设置。我们在publisher模块定义一个配置类
Configuration
RequiredArgsConstructor
Slf4j
public class MqConfig {private final RabbitTemplate rabbitTemplate;PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error(触发returnCallback);log.debug(exchange:{},returnedMessage.getExchange());log.debug(message:{},returnedMessage.getMessage());log.debug(routingKey:{},returnedMessage.getRoutingKey());log.debug(replyCode:{},returnedMessage.getReplyCode());log.debug(replyText:{},returnedMessage.getReplyText());}});}
}
由于每个消息发送时的处理逻辑不一定相同因此ConfirmCallback需要在每次发消息时定义。具体来说是在调用RabbitTemplate中的convertAndSend方法时多传递一个参数
RabbitListener(bindings QueueBinding(value Queue(name simple2.queue),exchange Exchange(name simple2.direct),key simple2))public void testPublisherConfirmRule() {CorrelationData cd new CorrelationData(UUID.randomUUID().toString());cd.getFuture().addCallback(new ListenableFutureCallbackCorrelationData.Confirm() {Overridepublic void onFailure(Throwable ex) {// Future发生异常时的处理逻辑基本不会触发log.error(Future发生异常时的处理逻辑, ex);}Overridepublic void onSuccess(CorrelationData.Confirm result) {if (result.isAck()){log.debug(发送消息成功收到ACK);}else {log.error(发送消息失败收到NACK);//TODO: 重发消息}}});//交换机名称String exchange simple2.direct;//消息String message hello mq;//发送消息rabbitTemplate.convertAndSend(exchange, simple2, message, cd);}
3.MQ的可靠性 在默认情况下RabbitMQ会将接受到的消息保存在内存中以降低消息收发的延迟。这样会导致两个问题
一旦MQ宕机内存中的消息会丢失内存空间有限当消费者故障或处理过慢时会导致消息积压引发MQ阻塞
方案一数据持久化
数据持久化包括三个方面交换机持久化队列持久化消息持久化
SpringAMQP默认生成的交换机和队列以及发消息都是持久化的
方案二lazy queue既能保证并发能力又不用写内存 从RabbitMQ的3.6.0版本开始就增加了LazyQueue的概念也就是惰性队列。惰性队列的特征如下:
接收到消息后直接存入磁盘不再存储到内存消费者要消费消息时才会从磁盘中读取并加载到内存(可以提前缓存部分消息到内存最多2048条)在3.12版本后所有队列都是LazyQueue模式无法更改。
那如何把queue变成lazy queue可以基于声明bean的形式也可以通过注解的方式
Bean
public Queue lazyQueue(){return QueueBuilder.durable(lazy.queue).lazy() // 开启Lazy模式.build();
}
RabbitListener(queuesToDeclare Queue(name lazy.queue,durable true,arguments Argument(name x-queue-mode, value lazy) //这两个是固定的
))
public void listenLazyQueue(String msg){log.info(接收到 lazy.queue的消息{}, msg);
}
开启持久化和生产者确认时RabbitMQ只有在消息持久化完成后才会给生产者返回ACK回执
4.消费者的可靠性 4-1消费者确认机制 消费者确认机制时为了确认消费者是否成功处理消息。当消费者处理消息结束后应该向RabbitMQ发送一个回执告知RabbitMQ自己消息的处理状态。状态有三种
ack成功的处理了消息RabbitMQ队列中删除这个消息nack消息处理失败RabbitMQ就会再次给消费者投递消息持续投递reject消息处理失败RabbitMQ队列中删除这个消息一般是消息内容有问题所以拒绝 SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式有三种方式:
none:不处理。即消息投递给消费者后立刻ack消息会立刻从MQ删除。非常不安全不建议使用manual:手动模式。需要自己在业务代码中调用api发送ack或reject存在业务入侵但更灵活auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强当业务正常执行时则自动返回ack.当业务出现异常时根据异常判断返回不同结果:
如果是业务异常会自动返回nack如果是消息处理或校验异常自动返回reject
需要配置在消费的application.yaml中
spring:rabbitmq:listener:simple:acknowledge-mode: auto # 自动ack
4-2失败重试机制 SpringAMQP提供了消费者失败重试机制在消费者出现异常时利用本地重试而不是无限的requeue到mq。通过在消费者的application.yaml文件中添加配置来开启重试机制
spring:rabbitmq:listener: # 这里注意区别发布者有个失败重连机制和这个配置很像simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false 在开启重试模式后重试次数耗尽如果消息依然失败则需要有MessageRecoverer接口来处理它包含三种不同的实现
RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机
第三种示例图 1在consumer服务中定义处理失败消息的交换机和队列
Bean
public DirectExchange errorMessageExchange(){return new DirectExchange(error.direct);
}
Bean
public Queue errorQueue(){return new Queue(error.queue, true);
}
Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);
}
2定义一个RepublishMessageRecoverer关联队列和交换机
Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);
}
完整代码
Configuration
ConditionalOnProperty(name spring.rabbitmq.listener.simple.retry.enabled, havingValue true)
public class ErrorMessageConfig {Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange(error.direct);}Beanpublic Queue errorQueue(){return new Queue(error.queue, true);}Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);}Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error); //第二个参数是routingKey}
}
4-3幂等性业务 在程序开发中是指同一个业务执行一次或多次对业务状态的运行是一致的。 如果网络问题出现故障有可能出现把一个业务做了多次比如扣减库存总不能扣减两次吧于是可以采取对每一个消息指定一个消息idid值唯一然后执行完这个消息后把这个消息id存到数据库里然后每次执行消息的时候可以去数据库查一下有没有这个消息如果有就代表这个消息之前执行过了于是就不对这个消息做处理。 在做注册消息转换器为bean的时候可以设置消息的id然后我们接受消息的时候用Message来接就可以
Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jjmc new Jackson2JsonMessageConverter();// 2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息jjmc.setCreateMessageIds(true);return jjmc;
} 用Message接然后message.getMessageProperties().getMessageId() 获取id那这个id做业务判断就可以了。
以上是对于非幂等业务的一种方案但明显这种方案不太好。影响mq性能。
另一种就是基于具体业务逻辑来进行判断来实现业务的幂等比如我在一个业务执行前先判断这个业务的一个状态如果状态以及修改过了我直接不做处理就行了如果没有我在进行修改。
5.延迟消息 5-1死信交换机
当一个队列中的消息满足下列情况之一时可以成为死信dead letter 消费者使用basic.reject或 basic.nack声明消费失败并且消息的requeue参数设置为false 消息是一个过期消息超时无人消费 要投递的队列消息满了无法投递 如果一个队列中的消息已经成为死信并且这个队列通过dead-letter-exchange属性指定了一个交换机那么队列中的死信就会投递到这个交换机中而这个交换机就称为死信交换机Dead Letter Exchange。而此时加入有队列与死信交换机绑定则最终死信就会被投递到这个队列中。 死信交换机有什么作用呢 收集那些因处理失败而被拒绝的消息 收集那些因队列满了而被拒绝的消息 收集因TTL有效期到期的消息
声明正常队列和正常交换机和的时候使用bean方式因为正常队列要指定死信交换机 Beanpublic Queue normalQueue(){return QueueBuilder.durable(normal.queue).deadLetterExchange(dlx.direct).build();}
发送者示例代码给正常队列设置一个TTL如果时间到了把消息给死信交换机模拟延迟消息
rabbitTemplate.convertAndSend(normal.direct, hi, hello, new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration(1000000); return message;}});
5-2延迟消息插件
先去社区下载 https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
去查看RabbitMQ的插件目录对应的数据卷。
docker volume inspect mq-plugins
然后进入插件那么目录
docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange
这样插件就装好了
然后声明一个延迟交换机基于注解方式
RabbitListener(bindings QueueBinding(value Queue(name delay.queue, durable true),exchange Exchange(name delay.direct, delayed true),key delay
))
public void listenDelayMessage(String msg){log.info(接收到delay.queue的延迟消息{}, msg);
}
基于bean方式
Bean
public DirectExchange delayExchange(){return ExchangeBuilder.directExchange(delay.direct) // 指定交换机类型和名称.delayed() // 设置delay的属性为true.durable(true) // 持久化.build();
}
发送延迟消息
Test
void testPublisherDelayMessage() {// 1.创建消息String message hello, delayed message;// 2.发送消息利用消息后置处理器添加消息头rabbitTemplate.convertAndSend(delay.direct, delay, message, new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 添加延迟消息属性message.getMessageProperties().setDelay(5000);return message;}});
}
注意
延迟消息插件内部会维护一个本地数据库表同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长可能会导致堆积的延迟消息非常多会带来较大的CPU开销同时延迟消息的时间会存在误差。
因此不建议设置延迟时间过长的延迟消息。
