企业网站建设属于什么科目,石家庄优化哪家好,石家庄自己的网站,线上广告代理平台目录
1.如何保证消息的可靠性
1.1.消息的可靠投递
confirm机制
return机制
1.2.如何保证消息在队列中不丢失
1.3.确保消息能可靠的被消费掉
2.延迟队列
2.1.TTL
2.2.死信队列
2.3.延迟队列
3.如何防止消费者重复消费消息 1.如何保证消息的可靠性
1.1.消息的可靠投递…目录
1.如何保证消息的可靠性
1.1.消息的可靠投递
confirm机制
return机制
1.2.如何保证消息在队列中不丢失
1.3.确保消息能可靠的被消费掉
2.延迟队列
2.1.TTL
2.2.死信队列
2.3.延迟队列
3.如何防止消费者重复消费消息 1.如何保证消息的可靠性
1.1.消息的可靠投递 在生产环境中由于一些不明原因导致 rabbitmq 重启在 RabbitMQ 重启期间生产者消息投递失败导致消息丢失需要手动处理和恢复。于是我们开始思考如何才能进行 RabbitMQ 的消息可靠投递呢特别是在这样比较极端的情况RabbitMQ 集群不可用的时候无法投递的消息该如何处理呢 在使用 RabbitMQ 的时候作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。 confirm 确认模式return 退回模式消息从 producer 到 exchange 则会返回一个 confirmCallback 。消息从 exchange--queue 投递失败则会返回一个 returnCallback 。 默认rabbitmq不开启上面两种模式 我们将利用这两个 callback 控制消息的可靠性投递 confirm和return的实现 设置ConnectionFactory的publisher-confirm-type: correlated开启 确认模式。 使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack如果为true则发送成功如果为false则发送失败需要处理。 设置ConnectionFactory的publisher-returnstrue 开启 退回模式。 使用rabbitTemplate.setReturnCallback设置退回函数当消息从exchange路由到queue失败后执行回调函数returnedMessage。 confirm机制 演示:4.springboot整合RabbitMQ (1).配置文件中开启confirm #开启confirm确认机制
spring.rabbitmq.publisher-confirm-typecorrelated (2)设置rabbitTemplate的confirmCallback回调函数 /*** 使用confirm机制* 1需要开启confirm机制。-----配置文件中加入spring.rabbitmq.publisher-confirm-typecorrelated* (2)为rabbitTemplate指定setConfirmCallback回调函数*/Testvoid test001() {//只能保证消息从生产者到交换机的可靠性rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {//触发该方法if (ack false) {System.out.println(未来根据项目的需要完成相应的操作);}}});rabbitTemplate.convertAndSend(Topics-exchange, lazy.aaa, Hello RabbitMQ...);}return机制 (1).配置文件中开启return #开启return机制
spring.rabbitmq.publisher-returnstrue (2)设置rabbitTemplate的return回调函数 /*** return机制* 1开启return机制---配置文件加入spring.rabbitmq.publisher-returnstrue* 2为rabbitTemplate设置return的回调函数*/Testvoid test002() {//只有当消息无法从交换机到队列时才会触发rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {//为交换机到队列分发消息失败时会触发System.out.println(replyCode replyCode);System.out.println(replyText replyText);}});rabbitTemplate.convertAndSend(Topics-exchange, lazy.aaa, Hello RabbitMQ...);} 1.2.如何保证消息在队列中不丢失 (1)设置队列为持久化 (2)设置消息的持久化 1.3.确保消息能可靠的被消费掉 ACK确认机制 多个消费者同时收取消息收取消息到一半突然某个消费者挂掉要保证此条消息不丢失就需要acknowledgement机制就是消费者消费完要通知服务端服务端才将数据删除 这样就解决了即使一个消费者出了问题没有同步消息给服务端还有其他的消费端去消费保证了消息不丢的case。 ACK的实现 ack指Acknowledge确认。 表示消费端收到消息后的确认方式。 有三种确认方式 自动确认acknowledgenone手动确认acknowledgemanual根据异常情况确认acknowledgeauto这种方式使用麻烦并且不常用 其中自动确认是指当消息一旦被Consumer接收到则自动确认收到并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中很可能消息接收到业务处理出现异常那么该消息就会丢失。 如果设置了手动确认方式则需要在业务处理成功后调用channel.basicAck()手动签收如果出现异常则调用channel.basicNack()方法让其自动重新发送消息。 消费端: (1).修改消费端----手动确认消息 #修改消费端----手动确认消息
spring.rabbitmq.listener.simple.acknowledge-modemanual (2).修改代码 package com.wqg.listener;import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.amqp.core.Message;import java.io.IOException;/*** fileName:MyListener* description:* author:wqg* createTime:2023/7/12 18:57*/
Component
public class MyListener {//basicAck:确认消息----rabbit服务端删除//basicNack服务继续发送消息RabbitListener(queues {Topics-queue002})//queues:表示你监听的队列名public void h(Message message , Channel channel) throws IOException {long deliveryTag message.getMessageProperties().getDeliveryTag();//把监听到的消息封装到Message类对象中byte[] body message.getBody();String s new String(body);System.out.println(消息内容s);try {//int a 10/0; //模拟宕机System.out.println(核心业务的处理~~~);/***long deliveryTag 消息的标注* boolean multiple是否把该消息之前未确认的消息一起确认掉*/channel.basicAck(deliveryTag,true);//确认消息} catch (Exception e) {/*** long deliveryTag; boolean multiple;* boolean requeue是否要求rabbitmq服务重新发送该消息*/channel.basicNack(deliveryTag,true,true);}}
}总结: 如何保证消息的可靠性? 保证消息的可靠性投递: confirm机制和return机制队列中:---持久化使用ack机制保证消费者的可靠性消费。 2.延迟队列
2.1.TTL TTL 全称 Time To Live存活时间/过期时间。 当消息到达存活时间后还没有被消费会被自动清除。 RabbitMQ可以对消息设置过期时间也可以对整个队列Queue设置过期时间。 演示 使用图形化界面创建 生产者测试 结果 队列设置了过期时间而消息也设置了过期时间-----按照时间短的执行 小结: 设置队列过期时间使用参数x-message-ttl单位ms(毫秒)会对整个队列消息统一过期。设置消息过期时间使用参数expiration。单位ms(毫秒)当该消息在队列头部时消费时会单独判断这一消息是否过期。如果两者都进行了设置以时间短的为准。 2.2.死信队列 死信队列英文缩写DLX 。Dead Letter Exchange死信交换机当消息成为Dead message后可以被重新发送到另一个交换机这个交换机就是DLX。 什么样的消息会成为死信消息 队列消息长度到达限制消费者拒接消费消息basicNack/basicReject,并且不把消息重新放入原目标队列,requeuefalse原队列存在消息过期设置消息到达超时时间未被消费 队列绑定死信交换机 演示 使用图形化界面创建 生产者测试 结果 2.3.延迟队列 延迟队列即消息进入队列后不会立即被消费只有到达指定时间后才会被消费。 需求 下单后30分钟未支付取消订单回滚库存。 新用户注册成功7天后发送短信问候。 实现方式 定时器性能差---每隔一段时间要进行数据库查询。 延迟队列 通过消息队列完成延迟队列的功能 在RabbitMQ中并未提供延迟队列功能。但是可以使用TTL死信队列 组合实现延迟队列的效果。 演示 队列为空 创建springboot项目 配置文件
#rabbitmq的配置
spring.rabbitmq.host192.168.75.129
spring.rabbitmq.usernameguest
spring.rabbitmq.passwordguest
spring.rabbitmq.virtual-host/ (1). 引入依赖 dependenciesdependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.83/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-rabbit-test/artifactIdscopetest/scope/dependency/dependencies (2).创建OrderController.java模拟订单 package com.wqg.controller;import com.alibaba.fastjson.JSON;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.UUID;/*** fileName:OrderController* description:订单* author:wqg* createTime:2023/7/13 16:24*/
RestController
RequestMapping(/order)
public class OrderController {Autowiredprivate RabbitTemplate rabbitTemplate;GetMapping(/saveOrder)public String save(Integer pid, Integer num) {//生成一个订单号String orderId UUID.randomUUID().toString().replace(-, );System.out.println(下单成功订单号为 orderId);HashMapObject, Object map new HashMap();map.put(orderId, orderId);map.put(pid, pid);map.put(num, num);rabbitTemplate.convertAndSend(pt_exchange, qy165.aaa, JSON.toJSONString(map));return 下单成功;}
}(3).创建MyListener.java模拟监听 package com.wqg.listener;import com.alibaba.fastjson.JSON;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.HashMap;/*** fileName:MyListener* description:* author:wqg* createTime:2023/7/13 16:35*/
Component
public class MyListener {RabbitListener(queues {dead_queue})public void hello(Message message){byte[] body message.getBody();String s new String(body);HashMap hashMap JSON.parseObject(s, HashMap.class);System.out.println(messagehashMap);//取消订单System.out.println(取消订单号为hashMap.get(orderId));}
}测试 控制台 3.如何防止消费者重复消费消息 消息的幂等性---无论操作几次结果都是一样。 生成全局id存入redis或者数据库在消费者消费消息之前查询一下该消息是否有消费过。如果该消息已经消费过则告诉mq消息已经消费将该消息丢弃手动ack。如果没有消费过将该消息进行消费并将消费记录写进redis或者数据库中。 简单描述一下需求如果订单完成之后需要为用户累加积分又需要保证积分不会重复累加。那么再mq消费消息之前先去数据库查询该消息是否已经消费如果已经消费那么直接丢弃消息。 演示 生产者 import com.alibaba.fastjson.JSONObject;
import com.xiaojie.score.entity.Score;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;import java.util.UUID;/*** author * version 1.0* description:发送积分消息的生产者* date*/
Component
Slf4j
public class ScoreProducer implements RabbitTemplate.ConfirmCallback {Autowiredprivate RabbitTemplate rabbitTemplate;//定义交换机private static final String SCORE_EXCHANGE ykq_score_exchaneg;//定义路由键private static final String SCORE_ROUTINNGKEY score.add;/*** description: 订单完成* param:* return: java.lang.String* author xiaojie* date: */public String completeOrder() {String orderId UUID.randomUUID().toString();System.out.println(订单已完成);//发送积分通知Score score new Score();score.setScore(100);score.setOrderId(orderId);String jsonMSg JSONObject.toJSONString(score);sendScoreMsg(jsonMSg, orderId);return orderId;}/*** description: 发送积分消息* param:* param: message* param: orderId* return: void* author * date:*/Asyncpublic void sendScoreMsg(String jsonMSg, String orderId) {this.rabbitTemplate.setConfirmCallback(this);rabbitTemplate.convertAndSend(SCORE_EXCHANGE, SCORE_ROUTINNGKEY, jsonMSg, message - {//设置消息的id为唯一message.getMessageProperties().setMessageId(orderId);return message;});}Overridepublic void confirm(CorrelationData correlationData, boolean ack, String s) {if (ack) {log.info(消息发送成功:correlationData:{},ack:{},s:{}, correlationData, ack, s);} else {log.info(消息发送失败{}, ack);}}
} 消费者 import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.score.entity.Score;
import com.xiaojie.score.mapper.ScoreMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.Map;/*** author * version 1.0* description: 积分的消费者* date */
Component
Slf4j
public class ScoreConsumer {Autowiredprivate ScoreMapper scoreMapper;RabbitListener(queues {ykq_score_queue})public void onMessage(Message message, Headers MapString, Object headers, Channel channel) throws IOException {String orderId message.getMessageProperties().getMessageId();if (StringUtils.isBlank(orderId)) {return;}log.info(消息id是{}, orderId);String msg new String(message.getBody());Score score JSONObject.parseObject(msg, Score.class);if (score null) {return;}//执行前去数据库查询是否存在该数据存在说明已经消费成功不存在就去添加数据添加成功丢弃消息Score dbScore scoreMapper.selectByOrderId(orderId);if (dbScore ! null) {//证明已经消费消息告诉mq已经消费,丢弃消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return;}Integer result scoreMapper.save(score);if (result 0) {//积分已经累加删除消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return;} else {log.info(消费失败采取相应的人工补偿);} }
}
