如何做网站推广方法,wordpress设置标题字体,pmp,为什么网站不需要icp备案微服务-高级篇 一.微服务保护1.初识Sentinel2.微服务整合Sentinel3.限流4.隔离和降级5.授权规则6.规则管理模式 二、分布式事务1.什么是分布式事务#xff1f;2.理论基础3.部署与集成Seata4.Seata的四种模式5.Seata高可用 三、分布式缓存1.Redis持久化2.搭建主从架构与哨兵模式… 微服务-高级篇 一.微服务保护1.初识Sentinel2.微服务整合Sentinel3.限流4.隔离和降级5.授权规则6.规则管理模式 二、分布式事务1.什么是分布式事务2.理论基础3.部署与集成Seata4.Seata的四种模式5.Seata高可用 三、分布式缓存1.Redis持久化2.搭建主从架构与哨兵模式3.RedisTemplate的哨兵模式4.Redis分片集群数据迁移 5.RedisTemplate访问分片集群 四、多级缓存1.多级缓存的意义及解决方案2.进程缓存-caffeine3.lua语言4.初识OpenResty5.获取请求参数6.查询Tomcat7.查询Redis缓存8.添加nginx本地缓存9.缓存同步 五、服务异步通讯-rabbitMQ的高级特性1.消息可靠性2.消息持久化3.消费者消息确认4.失败重试机制5.如何确保RabbitMQ消息的可靠性6.死信交换机7.延迟队列8.惰性队列7.延迟队列8.惰性队列 一.微服务保护
1.初识Sentinel
雪崩问题及解决方案 微服务调用链路中的某个服务故障引起整个链路中的所有微服务都不可用这就是雪崩。 解决方案 超时处理设置超时时间请求超过一定时间没有响应就返回错误信息不会无休止等待舱壁模式限定每个业务能使用的线程数避免耗尽整个tomcat的资源因此也叫线程隔离熔断降级由断路器统计业务执行的异常比例如果超出阈值会熔断该业务拦截该业务的一切请求流量控制限制业务访问的QPS避免服务因流量的突增而故障
2.微服务整合Sentinel 引入sentinel依赖 dependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-alibaba-sentinel/artifactId
/dependency配置控制台地址 spring:cloud:sentinel:transport:dashboard: localhost:80803.限流 簇点链路 就是项目内的调用链路链路中被监控的每个接口就是一个资源默认情况下sentinel会监控springMVC的每一个端点因此springMVC的每一个端点就是调用链路的一个资源 流控模式有哪些 直接对当前资源限流关联高优先级资源触发阈值对低优先级资源限流链路阈值统计时只统计从指定资源进入当前资源的请求是对请求来源的限流 sentinel默认只标记Controller中的方法资源要标记其他方法需要使用SentinelResource注解 关闭将Controller方法做两content整合,否则链路模式失效 spring:cloud:sentinel:transport:dashboard: localhost:8080web-context-unify: false流控效果 快速失败达到阈值后新的请求会被立即拒绝并抛出FlowException异常默认处理方式warm up预热模式对超出阈值的请求同样是拒绝并抛出异常。但这种模式阈值会动态变化从一个较小值逐渐加到最大阈值排队等待让所有的请求按照先后次序排队执行两个请求的间隔不能小于指定时长 热点参数限流 分别统计参数值相同的请求判断是否超过QPS阈值热点参数限流对默认的springMVC资源无效
4.隔离和降级 虽然限流可以避免因高并发而引起的服务故障但服务还会因为其他原因而故障。而要将这些故障控制在一定范围避免雪崩就要靠线程隔离和熔断降级手段了。 Feign整合Sentinel 修改OrderService的application.yml文件开启Feign的Sentinel功能 feign:client:config:default: # 这里是default就是全局配置如果是写服务名则针对某个微服务的配置logger-level: FULL # 日志级别sentinel:enabled: true # 开启Sentinel给FeignClient编写失败后的降级逻辑 方式一FallbackClass无法对远程调用的异常做处理 方式二FallbackFactory可以对远程调用的异常做处理 在feign-api项目中定义类实现FallbackFactory Slf4j
public class UserClientFallbackFactory implements FallbackFactoryUserServiceClient {Overridepublic UserServiceClient create(Throwable throwable) {return new UserServiceClient() {Overridepublic User findById(Long id) {// 记录信息异常log.error(查询用户失败 , throwable);// 根据业务需要返回数据return new User();}};}
}在配置类中注入该回调类 Configuration
public class FeignClientConfiguration {Beanpublic Logger.Level feignLogLevel() {return Logger.Level.FULL;}Beanpublic UserClientFallbackFactory userClientFallbackFactory() {return new UserClientFallbackFactory();}
}在UserserviceClient中指定失败回调 FeignClient(value userservice , fallbackFactory UserClientFallbackFactory.class)
public interface UserServiceClient {GetMapping(user/{id})User findById(PathVariable Long id);
}线程隔离 线程池隔离基于线程池有额外开销但隔离性更强信号量隔离Sentinel默认采用基于技术器模式简单开销小 熔断降级 断路器熔断策略 慢调用业务的响应时长大于指定时长的请求认定为慢调用请求。在指定时间内如果请求数量超过设定的最小数量慢调用比例大于设定的阈值则触发熔断异常比例或异常数统计指定时间内的调用如果调用次数超过指定请求数并且出现异常的比例达到设定的比例阈值或超过指定异常数则触发熔断
5.授权规则 授权规则可以对调用方的来源做控制有白名单和黑名单两种方式 在gateway中添加全局过滤器添加请求头 spring:application:name: gatewaycloud:nacos:server-addr: localhost:8848 # nacos地址gateway:routes: # 网关路由配置- id: user-service # 路由id 自定义uri: lb://userservice # 路由的目标地址 lb是负载均衡predicates: # 路由断言- Path/user/** # 这个是按照路径匹配只要以/user/开头就符合要求- id: order-serviceuri: lb://orderservicepredicates: # 路由断言- Path/order/** # 这个是按照路径匹配只要以/order/开头就符合要求filters:- AddRequestHeadercolor, bluedefault-filters:- AddRequestHeaderorigin, gateway Sentinel通过RequestOriginParser这个接口的parseOrigin来获取请求的来源 Component
public class HeaderOriginParser implements RequestOriginParser {Overridepublic String parseOrigin(HttpServletRequest request) {String origin request.getHeader(origin);if (StringUtils.isEmpty(origin)){origin blank;}return origin; // 该结果会被sentinel定义的授权规则检验}
}在sentinel控制台中定义授权规则 自定义异常结果 默认情况下发生限流、降级、授权拦截时都会抛出异常到调用方限流异常自定义异常返回结果需要实现BlockExceptiionHandler BlockExceptiionHandler的子类 异常说明FlowException限流异常ParamFlowException热点参数限流异常DegradeException降级异常AuthorityException授权规则异常SystemBlockException系统规则异常 实现BlockExceptiionHandler Component
public class SentinelExceptionHandler implements BlockExceptionHandler {Overridepublic void handle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, BlockException e) throws Exception {String msg 未知异常;int status 429;if (e instanceof FlowException) {msg 请求被限流;} else if (e instanceof ParamFlowException) {msg 热点参数限流;} else if (e instanceof DegradeException) {msg 请求被降级;} else if (e instanceof AuthorityException) {msg 授权规则异常;status 401;} else if (e instanceof SystemBlockException) {msg 系统规则异常;}httpServletResponse.setContentType(application/json;charsetutf-8);httpServletResponse.setStatus(status);httpServletResponse.getWriter().println(msg: msg status: status);}
}6.规则管理模式 原始模式Sentinel的默认模式讲规则保存在内存中重启服务丢失 pull模式控制台将配置的规则推送到Sentinel客户端而客户端会将配置规则保存在本地文件或数据库中。以后会定时去本地文件或数据库中查询更新本地规则。 push模式控制台将配置规则推送到远程配置中心。监听变更实时更新 sentinel规则持久化.md
二、分布式事务
1.什么是分布式事务
在分布式系统下一个业务跨越多个服务或数据源每个服务都是一个分支事务要保证所有分支事务最终状态一致这样的事务就是分布式事务。
2.理论基础 CAP定理分布式系统三个指标分布式系统无法同时满足这三个指标。 分布式系统节点通过网络连接一定会出现分区问题当分区出现时系统的一致性和可用性就无法同时满足Consistency一致性用户访问分布式系统的任意节点得到的数据必须一致Availability可用性用户访问集群中的任意健康节点必须能得到响应而不是超时拒绝Partition tolerance分区容错性 Partition分区 因为网络故障或其他原因导致分布式系统中的部分节点与其他节点失去连接形成独立分区。Tolerance容错在集群出现分区时整个系统也要持续对外提供服务 BASE理论BASE理论是对CAP的一种解决思路包含三个思想 Basically Available基本可用分布式系统在出现故障时允许损失部分可用性即保证核心可用Sorf State软状态在一定时间内允许出现中间状态比如临时的不一致状态Eventually Consistent最终一致性虽然无法保证强一致性但是在软状态结束后最终达到数据一致。 分布式事务最大的问题是各个子事物的一致性问题可以借鉴CAP定理和BASE理论 AP模式各子事务分别执行和提交允许出现结果不一致然后采用弥补措施恢复数据即可实现最终一致CP模式各个子事务执行相互等待同时提交同时回滚达成强一致。但事务等待过程中处于弱可用状态 分布式事务模型解决分布式事务各个子系统之间必须能感知到彼此的事务状态才能保证状态一致因此需要一个事务协调者来协调每个一个事务的参与者子系统事务。 子系统事务称为分支事务有关联的各个分支事务在一起称为全局事务
3.部署与集成Seata
seata的部署和集成.md
4.Seata的四种模式 XA模式强一致性XA规范是X/Open组织定义的分布式事务处理标准XA规范描述了全局的TM与局部的RM之间的接口几乎所有主流的数据库都对XA规范提供了支持。 Seata的XA模式 RM一阶段的工作 注册分支事务到TC执行分支业务SQL但不提交报告执行状态到TC TC二阶段的工作TC检测各分支事务执行状态 如果都成功通知所有RM提交事务如果有失败通知所有RM回滚事务 RM二阶段的工作接收TC指令提交或回滚事务 XA模式的优点 事务一致性强满足ACID原则常用数据库都支持实现简单并且没有代码侵入缺点 因为一阶段需要锁定数据库资源等待二阶段结束才释放性能较差依赖关系型数据库实现事务 实现XA模式 修改application.yaml文件每个参与事务的微服务开启XA模式 seata:registry: # TC服务注册中心的配置微服务根据这些信息去注册中心获取tc服务地址# 参考tc服务自己的registry.conf中的配置type: nacosnacos: # tcserver-addr: 127.0.0.1:8848namespace: group: DEFAULT_GROUPapplication: seata-tc-server # tc服务在nacos中的服务名称cluster: SHtx-service-group: seata-demo # 事务组根据这个获取tc服务的cluster名称service:vgroup-mapping: # 事务组与TC服务cluster的映射关系seata-demo: SHdata-source-proxy-mode: XA #开启数据源代理的XA模式给发起全局事物的入口方法添加GlobalTransactional注解 OverrideGlobalTransactionalpublic Long create(Order order) {// 创建订单orderMapper.insert(order);try {// 扣用户余额accountClient.deduct(order.getUserId(), order.getMoney());// 扣库存storageClient.deduct(order.getCommodityCode(), order.getCount());} catch (FeignException e) {log.error(下单失败原因:{}, e.contentUTF8(), e);throw new RuntimeException(e.contentUTF8(), e);}return order.getId();}AT模式AT模式同样是分阶段提交事务模型不过弥补了XA模型中资源锁定周期过长的缺陷。 Seata的AT模式 阶段一RM工作 记录undo_log(数据快照)执行业务sql并提交报告事务状态 阶段二提交时RM的工作删除undo_log即可阶段二回滚时RM的工作根据undo_log恢复数据到更新前 AT模式与XA模式最大的区别是 XA模式一阶段不提交事务锁定资源AT模式一阶段直接提交不锁定资源XA模式依赖数据库机制实现回滚AT模式利用数据快照实现数据回滚XA模式强一致AT模式最终一致 AT模式的优点 一阶段完成直接提交事务释放数据库资源、性能较好利用全局锁实现读写隔离没有代码侵入框架自动完成回滚和提交 AT模式的缺点 两阶段之间属于软状态属于最终一致框架的快照功能会影响性能但比XA模式好很多 实现AT模式与XA模式不同的是仅需修改yaml文件事务所实现方式 seata:registry: # TC服务注册中心的配置微服务根据这些信息去注册中心获取tc服务地址# 参考tc服务自己的registry.conf中的配置type: nacosnacos: # tcserver-addr: 127.0.0.1:8848namespace: group: DEFAULT_GROUPapplication: seata-tc-server # tc服务在nacos中的服务名称cluster: SHtx-service-group: seata-demo # 事务组根据这个获取tc服务的cluster名称service:vgroup-mapping: # 事务组与TC服务cluster的映射关系seata-demo: SHdata-source-proxy-mode: AT #开启数据源代理的AT模式TCC模式 原理TCC模式与AT模式非常相似每阶段都是独立事务不同的是TCC通过人工编码来实现恢复数据。需要实现3个方法 Try资源的检测和预留Confirm完成资源操作业务要求Try成功Confirm要能成功。Cancel预留资源释放可以理解为try的反向操作 TCC模式的优点 一阶段直接提交事务释放数据库资源性能好相比于AT模型无需生成快照无需使用全局锁性能最强不依赖数据库业务而是依赖补偿操作可以用于非事务型数据库 TCC模式的缺点 有代码侵入需要人为编写try,confirm,cancel接口太麻烦软状态食物最终一致需要考虑Confirm和Cancel失败的情况做好幂等处理 实现TCC 准备数据库冻结表 SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS 0;-- ----------------------------
-- Table structure for account_freeze_tbl
-- ----------------------------
DROP TABLE IF EXISTS account_freeze_tbl;
CREATE TABLE account_freeze_tbl (xid varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,user_id varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,freeze_money int(11) UNSIGNED NULL DEFAULT 0,state int(1) NULL DEFAULT NULL COMMENT 事务状态0:try1:confirm2:cancel,PRIMARY KEY (xid) USING BTREE
) ENGINE InnoDB CHARACTER SET utf8 COLLATE utf8_general_ci ROW_FORMAT COMPACT;-- ----------------------------
-- Records of account_freeze_tbl
-- ----------------------------SET FOREIGN_KEY_CHECKS 1; 编写TCC业务逻辑层接口 LocalTCC
public interface TCCAccountService {TwoPhaseBusinessAction(name deduct , commitMethod confirm , rollbackMethod cancel)void deduct(BusinessActionContextParameter(paramName userId) String userId , BusinessActionContextParameter(paramName money) int money);boolean confirm(BusinessActionContext ctx);boolean cancel(BusinessActionContext ctx);
}实现TCC业务逻辑接口 Service
public class TCCAccountServiceImpl implements TCCAccountService {Autowiredprivate AccountMapper accountMapper;Autowiredprivate AccountFreezeMapper accountFreezeMapper;OverrideTransactionalpublic void deduct(String userId, int money) {String xid RootContext.getXID();// 防止业务悬挂AccountFreeze oldFreeze accountFreezeMapper.selectById(xid);if (oldFreeze ! null) {// cancel执行过拒绝执行return;}// 扣款accountMapper.deduct(userId , money);// 记录冻结资金AccountFreeze freeze AccountFreeze.builder().freezeMoney(money).xid(xid).userId(userId).state(AccountFreeze.State.TRY).build();accountFreezeMapper.insert(freeze);}Overridepublic boolean confirm(BusinessActionContext ctx) {String xid ctx.getXid();int count accountFreezeMapper.deleteById(xid);return count 0;}Overridepublic boolean cancel(BusinessActionContext ctx) {AccountFreeze freeze accountFreezeMapper.selectById(ctx.getXid());// 获取UserIdString userId ctx.getActionContext(userId).toString();// 防止空回滚if (freeze null) {AccountFreeze freezeNUll AccountFreeze.builder().state(AccountFreeze.State.CANCEL).xid(ctx.getXid()).userId(userId).freezeMoney(0).build();accountFreezeMapper.insert(freezeNUll);return true;}// 幂等判断if (freeze.getState() AccountFreeze.State.CANCEL) {return true;}// 回滚数据accountMapper.refund(freeze.getUserId(), freeze.getFreezeMoney());// 设置事务状态为cancelfreeze.setState(AccountFreeze.State.CANCEL);int count accountFreezeMapper.updateById(freeze);return count 1;}
}Saga模式Saga模式是seata提供的长事务解决方案。 Saga的两个阶段 一阶段直接提交本地事务二阶段成功则什么都不做失败则通过编写补偿业务代码来回滚 Saga模式的优点 事务参与者可以基于事件驱动实现异步调用吞吐高一阶段直接提交事务无锁性能好不用编写TCC中的三个阶段实现简单 缺点 软状态持续时间不确定时效性差没有锁没有事务隔离会有脏写
5.Seata高可用
seata的部署和集成.md
三、分布式缓存
1.Redis持久化
RDB持久化全称Redis DataBase Backup fileRedis数据备份文件也被叫做Redis数据快照。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启时从磁盘读取快照文件恢复数据 RDB方式bgsave的基本流程 fork主进程得到一个子进程共享内存空间子进程读取内存数据并写入新的RDB文件用新的RDB文件替换旧的RDB文件 RDB什么时候执行save 60 1000代表什么含义 默认是服务正常停止时代表60秒内至少执行1000次修改则触发RDB RDB的缺点 RDB执行间隔时间长两次RDB之间写入数据有丢失风险fork子进程、压缩、写出RDB文件都比较耗时 AOF持久化AOF全称为Append Only File追加文件。Redis处理的每一个写命名都会记录在AOF文件可以看做是命令日志文件。
2.搭建主从架构与哨兵模式
Redis集群.md
3.RedisTemplate的哨兵模式 在Sentinel集群监管下的Redis主从集群其节点会因为自动故障转移而发生变化Redis的客户端必须感知这种变化及时更新连接信息。Spring的RedisTemplate底层使用lettuce实现了节点的感知和自动切换 实现 在pom文件中引入redis的starter依赖 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependency然后在配置application.yml中指定sentinel的信息 spring:redis:sentinel:master: mymaster # 指定master 名称nodes: # 指定redis-sentinel集群信息- 192.168.88.101:27001- 192.168.88.101:27002- 192.168.88.101:27003配置主从读写分离 Beanpublic LettuceClientConfigurationBuilderCustomizer configurationBuilderCustomizer() {return clientConfigurationBuilder - clientConfigurationBuilder.readFrom(ReadFrom.REPLICA_PREFERRED);}4.Redis分片集群 搭建分片集群 Redis集群.md 散列插槽 Redis会把每一个master节点映射到0~16383共16384个插槽上(hash slot)。数据key不是与节点绑定而是与插槽绑定Redis会根据key的有效部分计算插槽值。分两种情况 key中包含“{}”,且“{}”中至少包含一个字符“{}”中的部分是有效部分key中不包含“{}”整个key都是有效部分 Redis如何判断某个key在集群中哪个实例 将16384个插槽分配到不同的实例根据key的有效部分计算哈希值对16384取余余数作为插槽寻找插槽所在的实例即可 如何将同一类数据固定的保存在同一个Redis实例 这一类数据使用相同的有效部分例如key都以{typeId}为前缀 集群伸缩 添加一个节点到集群 # 后一个参数为集群中已存在的节点
redis-cli --cluster add-node 192.168.88.101:7004 192.168.88.101:7001为新节点分配分片 # 找出要被分配的节点
redis-cli --cluster reshard 192.168.88.101:7001
# 指定要移动的分片的数量
3000
# 接收者的redis集群ID
739892758277523532hf24h4v2
# 指定数据源ID
1824771274fsd88h43242h5532
# 结束
done
# 确定
yes故障转移 首先是该实例与其他实例失去连接然后是疑似宕机最后是确定下线自动提升一个slave为新的master 数据迁移 利用cluster failover命令可以手动让集群中的某个master宕机切换到执行cluster failover命令的这个slave节点实现无感知的数据迁移 cluster failover5.RedisTemplate访问分片集群 RedisTemplate底层同样基于lettuce实现了分片集全的支持而使用的步骤与哨兵模式基本一致 引入redis的starter依赖配置分片集群地址配置读写分离 与哨兵模式相比其中只有分片集群的配置方式略有差异 spring:redis:cluster:nodes: # 指定redis-sentinel集群信息- 192.168.88.101:7001- 192.168.88.101:7002- 192.168.88.101:7003- 192.168.88.101:8001- 192.168.88.101:8002- 192.168.88.101:8003四、多级缓存
1.多级缓存的意义及解决方案
传统缓存的问题 请求要经过Tomcat处理Tomcat的性能成为整个系统的瓶颈Redis缓存失效时会对数据库产生冲击 多级缓存方案多级缓存就是充分利用请求处理的每个环节分别添加缓存减轻Tomcat压力提升服务性能。
2.进程缓存-caffeine 缓存分类缓存在日常开发中起到至关重要的作用由于是存储在内存中数据的读取速度是非常快的能大量减少对数据库的访问减少数据库的压力 分布式缓存例如Redis 优点存储容量大、可靠性好、可以在集群间共享缺点访问缓存有网络开销场景缓存数据量较大、可靠性要求较高、需要在集群间共享 进程本地缓存例如HashMap、GuavaCache 优点读取本地内存没有网络开销速度更快缺点存储容量有限、可靠性较低、无法共享场景性能要求较高缓存数据量较小 CaffeineCaffeine是一个基于java8开发的提供了近乎最佳命中率的高性能本地缓存库。目前Spring内部的缓存使用的就是Caffeine Caffeine示例 /*基本用法测试*/Testvoid testBasicOps() {// 创建缓存对象CacheString, String cache Caffeine.newBuilder().build();// 存数据cache.put(name , 小刚);// 取数据String name cache.getIfPresent(name);System.out.println(name name);// 未命中的参数则从数据库查String age cache.get(age, item - {return 18; // 此处模拟数据库查询});System.out.println(age age);}Caffeine提供了三种缓存驱逐策略 基于容量设置缓存的数量上限 /*基于大小设置驱逐策略*/Testvoid testEvictByNum() throws InterruptedException {// 创建缓存对象CacheString, String cache Caffeine.newBuilder()// 设置缓存大小上限为 1.maximumSize(1).build();// 存数据cache.put(gf1, 柳岩);cache.put(gf2, 范冰冰);cache.put(gf3, 迪丽热巴);// 延迟10ms给清理线程一点时间Thread.sleep(10L);// 获取数据System.out.println(gf1: cache.getIfPresent(gf1));System.out.println(gf2: cache.getIfPresent(gf2));System.out.println(gf3: cache.getIfPresent(gf3));}基于时间设置缓存的有效时间 /*基于时间设置驱逐策略*/Testvoid testEvictByTime() throws InterruptedException {// 创建缓存对象CacheString, String cache Caffeine.newBuilder().expireAfterWrite(Duration.ofSeconds(1)) // 设置缓存有效期为 10 秒.build();// 存数据cache.put(gf, 柳岩);// 获取数据System.out.println(gf: cache.getIfPresent(gf));// 休眠一会儿Thread.sleep(1200L);System.out.println(gf: cache.getIfPresent(gf));}基于引用设置缓存为软引用或弱引用利用GC来回收缓存数据量。性能差 在默认情况下当一个缓存元素过期的时候Caffeine不会自动立即将其清理或驱逐。而是在一次读或写操作后或者在空闲时间完成对失效数据的驱逐。 Caffeine实践 引入依赖 dependencygroupIdcom.github.ben-manes.caffeine/groupIdartifactIdcaffeine/artifactId/dependency编写配置类注入Bean Configuration
public class CaffeineConfig {Beanpublic CacheLong , Item itemCaffeine() {return Caffeine.newBuilder().initialCapacity(100).maximumSize(10000).build();}Beanpublic CacheLong , ItemStock itemStockCaffeine() {return Caffeine.newBuilder().initialCapacity(100).maximumSize(10000).build();}
}编写业务逻辑 RestController
RequestMapping(item)
public class ItemController {Autowiredprivate IItemService itemService;Autowiredprivate IItemStockService stockService;Autowiredprivate CacheLong , Item itemCache;Autowiredprivate CacheLong , ItemStock itemStockCache;GetMapping(/{id})public Item findById(PathVariable(id) Long id){return itemCache.get(id , item - itemService.query().ne(status, 3).eq(id, item).one());}GetMapping(/stock/{id})public ItemStock findStockById(PathVariable(id) Long id){return itemStockCache.get(id , item - stockService.getById(item));}
}
3.lua语言 Lua是一种轻量小巧的脚本语言用标准C语言编写并以源代码形式开放其设计目的是为了嵌入应用程序中从而为应用程序提供灵活的扩展和定制功能。 数据类型 数据类型描述nil只有值nil属于该类表示一个无效值相当于falseboolean包含两个值false和truenumber表示双精度类型的实浮点数string字符串由一对双引号或单引号来表示function由C或Lua编写的函数tableLua中的表其实是一个关联数组数组的索引可以是数字、字符串或表类型。在Lua里table的创建是通过“构造表达式”来完成最简单的表达式是{}用来创建一个空表 判断变量的值 print(type(你好))变量:lua声明变量的时候并不需要指定数据类型 local str hellolocal num 123local flag truelocal arr {1 , 2 , 3}local map {name jack , age 21}访问table -- 访问数组lua数组从角标1开始
print(arr[1])
-- 访问table
print(map[name])
print(map.name)循环 遍历数组 -- 声明数组
local arr {1 , 2 , 3}
-- 遍历数组
for index,value in ipairs(arr) doprint(index , value)
end遍历table -- 声明map
local map {namejack , age21}
-- 遍历map
for key,value in pairs(map) doprint(key , value)
end函数 函数的定义 function 函数名(arg1 , arg2...)-- 函数体return 返回值
end条件控制 if(布尔表达式)
then-- true时执行
else -- false时执行
end逻辑运算基于英文 操作符描述and逻辑与or逻辑或not逻辑非
4.初识OpenResty
安装OpenResty.md OpenResty是一个基于Nginx的高性能web平台用于方便的搭建能够处理超高并发、扩展性极高的动态web应用、web服务和动态网关。具备下列特点 具备Nginx的完整功能基于Lua语言进行扩展集成了大量精良的Lua库、第三方模块允许使用Lua自定义业务逻辑、自定义库 OpenResty快速使用 在nginx反向代理服务器上配置服务代理将请求转发到OpenResty中
#user nobody;
worker_processes 1;events {worker_connections 1024;
}http {include mime.types;default_type application/octet-stream;sendfile on;#tcp_nopush on;keepalive_timeout 65;# OPenResty地址upstream nginx-cluster{server 192.168.88.101:8081;}server {listen 80;server_name localhost;location /api {# 请求转发给OpenRestyproxy_pass http://nginx-cluster;}location / {root html;index index.html index.htm;}error_page 500 502 503 504 /50x.html;location /50x.html {root html;}}
} 在OpenResty中的nginx的nginx.conf文件中写入处理配置
#user nobody;
worker_processes 1;
error_log logs/error.log;events {worker_connections 1024;
}http {include mime.types;default_type application/octet-stream;sendfile on;keepalive_timeout 65;#lua 模块lua_package_path /usr/local/openresty/lualib/?.lua;;;# c模块 lua_package_cpath /usr/local/openresty/lualib/?.so;;; server {listen 8081;server_name localhost;location /api/item {# 默认的响应类型default_type application/json;# 响应结果由lua/item.lua来决定content_by_lua_file lua/item.lua;}location / {root html;index index.html index.htm;}error_page 500 502 503 504 /50x.html;location /50x.html {root html;}}
}编写lua/item.lua脚本此处默认返回静态资源假数据 -- ngx.say() 函数相当于response给客户端响应结果
ngx.say({id:10001,name:SALSA AIR,title:RIMOWA 21寸托运箱拉杆箱 SALSA AIR系列屎黄色 820.70.36.4,price:199999999999,image:https://m.360buyimg.com/mobilecms/s720x720_jfs/t6934/364/1195375010/84676/e9f2c55f/597ece38N0ddcbc77.jpg!q70.jpg.webp})重新加载nginx ngnix -s reload5.获取请求参数 [!tip] OpenResty提供了各种ApI用来获取不同类型的请求参数 参数格式参数示例参数解析代码示例路径占位符/item/10011.正则表达式匹配location ~ /item/(\d){content_by_lua_file lua/item.lua;} 2.匹配到的参数会存入ngx.var数组中可用角标获取local id ngx.var[1]请求头id:1001返回值是table类型local headers ngx.req.get_headers()Get请求参数?id1001返回值是table类型local getParams ngx.req.get_uri_args()Post表单参数id10011.读取请求体ngx.req.read_body() 2.返回值是table类型local postParms ngx.req.get_post_args()JSON参数{“id”:1001}1.读取请求体ngx.req.read_body() 2.返回值是string类型local jsonBody ngx.req.get_body_data()
6.查询Tomcat nginx提供了内部API用以发送http请求 local resp ngx.location.capture(/path , {method ngx.HTTP_GET, -- 请求方式args {a1,b2}, -- get方式传参数body c3d4 -- post方式传参数})[!TIP] 返回的响应内容包括 resp.status响应状态码resp.header响应头是一个tableresp.body响应体就是一个响应数据 [!WARNING] 注意此处的/path是路径并不包含IP和端口。这个请求会被nginx内部的server监听并处理。 但是我们希望这个请求发送到Tomcat服务器所以还需要编写一个server来对这个路径做反向代理 location /path {# 这里是windows电脑的ip和java服务端接口需要确保windows的防火墙处于关闭状态proxy_pass http://192.168.88.1:8081;
}封装http查询的函数 [!TIP] 我们可以把http查询的请求 封装为一个函数放到OpenResty函数库中方便后期使用。 在/usr/local/openresty/lualib目录下创建common.lua文件 vi /usr/local/openresty/lualib/common.lua在common.lua中封装http查询的函数 -- 封装函数发送http请求并解析响应
local function read_http(path, params)local resp ngx.location.capture(path,{method ngx.HTTP_GET,args params,})if not resp then-- 记录错误信息返回404ngx.log(ngx.ERR, http not found, path: , path , , args: , args)ngx.exit(404)endreturn resp.body
end
-- 将方法导出
local _M { read_http read_http
}
return _M实现 编写item.lua -- 导入common函数库
local common require(common)
local read_http common.read_http
-- 导入cjson库该库可以处理json数据
local cjson require(cjson)-- 获取请求参数
local id ngx.var[1]-- 查询商品信息
local itemJson read_http(/item/ .. id , nil)
-- 查询库存信息
local stockJson read_http(/item/stock/ .. id , nil)-- 使用cjson转化为对象
local item cjson.decode(itemJson)
local stock cjson.decode(stockJson)
-- 组合数据
item.stock stock.stock
item.sold stock.sold-- 返回结果
ngx.say(cjson.encode(item))在OpenResty内部的ngnix服务器编写接受请求的location location /item {# 这里是windows电脑的ip和java服务端接口需要确保windows的防火墙处于关闭状态proxy_pass http://192.168.88.1:8081;} [!tip] 实际情况tomcat服务往往存在多个实例本地缓存的数据只在实例中不能共享而nginx的负载均衡策略默认轮询会轮流访问多个实例。因此为了提高本地缓存命中率我们可以对uri路径进行哈希操作对于同一个uri访问相同的实例此时本地缓存命中率大大提升。
7.冷启动与缓存预热 冷启动服务刚刚启动时redis中并没有缓存如果所有商品数据都在第一次查询时添加缓存可能会给数据库带来较大压力。 缓存预热在实际开发中我们可以利用大数据统计用户访问的热点数据在项目启动时将这些热点数据提前查询并保存到redis中。 缓存预热实现 利用docker安装redis docker run \
--name redis \
-p 6379:6379 \
-d redis redis-server \
--appendonly yesjava项目引入依赖 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependency配置redis地址 spring:redis:port: 6379host: 192.168.88.101编写RedisHandler Component
public class RedisHandler implements InitializingBean {Autowiredprivate RedisTemplate redisTemplate;Autowiredprivate IItemService itemService;Autowiredprivate IItemStockService iItemStockService;Autowiredprivate ObjectMapper objectMapper;Overridepublic void afterPropertiesSet() throws Exception {// 查询数据库数据ListItem itemList itemService.list();ListItemStock itemStockList iItemStockService.list();// 添加数据进redis缓存itemList.stream().forEach(item - {// 将item序列化为jsontry {String json objectMapper.writeValueAsString(item);redisTemplate.opsForValue().set(item:id: item.getId() , json);} catch (JsonProcessingException e) {throw new RuntimeException(e);}});itemStockList.stream().forEach(item - {// 将item序列化为jsontry {String json objectMapper.writeValueAsString(item);redisTemplate.opsForValue().set(stockItem:id: item.getId() , json);} catch (JsonProcessingException e) {throw new RuntimeException(e);}});}
}
7.查询Redis缓存 封装函数 [!Tip] OpenResty提供了操作Redis的模块我们只需要引入该模块就能直接使用 引入Redis模块并初始化Redis对象 -- 引入该模块
local redis require(resty.redis)
-- 初始化Redis对象
local red redis:new()
-- 设置Redis超时时间
red:set_timeouts(1000 , 1000 , 1000)重新封装函数 -- 关闭redis连接的工具方法,其实是放入连接池
local function close_redis(red)local pool_max_idle_time 1000 -- 连接的空闲时间单位毫秒local pool_size 100 -- 连接池大小local ok,err red:set_Keepalive(pool_max_idle_time , pool_size)if not ok thenngx.log(ngx.ERR,放入Redis连接池失败,err)end
endOpenResty提供了操作Redis模块我们只要引入该模块就能直接使用 -- 查询redis的方法ip和port是redis地址key是查询的key
local function read_redis(ip , port , key)-- 获取一个链接local ok,err red:connect(ip , port)if not ok thenngx.log(ngx.ERR , 连接Redis失败: , err)return nilend-- 查询Redislocal resp,err red:get(key)-- 查询失败处理if not resp thenngx.log(ngx.ERR , 查询Redis失败: , err , key , key)end-- 得到的数据为空处理if resp ngx.null thenresp nilngx.log(ngx.ERR , 查询到的Redis数据为空key , key)endclose_redis(red)return resp
end编写item.lua -- 导入common函数库
local common require(common)local read_redis common.read_redis
local read_http common.read_http
-- 导入cjson库该库可以处理json数据
local cjson require(cjson)-- 封装查询函数
function read_data(key , path , params)-- 查询redislocal resp read_redis(127.0.0.1 , 6379 , key)-- 判断查询结果if not resp thenngx.log(ngx.ERR , 查询为空key: , key)-- redis查询失败去查询httpresp read_http(path , params)endreturn resp
end-- 获取请求参数
local id ngx.var[1]-- 查询商品信息
local itemJson read_data(item:id: .. id , /item/..id , nil)
ngx.log(ngx.ERR , itemJson------: , itemJson)
-- 查询库存信息
local stockJson read_data(stockItem:id: .. id , /item/stock/..id , nil)
ngx.log(ngx.ERR , stockJson----: , stockJson)-- 使用cjson转化为对象
local item cjson.decode(itemJson)
local stock cjson.decode(stockJson)
-- 组合数据
item.stock stock.stock
item.sold stock.sold-- 返回结果
ngx.say(cjson.encode(item))8.添加nginx本地缓存 OpenResty为Nginx提供了shard dict的功能可以在nginx的多个worker之间共享数据实现缓存功能。 开启共享字典在ngnix.conf的http下添加配置 # 共享字典也就是本地缓存名称叫做item_cache大小150m
lua_shared_dict item_cache 150m; 操作共享词典 -- 获取本地缓存对象
local item_cache ngx.shared.item_cache
-- 存储指定key、value、、过期时间单位s默认为0代表永不过期
item_cache:set(key , value , 1000)
-- 读取
local val item_cache:get(key)修改查询代码 -- 导入common函数库
local common require(common)local read_redis common.read_redis
local read_http common.read_http
-- 导入cjson库该库可以处理json数据
local cjson require(cjson)
-- 导入共享词典本地缓存
local item_cache ngx.shared.item_cache-- 封装查询函数
function read_data(key , path , params)-- 首先查询本地缓存local val item_cache:get(key)if not val then-- 如果查询不到再查询Redis或httpngx.log(ngx.ERR, 查询本地缓存失败查询Redis Key: , key )-- 查询redislocal resp read_redis(127.0.0.1 , 6379 , key)val resp-- 判断查询结果if not resp thenngx.log(ngx.ERR , redis查询为空key: , key , 尝试查询HTTP)-- redis查询失败去查询httpval read_http(path , params)endend-- 查询成功写入内存item_cache:set(key , val , 60)return val
end-- 获取请求参数
local id ngx.var[1]-- 查询商品信息
local itemJson read_data(item:id: .. id , /item/..id , nil)
ngx.log(ngx.ERR , itemJson------: , itemJson)
-- 查询库存信息
local stockJson read_data(stockItem:id: .. id , /item/stock/..id , nil)
ngx.log(ngx.ERR , stockJson----: , stockJson)-- 使用cjson转化为对象
local item cjson.decode(itemJson)
local stock cjson.decode(stockJson)
-- 组合数据
item.stock stock.stock
item.sold stock.sold-- 返回结果
ngx.say(cjson.encode(item))9.缓存同步 数据同步策略 设置有效期同步双写异步通知 Canal译为水道、管道canal是阿里巴巴旗下的一款开源项目基于java开发。基于数据库增量日志解析提供增量数据订阅消费。canal基于mysql主从同步实现 业务实现 引入Canal依赖 dependencygroupIdtop.javatool/groupIdartifactIdcanal-spring-boot-starter/artifactIdversion1.2.1-RELEASE/version
/dependency编写配置 canal:destination: heima # canal实例名要跟canal-server运行时的destination一致server: 192.168.150.101:11111 # canal地址Canal推送给Canal-client的是被修改的这一行数据row而我们引入的canal-client则会帮我们把行数据封装到Item实体类中。这个过程中需要数据库与实体类的映射关系需要用到JPA的几个注解 Data
TableName(tb_item)
public class Item {TableId(type IdType.AUTO)Id // 标记表中的Id字段private Long id;//商品idColumn(name name) // 字段名不一致可以指定private String name;//商品名称private String title;//商品标题private Long price;//价格分private String image;//商品图片private String category;//分类名称private String brand;//品牌名称private String spec;//规格private Integer status;//商品状态 1-正常2-下架private Date createTime;//创建时间private Date updateTime;//更新时间TableField(exist false)Transient // 表中没有的字段private Integer stock;TableField(exist false)Transient // 表中没有的字段private Integer sold;
} 编写监听器监听Canal消息
CanalTable(tb_item) // 指定要监听的表
Component
public class ItemHandler implements EntryHandlerItem {Autowiredprivate RedisHandler redisHandler;Autowiredprivate CacheLong , Item itemcache;Overridepublic void insert(Item item) {// 写数据到RedisredisHandler.insertOrUpdateItem(item);// 写数据到jvm缓存itemcache.put(item.getId() , item);}Overridepublic void update(Item before, Item after) {// 更新数据到redisredisHandler.insertOrUpdateItem(after);// 写数据到jvm缓存itemcache.put(after.getId() , after);}Overridepublic void delete(Item item) {// 删除数据到RedisredisHandler.deleteItem(item);// 删除数据到jvm缓存itemcache.invalidate(item.getId());}
}
五、服务异步通讯-rabbitMQ的高级特性
1.消息可靠性 消息确认机制RabbitMQ提供了publisher confirm机制来避免消息发送到MQ中丢失。消息发送到MQ以后会返回一个结果给发送者表示消息是否处理成功。结果有两种请求 publisher-confirm发送者确认 消息成功投递到交换机返回ack消息未投递到交换机返回nack publisher-return发送者回执 消息投递到交换机了但是没有路由到队列返回ACK及路由失败原因 SpringAMQP实现生产者消息确认 在publisher这个微服务的application.yaml中添加配置 spring:rabbitmq:addresses: 192.168.88.101 # 地址名port: 5672 # 端口virtual-host: / # 虚拟主机username: itcastpassword: 123321publisher-confirm-type: correlated # 消息确认为异步回调 simple为同步回调template:mandatory: true # 定义消息路由失败策略。true则调用ReturnCallbackfalse则直接丢弃消息每一个RabbitTemplate只能配置一个ReturnCallback因此需要在项目启动过程中配置 Slf4j
Component
public class CommonConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取rabbitTemplate实例RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info(消息发送失败应答码{}原因{}交换机{}路由键{}消息{},replyCode , replyText , exchange , routingKey , message.toString());// 如需需要可以设置重发...}});}
}发送消息指定消息ID、消息ConfirmCallback Testpublic void testSendMessage2SimpleQueue(){// 1.准备消息String message hello , Spring amqp!;// 2.指定交换机String exchange amq.topic;// 4.准备routing keyString routingkey simple.test;// 5.准备CorrelationDataCorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());correlationData.getFuture().addCallback(result - {// 判断是否发送成功if (result.isAck()) {log.info(消息投递成功消息ID correlationData.getId());} else {log.error(消息投递到交换机失败! 消息ID correlationData.getId());}}, ex - {log.error(消息发送失败error ex 消息ID correlationData.getId());// 如需需要可以设置重发...});rabbitTemplate.convertAndSend(exchange , routingkey , message , correlationData);}2.消息持久化 交换机持久化 Bean
public DirectExchange simpleExchange() {// 三个参数交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange(simple.direct , true , false);
}队列持久化 Bean
public Queue simpleQueue() {// 使用QueueBuilder构造队列durable就是持久的return QueueBuilder.durable(simple,queue).build();
}消息持久化springAMQP中的消息默认是持久的可以通过属性指定 Message msg MessageBuilder.withBody(message.getBytes(StandarCharsets.UTF_8)) // 消息体.setDeliverMode(MessageDeliveryMode.PERSISTENT) // 持久化.build();3.消费者消息确认 RabbitMQ支持消费者确认机制即消费者处理消息后可以向MQ发送ack回执MQ收到ack回执后才会删除消息。 SpringMVC的三种确认模式 manual手动ack需要在业务代码结束后调用api发送ackauto自动ack由spring监测llistener代码是否出现异常没有异常则返回ack抛出异常则返回nacknone关闭ackMQ假定消费者获取消息后会成功处理因此消息投递后立即被删除 AMQP开启消息确认 spring:rabbitmq:addresses: 192.168.88.101 # 地址名port: 5672 # 端口virtual-host: / # 虚拟主机username: itcastpassword: 123321publisher-confirm-type: correlated # 消息确认为异步回调 simple为同步回调template:mandatory: true # 定义消息路由失败策略。true则调用ReturnCallbackfalse则直接丢弃消息listener:simple:prefetch: 1acknowledge-mode: auto # 消息确认机制 none 关闭 manual手动 auto自动4.失败重试机制 当消费者出现异常之后消息会不断requeue重新入队到队列再重新发送给消费者然后再次异常再次requeue无限循环导致mq的消息处理飙升带来不必要的处理压力我们可以利用Spring的retry机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。 AMQP开启消息失败重试 spring:rabbitmq:addresses: 192.168.88.101 # 地址名port: 5672 # 端口virtual-host: / # 虚拟主机username: itcastpassword: 123321listener:simple:prefetch: 1 # 修改消费者提前把握的最大数量retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初始的失败等待市场为1秒multiplier: 2 # 下次失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 4 # 最大重试次数stateless: true # true无状态false为有状态如果业务中包含事务这里改为false消费者失败消息处理策略在开启重试模式后重试次数用尽如果消息依然失败则需要有MessageRecoverer接口来处理它包含三种不同的实现 RejectAndDontRequestRecoverer重试耗尽后直接reject丢失消息。默认就是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重新耗尽后将失败消息投递到指定的交换机 AMQP实现失败消息处理 Configuration
public class ErrorMessageConfig {Beanpublic Queue errorQueue() {return new Queue(error.queue , true , true, false);}Beanpublic DirectExchange errorDirectChange() {return new DirectExchange(error.direct , true , false);}Beanpublic Binding errorDirectAndErrorQueueBinding() {return BindingBuilder.bind(errorQueue()).to(errorDirectChange()).with(error);}Beanpublic MessageRecoverer errorMessageRecoverer(RabbitTemplate rabbitTemplate ) {return new RepublishMessageRecoverer(rabbitTemplate , error.direct , error);}}
5.如何确保RabbitMQ消息的可靠性
开启生产者确认模式确保生产者的消息能够到达队列开启持久化功能确保消息未消费前在队列中不会丢失开始消费者确认机制为auto由spring确认消息处理成功后完成ack开启消费者失败重试机制并设置MessageRecoverer多次重试失败后将消息投递到异常交换机交由人工处理
6.死信交换机 初识死信交换机 死信当一个队列中的消息满足下列情况之一时可以成为死信dead letter 消费这使用basic.reject或basic.nack声明消费失败并且消息的requeue参数设置为false消息是一个过期消息超时无人消费要投递的队列消息堆积满了最早的消息可能成为死信如果该队列配置了dead-letter-exchange属性指定了一个交换机。那么队列中的死信就会投递到这个交换机中而这个交换机称为死信交换机 如何给队列绑定死信交换机 给队列设置dead-letter-exchange属性指定一个交换机给队列设置dead-letter-routing-key属性设置死信交换机与死信队列的RoutingKey TTL也就是Time-To-Live。如果一个队列中的消息TTL结束仍未消费则会变为死信ttl超时分为两种情况 消息所在的队列设置了存活时间消息本身设置了存活时间 AMQP实现死信交换机 绑定死信交换机与死信队列 RabbitListener(bindings QueueBinding(value Queue(name dl.queue , durable true),exchange Exchange(name dl.direct , type direct),key dl))public void listenDlQueue(String msg) {log.info(死信队列消息者收到消息{} , msg);}绑定ttl交换机和ttl队列 Configuration
public class TTLMessageConfig {Beanpublic DirectExchange ttlDirectExchange() {return new DirectExchange(ttl.direct , true , false);}Beanpublic Queue ttlQueue() {return QueueBuilder.durable(ttl.queue).ttl(10000).deadLetterExchange(dl.direct).deadLetterRoutingKey(dl).build();}Beanpublic Binding ttlDirectAndTtlQueueBinding() {return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with(ttl);}
}发送消息 Testpublic void testTTl(){String msg java是世界上最好的语言;Message message MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8))// .setExpiration(5000) // 此处可以设置消息本身的超时时间.build();CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());correlationData.getFuture().addCallback(result - {if (result.isAck()) {log.info(消息投递成功消息ID为 correlationData.getId());}},ex - {log.error(消息投递失败: ex);});rabbitTemplate.convertAndSend(ttl.direct , ttl , message , correlationData);}消息超时的两种方式是 给队列设置ttl属性进入队列后超过ttl时间的消息变为死信给消息设置ttl属性队列接收到消息超过ttl时间后变为死信两者共存以时间短的ttl为准
7.延迟队列 介绍利用TTL结合死信交换机我们实现了消息发出后消费者延迟收到消息的效果。这种消息模式就称为延迟队列模式 延迟发送短信用户下单如果用户在15分钟内未支付则自动取消预约工作会议20分钟后自动停止所有参会人员 安装延迟队列插件 见RabbitMQ安装文档
8.惰性队列 消息堆积问题当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。最早接收到的消息可能就会成为死信会被丢弃这就是消息堆积问题。 解决消息堆积的三种思路增加更多的消费者在消费者内开启线程池加快消息处理速度扩大队列容积提高堆积上限 惰性队列从RabbitMQ的3.6.0版本开始就增加了Lazy Queues的概念也就是惰性队列 接收到消息后直接存入磁盘而非内存消费者要消费消息时才会从磁盘中读取并加载到内存支持数百万条的消息存储 AMQP中声明惰性队列的两种方式 Bean方式 Bean
public Queue lazyQueue() {return QueueBuild.durable(lazy.queue).lazy() // 开启惰性队列.build();
}注解方式 RabbitListener(queuesToDeclare Queue(name lazy.queue ,durable true,arguments Argument(name x-queue-mode , value lazy)))public void listenQueue(String msg) {log.info(收到消息{} , msg);}getBytes(StandardCharsets.UTF_8)) // .setExpiration(“5000”) // 此处可以设置消息本身的超时时间 .build(); CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString()); correlationData.getFuture().addCallback( result - { if (result.isAck()) { log.info(“消息投递成功消息ID为” correlationData.getId()); } },ex - {log.error(消息投递失败: ex);});rabbitTemplate.convertAndSend(ttl.direct , ttl , message , correlationData);}消息超时的两种方式是 给队列设置ttl属性进入队列后超过ttl时间的消息变为死信给消息设置ttl属性队列接收到消息超过ttl时间后变为死信两者共存以时间短的ttl为准
7.延迟队列 介绍利用TTL结合死信交换机我们实现了消息发出后消费者延迟收到消息的效果。这种消息模式就称为延迟队列模式 延迟发送短信用户下单如果用户在15分钟内未支付则自动取消预约工作会议20分钟后自动停止所有参会人员 安装延迟队列插件 见RabbitMQ安装文档
8.惰性队列 消息堆积问题当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。最早接收到的消息可能就会成为死信会被丢弃这就是消息堆积问题。 解决消息堆积的三种思路增加更多的消费者在消费者内开启线程池加快消息处理速度扩大队列容积提高堆积上限 惰性队列从RabbitMQ的3.6.0版本开始就增加了Lazy Queues的概念也就是惰性队列 接收到消息后直接存入磁盘而非内存消费者要消费消息时才会从磁盘中读取并加载到内存支持数百万条的消息存储 AMQP中声明惰性队列的两种方式 Bean方式 Bean
public Queue lazyQueue() {return QueueBuild.durable(lazy.queue).lazy() // 开启惰性队列.build();
}注解方式 RabbitListener(queuesToDeclare Queue(name lazy.queue ,durable true,arguments Argument(name x-queue-mode , value lazy)))public void listenQueue(String msg) {log.info(收到消息{} , msg);}
