标签：缓存 return -- lock redis 压测 线程 public

高级篇--商城业务 部分

本笔记从谷粒商城的P141（性能测试）开始记录

一、性能与压力测试

​ 压力测试考察当前软硬件环境下系统所能承受的最大负荷并帮助找出系统瓶颈所在。压测都是为了系统在线上的处理能力和稳定性维持在一个标准范围内，做到心中有数。

​ 使用压力测试，我们有希望找到很多种用其他测试方法更难发现的错误。有两种错误类型是: 内存泄漏，并发与同步。

​ 有效的压力测试系统将应用以下这些关键条件:重复，并发，量级，随机变化。

1、性能指标介绍p141

• 响应时间（Response Time: RT）
  响应时间指用户从客户端发起一个请求开始，到客户端接收到从服务器端返回的响
  应结束，整个过程所耗费的时间。

• HPS（Hits Per Second） ：每秒点击次数，单位是次/秒。

• TPS（Transaction per Second）：系统每秒处理交易数，单位是笔/秒。

• QPS（Query per Second）：系统每秒处理查询次数，单位是次/秒。

  对于互联网业务中，如果某些业务有且仅有一个请求连接，那么 TPS=QPS=HPS，一般情况下用 TPS 来衡量整个业务流程，用 QPS 来衡量接口查询次数，用 HPS 来表示对服务器单击请求。

• 无论 TPS、QPS、HPS,此指标是衡量系统处理能力非常重要的指标，越大越好，根据经验，一般情况下：

  金融行业：1000TPS~50000TPS，不包括互联网化的活动
  保险行业：100TPS~100000TPS，不包括互联网化的活动
  制造行业：10TPS~5000TPS
  互联网电子商务：10000TPS~1000000TPS
  互联网中型网站：1000TPS~50000TPS
  互联网小型网站：500TPS~10000TPS

• 最大响应时间（Max Response Time） 指用户发出请求或者指令到系统做出反应（响应）的最大时间。

• 最少响应时间（Mininum ResponseTime） 指用户发出请求或者指令到系统做出反应（响应）的最少时间。

• 90%响应时间（90% Response Time） 是指所有用户的响应时间进行排序，第 90%的响应时间。

• 从外部看，性能测试主要关注如下三个指标

• 吞吐量：每秒钟系统能够处理的请求数、任务数。

• 响应时间：服务处理一个请求或一个任务的耗时。

• 错误率：一批请求中结果出错的请求所占比例。

2、安装JMeter

https://jmeter.apache.org/download_jmeter.cgi

下载对应的压缩包，解压运行jmeter.bat即可

3、JMeter 压测示例

1）、添加线程组

线程组参数详解：

• 线程数：虚拟用户数。一个虚拟用户占用一个进程或线程。设置多少虚拟用户数在这里也就是设置多少个线程数。

• Ramp-Up Period(in seconds)准备时长：设置的虚拟用户数需要多长时间全部启动。如果线程数为 10，准备时长为 2，那么需要 2 秒钟启动 10 个线程，也就是每秒钟启动 5 个线程。

• 循环次数：每个线程发送请求的次数。如果线程数为 10，循环次数为 100，那么每个线程发送 100 次请求。总请求数为 10*100=1000 。如果勾选了“永远”，那么所有线程会一直发送请求，一到选择停止运行脚本。

• Delay Thread creation until needed：直到需要时延迟线程的创建。

• 调度器：设置线程组启动的开始时间和结束时间(配置调度器时，需要勾选循环次数为永远)

• 持续时间（秒）：测试持续时间，会覆盖结束时间

• 启动延迟（秒）：测试延迟启动时间，会覆盖启动时间

• 启动时间：测试启动时间，启动延迟会覆盖它。当启动时间已过，手动只需测试时当前时间也会覆盖它。

• 结束时间：测试结束时间，持续时间会覆盖它。

2）、添加 HTTP 请求

3）、添加监听器

4）、启动压测&查看分析结果

结果分析：

•  有错误率同开发确认，确定是否允许错误的发生或者错误率允许在多大的范围内；
•  Throughput 吞吐量每秒请求的数大于并发数，则可以慢慢的往上面增加；若在压测的机器性能很好的情况下，出现吞吐量小于并发数，说明并发数不能再增加了，可以慢慢的往下减，找到最佳的并发数；
•  压测结束，登陆相应的 web 服务器查看 CPU 等性能指标，进行数据的分析;
•  最大的 tps，不断的增加并发数，加到 tps 达到一定值开始出现下降，那么那个值就是最大的 tps。
•  最大的并发数：最大的并发数和最大的 tps 是不同的概率，一般不断增加并发数，达到一个值后，服务器出现请求超时，则可认为该值为最大的并发数。
•  压测过程出现性能瓶颈，若压力机任务管理器查看到的 cpu、网络和 cpu 都正常，未达到 90%以上，则可以说明服务器有问题，压力机没有问题。
•  影响性能考虑点包括：
  数据库、应用程序、中间件（tomact、Nginx）、网络和操作系统等方面
•  首先考虑自己的应用属于 CPU 密集型还是 IO 密集型

我们先测一波之后，可以加大服务占用内存测试（以vm参数方式）再测试一波进行对比

4、JMeter在windows下地址占用bug解决

​ JMeter Address Already in use 错误解决

原因是windows 本身提供的端口访问机制的问题。

Windows 提供给 TCP/IP 链接的端口为 1024-5000，并且要四分钟来循环回收他们。就导致我们在短时间内跑大量的请求时将端口占满了。解决步骤如下：

1.cmd 中，用 regedit 命令打开注册表

2.在 HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters 下，
1 .右击 parameters，添加一个新的 DWORD，名字为 MaxUserPort
2 .然后双击 MaxUserPort，输入数值数据为 65534，基数选择十进制（如果是分布式运行的话，控制机器和 负载机器都需要这样操作哦）

3.修改配置完毕之后记得重启机器才会生效
https://support.microsoft.com/zh-cn/help/196271/when-you-try-to-connect-from-tcp-ports-grea
ter-than-5000-you-receive-t
TCPTimedWaitDelay：3

5、性能监控p144-p147

1）、jvm 内存模型

• 程序计数器 Program Counter Register：
• 记录的是正在执行的虚拟机字节码指令的地址，
• 此内存区域是唯一一个在JAVA虚拟机规范中没有规定任何OutOfMemoryError的区域
• 虚拟机：VM Stack
• 描述的是 JAVA 方法执行的内存模型，每个方法在执行的时候都会创建一个栈帧，用于存储局部变量表，操作数栈，动态链接，方法接口等信息
• 局部变量表存储了编译期可知的各种基本数据类型、对象引用
• 线程请求的栈深度不够会报 StackOverflowError 异常
• 栈动态扩展的容量不够会报 OutOfMemoryError 异常
• 虚拟机栈是线程隔离的，即每个线程都有自己独立的虚拟机栈
• 本地方法：Native Stack
• 本地方法栈类似于虚拟机栈，只不过本地方法栈使用的是本地方法
• 堆：Heap
• 几乎所有的对象实例都在堆上分配内存

2）、堆

​ 所有的对象实例以及数组都要在堆上分配。堆是垃圾收集器管理的主要区域，也被称为“GC堆”；也是我们优化最多考虑的地方。

堆可以细分为：

•  新生代
  •  Eden 空间
  •  From Survivor 空间
  •  To Survivor 空间
•  老年代
•  永久代/元空间
  • Java8 以前永久代，受 jvm 管理，java8 以后元空间，直接使用物理内存。因此，默认情况下，元空间的大小仅受本地内存限制。

垃圾回收

从 Java8 开始，HotSpot 已经完全将永久代（Permanent Generation）移除，取而代之的是一个新的区域—元空间（MetaSpace）

3）、jconsole 与 jvisualvm工具

​ JDK 的两个小工具 jconsole、jvisualvm（升级版的 jconsole）;通过命令行启动，可监控本地和远程应用。远程应用需要配置

1、jvisualvm 能干什么

监控内存泄露，跟踪垃圾回收，执行时内存、cpu 分析，线程分析...

• 运行：正在运行的
• 休眠：sleep
• 等待：wait
• 驻留：线程池里面的空闲线程
• 监视：阻塞的线程，正在等待锁

2、安装插件方便查看 gc

具体步骤：

1、 Cmd 启动 jvisualvm
2、工具->插件

如果在安装过程中有 503 错误，解决方法如下：

1、 打开网址 https://visualvm.github.io/pluginscenters.html

2、cmd 查看自己的 jdk 版本，找到对应的，如下：

​

3、复制上面的链接。并重新设置上即可，如下图：

然后安装我们需要的插件，我们目前主要需要如下插件：

安装完成重启一下jvisualvm工具即可

可以看到Visual GC 插件已经安装完成：

4）、监控指标p147

1、中间件指标

• 当前正在运行的线程数不能超过设定的最大值。一般情况下系统性能较好的情况下，线程数最小值设置 50 和最大值设置 200 比较合适。
• 当前运行的 JDBC 连接数不能超过设定的最大值。一般情况下系统性能较好的情况下，JDBC 最小值设置 50 和最大值设置 200 比较合适。
• ＧＣ频率不能频繁，特别是 FULL GC 更不能频繁，一般情况下系统性能较好的情况下，JVM 最小堆大小和最大堆大小分别设置 1024M 比较合适。

2、数据库指标

• SQL 耗时越小越好，一般情况下微秒级别。
• 命中率越高越好，一般情况下不能低于 95%。
• 锁等待次数越低越好，等待时间越短越好。

3、压测性能如下：

视频中老师机器压测性能如下：

ps:

1、响应时间的单位为毫秒

2、首页渲染（开缓存） 开的是thymleaf缓存

3、nginx动静分离p148

​ 还有一个优化手段就是资源的动静分离。

原因分析
目前是将所有的动态请求以及静态资源放在微服务中，这样的话，获取首页全量数据发的首页请求，无论是数据库的动态请求，还是每个页面的静态请求，都要交给Tomcat处理，这样，光静态请求就占用了Tomcat的很多资源，会导致吞吐量急剧下降，如何让静态资源快速返回，我们可以使用动静分离的手段。

优化首页全量数据获取

​ 因为首页里面的资源包换非常多的静态资源，所以适合做动静分离。

​ 以前，无论是动态或是静态请求，都会交给Nginx，再由Nginx转交给网关，由网关交给后台服务的集群，假设现在产生了很多首页的请求，一个就占用了图上微服务一半的资源，两个就会占满，最终吞吐量每秒就会只有两三个左右。

​ 我们的静态资源目前是放在微服务的静态资源文件夹下，如果我们将静态资源分离出来，放到Nginx中会怎么样呢？

​ 静态资源的请求，由于静态资源都移到了Nginx里面，所以Nginx会直接将资源返回；而动态请求，Nginx还是转交给网关处理，最终转交给微服务

	这样后台服务就只需要处理动态资源的请求，会提升很大的吞吐量。

具体实现动静分离的步骤：

1、静态资源搬家
将原idea的静态资源static文件夹，全部移到服务器的/mydata/nginx/html/目录下

2、修改index.html的资源请求路径
统一在html页面里的请求前加上/static/

同时禁用thymeleaf缓存

thymeleaf:
    cache: false

3、修改nginx配置
vim /mydata/nginx/conf/conf.d/gulimall.conf

注意：这一行配置的位置是nginx容器内部的静态资源存放的位置，因为这个位置的文件夹已经挂载（映射）到外面的Linux虚拟机里面的/mydata/nginx/html这个文件夹下面了，所以在1、静态资源搬家这一步直接将静态资源放在此处也就相当于放进了nginx容器的内部了，自然上面配置的位置也应该这样配置，也能访问得到资源。(myps：相当于给root字段赋值为/usr/share/nginx/html)

static开头的所有请求路径都去哪里找，都去root对应后面的目录去找，因为整个路径是完整的，不仅有static ,还有static文件夹，接下来有什么，就按照层级目录在文件夹下写了什么。除了static外，剩下的转给gulimall（说的是：location / 中的配置（网关的整个集群，而且以负载均衡的方式））

访问容器内部命令如下图示：

参考文章：(135条消息) docker exec命令详解_tiger_angel的博客-CSDN博客

P148第10分钟讲解修改配置

4、重启nginx服务

修改配置后重启nginx即可

docker restart nginx

现在首页的静态资源，全部都由nginx返回，首页的数据，全部都是由tomcat返回，这就是nginx的动静分离配置

最起码现在的tomcat只处理动态请求，占用的资源就会很小了。

4、、总结：

• 中间件越多，性能损失越大，大多都损失在网络交互了；
• 业务方面影响：
• Db（MySQL 优化）
• 模板的渲染速度（cpu 内存 缓存(最重要)）
• 静态资源（tomcat还要分一些线程来处理静态资源，吞吐量下降很多）

5）、JVM 分析&调优

ps：这部分是课件里的内容，不是依照视频讲解顺序（视频里好像也没有讲解。。）

​ jvm 调优，调的是稳定，并不能带给你性能的大幅提升。服务稳定的重要性就不用多说了，保证服务的稳定，gc 永远会是 Java 程序员需要考虑的不稳定因素之一。复杂和高并发下的服务，必须保证每次 gc 不会出现性能下降，各种性能指标不会出现波动，gc 回收规律而且干净，找到合适的 jvm 设置。Full gc 最会影响性能，根据代码问题，避免 full gc 频率。可以适当调大年轻代容量，让大对象可以在年轻代触发 yong gc，调整大对象在年轻代的回收频次，尽可能保证大对象在年轻代回收，减小老年代缩短回收时间；

1、几个常用工具

工具名称	作用
jstack	查看 jvm 线程运行状态，是否有死锁现象等等信息
jinfo	可以输出并修改运行时的 java 进程的 opts。
jps	与 unix 上的 ps 类似，用来显示本地的 java 进程，可以查看本地运行着几个 java程序，并显示他们的进程号。
jstat	一个极强的监视 VM 内存工具。可以用来监视 VM 内存内的各种堆和非堆的大小及其内存使用量。
jmap	打印出某个 java 进程（使用 pid）内存内的所有'对象'的情况（如：产生那些对象，及其数量）

2、命令示例

3、官方调优项文档

官方文档：https://docs.oracle.com/javase/8/docs/technotes/tools/unix/java.html#BGBCIEFC

P149 模拟线上内存崩溃宕机

性能压测-优化-模拟线上应用内存崩溃宕机情况

1）、具体实现步骤

1、调低服务内存

-Xmx100m

用200个线程来压

分配内存太小，很容易就会把新生代、老年代挤满，频繁的垃圾回收，内存崩溃

访问首页已经不能提供服务了，提示找不到实例了，这就是线上实例的整个过程，持续在运行期间，cpu，内存爆满卡死，将应用服务挤下线，

2）、调整参数进行优化

依旧是开启200个线程，将内存调大

-Xms1024m -Xmx1024m -Xmn512m

最终没有出现内存崩溃，服务一直处于可用状态

可以看出优化是有效果的

p150优化三级分类数据获取

具体流程

优化文件CategoryServiceImpl.java中的代码如下：

    private List<CategoryEntity> listByPrentCid(List<CategoryEntity> categories, Long parentCid) {
        List<CategoryEntity> finalCategories = categories.stream().filter(category ->
                category.getParentCid().equals(parentCid)).collect(Collectors.toList());
        return finalCategories;
    }

    @Override
    public Map<Long, List<Category2VO>> getCategoryJson() {
        // 查出所有分类，再使用stream处理
        List<CategoryEntity> allCategories = this.list();
        List<CategoryEntity> l1Categories = listByPrentCid(allCategories, 0L);

        Map<Long, List<Category2VO>> categoryMap = l1Categories.stream().collect(Collectors.toMap(k1 -> k1.getCatId(),
                v1 -> {
                    List<CategoryEntity> l2Categories = listByPrentCid(allCategories, v1.getCatId());
                    List<Category2VO> category2VOs = null;

                    if (l2Categories != null && l2Categories.size() > 0) {
                        category2VOs = l2Categories.stream().map(l2 -> {
                            // 根据当前2级分类查出所有3级分类
                            List<CategoryEntity> l3Categories = listByPrentCid(allCategories, l2.getCatId());
                            List<Category3VO> category3VOs = null;
                            if (l3Categories != null && l3Categories.size() > 0) {
                                category3VOs = l3Categories.stream().map(l3 -> new Category3VO(l2.getCatId(),
                                        l3.getCatId(), l3.getName())).collect(Collectors.toList());
                            }

                            return new Category2VO(v1.getCatId(), category3VOs, l2.getCatId(), l2.getName());
                        }).collect(Collectors.toList());
                    }
                    return category2VOs;

                }));
        return categoryMap;
    }

将循环查询数据库优化为先一次性查出所有的分类数据然后用stream的方式进行计算。

压测结果：

压测内容	压测线程数	吞吐量/s	90%响应时间(ms)	99%响应时间(ms)	主要原因
三级分类数据获取	50	168	504	868	数据库

缓存、缓存中间件、本地锁、分布式加锁

p151-pxxx 本部分主要讲解缓存原理和使用（包括缓存失效、缓存数据一致性等）、缓存中间件、本地锁、分布式加锁、Spring Cache

一、缓存、缓存中间件、分布式锁

1）、缓存使用

为了系统性能的提升，我们一般都会将部分数据放入缓存中，加速访问。而 db 承担数据落盘工作。

哪些数据适合放入缓存？

• 即时性、数据一致性要求不高的
• 访问量大且更新频率不高的数据（读多，写少）

举例：电商类应用，商品分类，商品列表等适合缓存并加一个失效时间(根据数据更新频率来定)，后台如果发布一个商品，买家需要 5 分钟才能看到新的商品一般还是可以接受的。

伪代码

data = cache.load(id);//从缓存加载数据
If(data == null){
    data = db.load(id);//从数据库加载数据
	cache.put(id,data);//保存到 cache 中
}
return data;

注意：在开发中，凡是放入缓存中的数据我们都应该指定过期时间，使其可以在系统即使没有主动更新数据也能自动触发数据加载进缓存的流程。避免业务崩溃导致的数据永久不一致问题。

本地缓存

例如在本类中使用Map集合作为缓存

本地缓存模式在分布式下的问题

本地缓存与当前工程在同一个项目里面，相当于是一个副本，

如果我们的应用只是一个单体应用，那永远不会有问题，很快就能查出来。

但是在分布式系统下，商品服务可能会部署在十几个服务器上，而每个服务都自带一个自己的本地缓存，那就会出现这样的问题：

问题一

举一个最简单的例子

第一次请求，负载均衡来到第一个微服务，第一个微服务查数据的时候，缓存中没有，它查了一次数据库，查到以后再放到缓存中，现在缓存中有数据了

如果下一次请求，还能负载均衡到第一个微服务，那我们就可以直接从缓存中拿，

但如果没有负载到第一个微服务，来到了第二个微服务，这个微服务的缓存，相当于还是没有数据，还得再查一遍数据库，

如果第三次，来到第三个微服务，第三个微服务，还是没有数据，我们还是得查询数据库，

所以，这是第一种问题，由于这些缓存都是分开的、各顾各的，没有的话都得自己查一遍。

问题二

如果我们对数据进行了修改，比如：我们修改了三级分类的数据，为了能读取到正确的数据，我们一般还要改缓存的数据。

假设，第一次修改请求来到了1号服务器，我们把分类数据修改了，我们把它的缓存改了，

但是2号、3号服务器，里面的数据，我们没法改，因为负载均衡是来到1号的

那这样，以后所有的请求，只要是定位到2号、3号服务器的，那它拿到的数据就是不一样的

这样，就产生了数据一致性的问题。

分布式缓存解决方案：

各个服务的缓存集中存放到一个缓存中间件中（最常用的是redis）

​ 我们将所有商品服务的缓存数据，放到一个缓存的中间件中，大家都给集中的一个地方缓存数据，这样，比如负载均衡，请求来到第1个服务器，第1个服务器查询缓存中没有，就会到数据库查出数据，1份用来显示，1份用来给缓存。

​ 假设第二次请求，来到了第2个服务器，由于第1次查询已经给缓存中放过数据了，所以第2个服务器就可以直接从缓存中查出数据，那就不需要再调用复杂的业务逻辑查询了，包括第3个服务器、乃至第N个，都可以直接从缓存中拿到数据。同样的，如果数据发生了修改，假设有一次请求来到了3号服务器，那不但会修改数据库，并且缓存中的数据也将被修改。这样，就是其他的请求来到了别的服务器，因为它们操作的都是一个地方的缓存，所以就不会出现数据不一致的问题。

​ 这就是分布式缓存的解决方案，不应该把缓存放在它本地的进程里面，而是要跟其它服务器共享一个集中式的缓存中间件。

​ 用缓存中间件的好处就是，如果装了一台redis，系统的数据量不够，或者性能也不足，我们适用redis搭建一个集群，包括可以让它分片存储，也就是说第1个存1-10000号数据，第2个存10001-20000号数据，依次类推，理论上，容量会做一个无限量的提升。

打破了本地缓存的容量限制，使用集中式的缓存中间件，使用起来也简单，而且单独维护缓存中间件，我们也可以做到高可用，高性能，

所以我们将使用redis作为缓存。

2）、整合 redis 作为缓存

1、pom中引入依赖：

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-redis</artifactId>
		</dependency>

2、yml文件中配置：

spring
  redis:
    host: 192.168.2.190
    port: 6379

3、使用 RedisTemplate 操作 redis

代码示例：自动注入了RedisTemplate即可使用

	@Autowired
	StringRedisTemplate stringRedisTemplate;

	@Test
	public void testStringRedisTemplate(){
		ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
		ops.set("hello","world_"+ UUID.randomUUID().toString());
		String hello = ops.get("hello");
		System.out.println(hello);
    }

为了让外面使用方便，spring还提供了StringRedisTemplate类：

泛型是String,String,key使用String类型的序列化机制来做的，value也是String类型的序列化做的，

3）、缓存-缓存使用-改造三级分类业务

p153

@Override
public Map<String, List<Catalog2Vo>> getCatalogJson() {
    //给缓存中放json字符串，拿出json字符串，还能逆转为能用的对象类型，【序列化与反序列化】

    //1、加入缓存逻辑，缓存中存储的是json字符串。
    //JSON跨语言，跨平台兼容
    String catalogJSON = stringRedisTemplate.opsForValue().get("catalogJSON");
    if (StringUtils.isEmpty(catalogJSON)) {
        //2、缓存中没有，查询数据库
        Map<String, List<Catalog2Vo>> catalogJsonFromDb = getCatalogJsonFromDb();

        //3、将查到的数据放到缓存,将对象转为json放到缓存中
        String s = JSON.toJSONString(catalogJsonFromDb);
        stringRedisTemplate.opsForValue().set("catalogJSON", s);
        return catalogJsonFromDb;
    }
    //转为我们指定的对象
    Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJSON, new TypeReference<Map<String, List<Catalog2Vo>>>() {
    });

    return result;
}
	
	更改以前查询方法的名字
    //从数据库查询并封装分类数据
    public Map<String, List<Catalog2Vo>> getCatalogJsonFromDb() {}

4）、缓存-缓存使用-压力测试出的内存泄露及解决

p154

对上面改造的分类接口进行压测，会出现大量异常，

老师的机器出现堆外异常：出现了大量的OutOfDirectMemoryError（堆外内存溢出异常）

问题分析：

• SpringBoot 2.0 以后默认使用 lettuce 作为操作 redis 的客户端，它底层使用 netty 进行网络通信

• lettuce 的 bug 导致 netty 操作堆外内存溢出；

• 如果外面对netty 没有指定堆外内存，则netty会默认使用 -Xmx 参数指定的内存（之前我们在idea中配了300m的），

• 可以通过 -Dio.netty.maxDirectMemory的vm参数来进行设置netty的内存大小；

  • 不能只使用 -Dio.netty.maxDirectMemory 去调大堆外内存，这样只会延缓异常出现的时间。

我们压测的时候内存设置的是300MB，我们也压了很久才出现异常，我们即使设置的再大，迟早也会出现内存溢出异常的，因为netty底层自己在计数，数字超过默认的容量限制，就会抛异常，这个就是堆外溢出异常。netty统计内存使用量，操作完了就会减内存使用量，一定是lettcure客户端，在哪一块操作的时候，没有及时调用掉减内存，导致堆外内存溢出，除了升级就是更换。

在源码中有异常原因如下图示：

解决方案有两种：

1、升级 lettuce 客户端

2、更换 jedis 客户端

lettuce 客户端使用的是netty，其优势是netty 的吞吐量是极大的，而 jedis 已经很久不更新了，

这两者都是操作 redis 的客户端，我们使用的 RedisTemplate 是对两者的进一步封装，所以在使用方面没有什么变化，

目前我们暂时选择 jedis ，等发布到生产环境时，我们还是会换回 lettuce，到时候会再监控整套日志，复现并定位这些错误。

下面我们切换成jedis

5）、切换使用 jedis

p154

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-redis</artifactId>
			<exclusions>
                # 需要先排除掉lettuce
				<exclusion>
					<groupId>io.lettuce</groupId>
					<artifactId>lettuce-core</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		
		# 然后再引入jedis
		<dependency>
			<groupId>redis.clients</groupId>
			<artifactId>jedis</artifactId>
		</dependency>
	</dependencies>

最终压力测试接口性能

优化1、2、3全开：

压测内容	压测线程数	吞吐量/s	90%响应时间(ms)	99%响应时间(ms)	主要原因
三级分类数据获取	50	707	87	138	数据库

ps : lettuce和jedis是操作redis的底层客户端，RedisTemplate是再次封装

6)、 缓存失效

缓存穿透

缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起为id为“-1”的数据或id为特别大不存在的数据。这时的用户很可能是攻击者，攻击会导致数据库压力过大。

解决：缓存空对象、布隆过滤器、mvc拦截器

假设有100W个请求，进来查询一个不存在的数据，因为我们的业务逻辑是，如果缓存不命中，就会查询数据库，所以会请求100W次数据库，会导致系统崩溃。

穿透，指查询一个永不存在的数据

缓存雪崩

缓存雪崩是指在我们设置缓存时key采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

解决方案：

• 规避雪崩：缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。
• 如果缓存数据库是分布式部署，将热点数据均匀分布在不同缓存数据库中。
• 设置热点数据永远不过期。
• 出现雪崩：降级 熔断，
• 事前：尽量保证整个 redis 集群的高可用性，发现机器宕机尽快补上。选择合适的内存淘汰策略。
• 事中：本地ehcache缓存 + hystrix限流&降级，避免MySQL崩掉，
• 事后：利用 redis 持久化机制保存的数据尽快恢复缓存=。

假设我们为缓存中放了非常多的数据，比如：商品、分类、品牌等等，包括商品数据从1-7000号都有，

但是我们在放数据的时候，每一个数据都设置了相同的过期时间，

举一个例子，放的时候，我们这1W个数据都是同一时间放进去的，它们拥有相同的过期时间，

到了那个时间之后，这些数据在缓存中过期了，redis自动将其移除，

此时有100W并发进来，正好都是查这1W条数据，这1W个数据在缓存中都没有，

它们就去查询数据库，然后导致系统崩溃

雪崩，指我们存的数据大面积失效

缓存击穿

缓存雪崩和缓存击穿不同的是：

• 缓存击穿 指 并发查同一条数据。缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力
• 缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。

解决方案：

• 设置热点数据永远不过期。
• 加互斥锁：业界比较常用的做法，是使用mutex。简单地来说，就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db去数据库加载，而是先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX或者Memcache的ADD）去set一个mutex key，当操作返回成功时，再进行load db的操作并回设缓存；否则，就重试整个get缓存的方法。

7)、加锁解决缓存击穿问题

p156

不好的方法是synchronized(this)，肯定不能这么写 ，不具体写了

锁时序问题：之前的逻辑是查缓存没有，然后取竞争锁查数据库，这样就造成多次查数据库。

解决方法：竞争到锁后，再次确认缓存中没有，再去查数据库。

1、实现代码

给查数据库的方法加上锁

得到锁之后，再去缓存中查询一次，如果没有，再继续查询

    @Override
    public Map<Long, List<Category2VO>> getCategoryJson() {
        String categoryJson = stringRedisTemplate.opsForValue().get("category.json");
        if (StringUtils.isEmpty(categoryJson)) {
            System.out.println("缓存不命中，将要查询数据库");
            Map<Long, List<Category2VO>> categoryJsonMap = getCategoryJsonFromDB();
            
            String newCategoryJson = JSON.toJSONString(categoryJsonMap);

            // 设置过期时间，解决缓存雪崩
            stringRedisTemplate.opsForValue().set("category.json", newCategoryJson, 1, TimeUnit.DAYS);

            return categoryJsonMap;
        }

        System.out.println("缓存命中，直接返回");
        // 不为空，则将JSON数据反序列化成页面需要的数据        
        return JSON.parseObject(categoryJson, new TypeReference<Map<Long, List<Category2VO>>>() {
        });
    }


    public Map<Long, List<Category2VO>> getCategoryJsonFromDB() {
        synchronized (this) {
            // 得到锁之后，应该再去缓存中确定一次，如果缓存有，就直接返回，没有再去查询
            String categoryJson = stringRedisTemplate.opsForValue().get("category.json");
            if (StringUtils.isNotEmpty(categoryJson)) {
                System.out.println("从缓存中获取数据");
                return JSON.parseObject(categoryJson, new TypeReference<Map<Long, List<Category2VO>>>() {
                });
            }
            System.out.println("查询了数据库");
			// 查询数据库...
        }
    }

2、测试

结果发现查询了两次数据库

3、原因

​ 假设现在有100W并发，进来之后这些线程先去看缓存，结果缓存里面都没有，都准备去查询数据库

以其中一台机器为例

​ 查数据库的时候，上来就加了一把锁，确保只让一个线程进来，假设为A线程，它在查询数据库之后，结果放入缓存之前，就将锁释放掉了，

​ 此时，锁住的B线程进来了，它进来之后，也是先确认缓存中有没有，此时A线程刚释放锁，要往缓存中放数据，然而放数据是一次网络交互，可能会很慢，包括系统刚启动、还要为redis建立连接、还要整线程池、线程池还没有初始化等等，所以第一次操作是一个很慢的过程，假设会花费30ms的时间。

B线程在这30ms中是无法获取到缓存数据的，也就是说A线程还没来得及放进去，B线程就去获取了，所以拿不到缓存，最终B线程又会去查询数据库，它查完之后，又释放锁，C线程进来。

可能C线程刚进来，A线程的数据才放到缓存中，B线程的数据还没放完，所以C线程判断缓存有没有数据的时候，可能判断的就是A线程之前给里边放的缓存数据，所以C线程就不会查询数据库了。

最终A线程查询了一次，B线程查询了一次

4、解决方法：

需要将结果放入缓存这个操作也放到锁中，这样就不会导致锁不住，查询两遍数据库了

    @Override
    public Map<Long, List<Category2VO>> getCategoryJson() {
        String categoryJson = stringRedisTemplate.opsForValue().get("category.json");
        if (StringUtils.isEmpty(categoryJson)) {
            System.out.println("缓存不命中，将要查询数据库");
            Map<Long, List<Category2VO>> categoryJsonMap = getCategoryJsonFromDB();
            return categoryJsonMap;
        }

        System.out.println("缓存命中，直接返回");
        return JSON.parseObject(categoryJson, new TypeReference<Map<Long, List<Category2VO>>>() {
        });
    }


    public Map<Long, List<Category2VO>> getCategoryJsonFromDB() {
        synchronized (this) {
            String categoryJson = stringRedisTemplate.opsForValue().get("category.json");
            if (StringUtils.isNotEmpty(categoryJson)) {
                System.out.println("从缓存中获取数据");
                return JSON.parseObject(categoryJson, new TypeReference<Map<Long, List<Category2VO>>>() {
                });
            }
            System.out.println("查询了数据库");
            // 查询数据库...
            
            // 将数据库结果放入缓存
            String newCategoryJson = JSON.toJSONString(categoryMap);

            stringRedisTemplate.opsForValue().set("category.json", newCategoryJson, 1, TimeUnit.DAYS);

            return categoryMap;            
        }
    }

5、分析原因总结

​ 无论我们是给方法块、还是代码块上加锁，都是将当前实例作为锁，当前实例在我们容器中是单实例，但是我们是一个服务对应一个容器，里面的每一个 this 只能代表当前实例的对象，以上图为例，八个容器就有八个锁， 每一个 this 都是不同的锁，最终导致的现象就是，第1个商品服务的一把锁，锁住了1W个请求，只放进了一个请求，第2个乃至后面的商品服务，都是如此，

有几台机器，最终就会放几个线程进来，那就相当于有8个线程同时进来，去数据库查相同的数据

比如：我们可以多创建几个商品服务，来模拟这种测试

idea模拟启动多个服务

由网关负载均衡到某一个商品服务，通过Nginx-> 网关 -> 多个商品服务，来进行压测

压测前的设置如下（为了多启动几个服务测试）修改port端口：

测试发现，每一个服务里都会查询一次数据库，这就是本地锁带来的问题

结论：在分布式环境下，使用本地锁，只能锁住当前服务的进程，所以我们需要使用分布式锁。

8）、分布式加锁（逐步演进）

p158

​ 在分布式环境下，要实现只让一个请求进来，其它进程全部锁住，我们就需要用分布式锁，它的缺点就是性能会差一些

1、基本原理

每个服务在执行业务之前，都需要先拿到锁，锁只有一把，谁能先抢到，谁就可以先执行这个业务，抢不到的，必须等待锁释放，才能进行下一次抢锁。

redis语法

set key value [EX secondes] [PX milliseconds] [NX|XX]

2、分布式锁演进-阶段一

简要代码如下：

1、在具体的查询数据库的业务方法里面去掉本地锁

    private Map<Long, List<Category2VO>> getCategoryJsonFromDB() {
    	// 查询数据库前去掉本地锁
    }

2、通过redis占坑加锁的阶段一代码：

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
        // 去redis占锁
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111");
        if (lock) {
            // 加锁成功，执行业务
            Map<Long, List<Category2VO>> categoryMap = getCategoryJsonFromDB();

            // 业务执行完，需要删除锁，别人就可以来占锁了
            stringRedisTemplate.delete("lock");
            return categoryMap;
        } else {
            // 加锁失败，休眠200ms重试
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
            // 重试，使用自旋的方式，模仿本地sync监听锁
            return getCategoryJsonWithRedisLock();
        }
    }

2、分布式锁演进-阶段二

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111");
        if (lock) {
            Map<Long, List<Category2VO>> categoryMap = getCategoryJsonFromDB();

            // 设置过期时间【重点改进了这里】
            stringRedisTemplate.expire("lock", 300, TimeUnit.MILLISECONDS);

            stringRedisTemplate.delete("lock");
            return categoryMap;
        } else {
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return getCategoryJsonWithRedisLock();
        }
    }

3、分布式锁演进-阶段三

加锁的同时，设置过期时间，保证整个操作是原子性的

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
    	// 加锁的同时，设置过期时间，保证整个操作是原子性的
        // 成功就都成功，失败就都失败
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111", 300, TimeUnit.MILLISECONDS);
        if (lock) {
            Map<Long, List<Category2VO>> categoryMap = getCategoryJsonFromDB();
            stringRedisTemplate.delete("lock");
            return categoryMap;
        } else {
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return getCategoryJsonWithRedisLock();
        }
    }

问题一
假设一开始A线程抢到了锁，一开始设置的锁过期时间为10秒，执行业务的时候，由于业务较为复杂，执行了30秒，

等要去删锁的时候，其实已经过期了，redis里面已经没有了，这还是比较好的情况

问题二
最坏的情况是这样，执行业务的代码超时了，花费了30秒

这30秒发生了很多事情

A线程，在业务执行到第10秒的时候，锁就过期了，redis把锁删除了，

此时，外面的线程都在等着抢占锁，结果发现锁可以抢了，直接就去抢锁，

B线程抢到了锁，又开始执行业务，它执行到第10秒的时候，它的锁也过期了，

C线程又抢到了锁，又开始执行业务，它执行到第10秒的时候，它的锁也过期了，

D线程又抢到了锁

由于是同步的过程，此时的A线程已经执行了30秒，也就是把业务执行完了，然后它会手动删除锁，但是在这30秒期间，A、B、C的锁早就因为过期自动被删了，

所以它真正删除的是D线程的锁，而D线程还在执行业务，它的锁一旦被删除，又会导致其它线程抢到锁，

如此循环下去，就会使这个锁失去作用

结论
就算业务超时，锁也会因为我们设置的过期时间，自动释放，别的线程就可以抢到锁，等到真的要手动删除锁的时候，很有可能删除的就是别人的锁。

由此引发的两个问题：

1. 业务超时了怎么办
2. 业务超时之后，由于自己的锁早就过期，当要手动删除锁的时候，把别人的锁删除了怎么办

4、分布式锁演进-阶段四

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
        String token = UUID.randomUUID().toString();        
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", token, 300, TimeUnit.MILLISECONDS);
        if (lock) {
            Map<Long, List<Category2VO>> categoryMap = getCategoryJsonFromDB();

            // 删除锁的时候，判断token，是自己的删，不是自己的不删
            if (stringRedisTemplate.opsForValue().get("lock").equals(token)) {
                stringRedisTemplate.delete("lock");
            }
            return categoryMap;
        } else {
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return getCategoryJsonWithRedisLock();
        }
    }

问题
假如说，业务执行到删除锁这里，由于删除锁的命令，需要到远程服务器拿到redis的数据，再让远程服务器将数据返回，

这中间是要花费一定时间的

如果锁在10秒过期，我们的业务已经执行到9.5秒了，彻底取到值要花费0.8秒，我们去服务器取数据，假设花费了0.3秒，

让服务器将数据返回给我们，这期间又要花费0.5秒，然而数据才刚走到一半，锁就过期了，然后就被B线程抢到了，

又过了0.3秒，数据回到了A线程，A线程判断这个值的确是当时设置的token，然后就将锁删了，殊不知它的锁早就因为过期被自动删了，

而它删除的，正是当前B线程的锁，相当于是给误删了。

结论
获取值+对比成功删除的操作不是原子操作，导致了上述问题的发生

5、分布式锁演进-阶段五（最终形式）

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
        String token = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", token, 300, TimeUnit.MILLISECONDS);
        if (lock) {
            Map<Long, List<Category2VO>> categoryMap = getCategoryJsonFromDB();

            // 这段脚本的意思是，如果获取key对应的值是传过来的值，那就调用删除方法，否则返回0
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis call('del', KEYS[1]) else return 0 end";
            // 原子删锁
            Long result = stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                    Arrays.asList("lock", token));
            return categoryMap;
        } else {
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return getCategoryJsonWithRedisLock();
        }

问题

假设我们的业务执行时间超长，我们就需要给锁自动续期

当然最简单的方法就是给锁设置的时间长一些， 比如说，设置个300秒，哪个业务也不可能让它执行300秒，我们不会等它

6、阶段六

    private Map<Long, List<Category2VO>> getCategoryJsonWithRedisLock() {
        String token = UUID.randomUUID().toString();
        // 不想做自动续期，只需要加大锁的过期时间即可
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", token, 300, TimeUnit.SECONDS);
        if (lock) {
            Map<Long, List<Category2VO>> categoryMap;
            System.out.println("获取分布式锁成功");

            // 无论执行业务出现崩溃还是怎么了，我们最终都会解锁
            try {
                categoryMap = getCategoryJsonFromDB();
            } finally {
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                Long result = stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                        Arrays.asList("lock"), token);
            }
            return categoryMap;
        } else {
            System.out.println("获取分布式锁失败...继续等待重试");            
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return getCategoryJsonWithRedisLock();
        }
    }

压测一下

7、总结

分布式锁核心分为两部分

1. 使用NX/EX对Redis进行原子加锁
2. 使用lua脚本对进行原子解锁

二、分布式锁Redisson

1 、简介

​ Redisson 是架设在 Redis 基础上的一个 Java 驻内存数据网格（In-Memory Data Grid）。充分的利用了 Redis 键值数据库提供的一系列优势， 基于 Java 实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。 使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

官方文档：https://github.com/redisson/redisson/wiki/目录

redisson的功能有很多，我们主要看分布式锁这部分

2、简单测试

		<!-- redisson-->
		<dependency>
			<groupId>org.redisson</groupId>
			<artifactId>redisson</artifactId>
			<version>3.12.0</version>
		</dependency>

gulimall-product/src/main/java/com/indi/gulimall/product/config/RedissonConfig.java

package com.indi.gulimall.product.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {
    /**
     * 所有对Redisson的使用都是通过RedissonClient对象
     * @return
     */
    @Bean(destroyMethod = "shutdown")   // 服务停止以后，会调用shutdown进行销毁
    RedissonClient redission() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.2.190:6379");
        return Redisson.create(config);
    }
}

gulimall-product/src/test/java/com/indi/gulimall/product/GulimallProductApplicationTests.java

    @Autowired
    BrandService brandService;

    @Resource
    private RedissonClient redissonClient;

    @Test
    public void testRedisson(){
        System.out.println(redissonClient);
    }

    @Test
    public void testRedis(){
        ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();

上述代码地址：

https://gitee.com/UnityAlvin/gulimall/commit/9238af02e6bfcc8b6478d71493cff9f931bceebc

3、可重入锁与不可重入锁

分布式锁：github.com/redisson/redisson/wiki/8.-分布式锁和同步器

可重入锁

A调用B。AB都需要同一锁，此时可重入锁就可以重入，A就可以调用B。不可重入锁时，A调用B将死锁

// 参数为锁名字
RLock lock = redissonClient.getLock("CatalogJson-Lock");//该锁实现了JUC.locks.lock接口
lock.lock();//阻塞等待
// 解锁放到finally // 如果这里宕机：有看门狗，不用担心
lock.unlock();

假设现在有A、B两个方法，A方法调用了B方法，A方法加了一把1号锁，B方法也想加这个1号锁，

如果是可重入的，那整个流程就应该是这样，

A方法执行之后，把1号锁加上了，里面调用了B方法，B方法一看，A方法已经加上1号锁了，直接就拿过来用了，B方法内部就可以直接执行，执行完之后，A释放锁

不可重入锁

还是A方法先执行，然后把1号锁持有了，里面调用了B方法，而B则需要等待A方法释放1号锁之后，它才能抢到1号锁，这就是不可重入锁。

这种锁是有问题的，A在调用B之前，压根就没释放过锁，所以B根本就拿不到这个锁，

A会等待B执行完之后才会释放锁，B连锁都拿不到，又怎么会执行呢？所以A也释放不了锁，最终会导致死锁。

结论

所以的锁，都应该设计成可重入锁，避免死锁问题

4、Redisson锁测试

对标ReentrantLock

    @ResponseBody
    @GetMapping("/hello")
    public String hello() {
        RLock lock = redissonClient.getLock("anyLock");
        try {
            lock.lock();
            try {
                System.out.println("加锁成功，执行业务..." + Thread.currentThread().getId());
                TimeUnit.SECONDS.sleep(30);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } finally {
            lock.unlock();
            System.out.println("释放锁..." + Thread.currentThread().getId());
        }
        return "hello";
    }

同时开启10000、10001两个商品服务，

​ 假设10000先抢到锁，它先执行业务，此时10001则是在外面等待，直到10000释放锁之后，10001才抢到了锁，等10001执行完，然后才释放锁

​ 假设还是10000先抢到锁，它在执行业务期间宕机了，没有释放锁，我们发现10001会一直在外面等待，最终抢到锁，然后执行业务，再释放锁，并没有出现死锁的现象。

我们发现Redisson内部的lock()实现，里面有一个死循环，会一直去获取锁。

• lock()是阻塞式等待，默认加的锁都是30s时间
• 如果执行业务时间过长，运行期间Redisson会给锁自动续期，每次都会续上30s，不会因为业务时间过长，导致锁自动删掉
• 等业务执行完，就不会给当前锁续期，即使不手动释放锁，锁也会在30s以后自动删除

5、Redisson看门狗原理

p161

未指定锁的过期时间

lock.lock();

如果我们未指定锁的过期时间（使用lock.lock()的方式），那Redisson内部会调用this.lock(-1L, (TimeUnit)null, false);

然后再调用Long ttl = tryAcquire(leaseTime, unit, threadId);

然后再调用return get(tryAcquireAsync(leaseTime, unit, threadId));

tryAcquireAsync方法如下：

上面的代码里面接着调用了如下

上面一步一步的调用下来，我们发现也是通过脚本的方式给redis发送命令（去占坑）

指定锁的过期时间

如果我们想 指定锁的过期时间 则使用如下方法：

lock.lock(10, TimeUnit.SECONDS); // 10 指定的锁的过期时间

如果我们指定了锁的过期时间，那Redisson内部会调用this.lock(leaseTime, unit, false);，

然后会调用以下方法

结论

推荐使用lock.lock(10, TimeUnit.SECONDS);这种方式省掉了自动续期，理由如下：

​ 指定一个长过期时间+手动解锁解决业务超时，比如说过期时间设置为30s，如果每个业务都能超过30s，那说明肯定是业务内部有问题，抛出或捕获异常更好。

6、Redisson读写锁测试

对标ReentrantReadWriteLock P162集

• 读写锁通常都是成对出现的
• 写锁控制了读锁
• 只要写锁存在，读锁就得等待
• 并发写，肯定得一个一个执行
• 如果写锁不存在，读锁一直在那加锁，那跟没加是一样的

不加锁用例

不加锁的测试的接口代码如下：

    /**
     * 什么锁也不加，单纯的往redis中写入一个值
     *
     * @return
     */
    @ResponseBody
    @GetMapping("/write-unlock")
    public String writeUnlock() {
        String uuid = "";
        try {
            uuid = UUID.randomUUID().toString();
            TimeUnit.SECONDS.sleep(30);
            stringRedisTemplate.opsForValue().set("writeValue", uuid);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return uuid;
    }

    /**
     * 什么锁也不加，单纯的从redis中读取一个值
     *
     * @return
     */
    @ResponseBody
    @GetMapping("/read-unlock")
    public String readUnlock() {
        String name = "";
        name = stringRedisTemplate.opsForValue().get("writeValue");
        return name;
    }

这种情况下可以随便读随便写（前端页面调用可以随便调用读写接口，都能有返回正确的值）

加锁用例一

    @ResponseBody
    @GetMapping("/write")
    public String write(){
        // getReadWriteLock()方法是获取的读写锁
        RReadWriteLock lock = redissonClient.getReadWriteLock("rwAnyLock");
        String uuid = "";
        RLock rLock = lock.writeLock();// writeLock()加上写锁，这个是排他锁，请求多了会排队
        try{
            rLock.lock();
            uuid = UUID.randomUUID().toString();
            TimeUnit.SECONDS.sleep(30);
            stringRedisTemplate.opsForValue().set("writeValue",uuid);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            rLock.unlock();
        }
        return uuid;
    }

    @ResponseBody
    @GetMapping("/read")
    public String read(){
        // getReadWriteLock()方法是获取的读写锁
        RReadWriteLock lock = redissonClient.getReadWriteLock("rwAnyLock");
        String uuid=  "";
        RLock rLock = lock.readLock();// readLock()加上读锁，这是共享锁，每次请求会立马返回
        try{
            rLock.lock();
            uuid = stringRedisTemplate.opsForValue().get("writeValue");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            rLock.unlock();
        }
        return uuid;
    }

测试一
先给redis中，添加一个writeValue的值，调用读取接口查看是否能读取到
结果：可以读取到

测试二
先调用写的方法，再调用读的方法
结果：读请求会一直加载（排队等待），等写请求执行完业务之后，读请求瞬间加载到数据
加读写锁的作用就是，保证一定能读到最新数据。

修改期间，写锁是一个互斥锁，读锁则是一个共享锁

读 + 读：相当于无锁，并发读，只会在redis中记录好当前的读锁，它们都会同时加锁成功

写 + 读：读必须等待写锁释放(必须等待读的方法执行完成释放了锁才能成功)

加锁用力二

    @ResponseBody
    @GetMapping("/write")
    public String write() {
        RReadWriteLock lock = redissonClient.getReadWriteLock("rwAnyLock");
        String uuid = "";
        RLock rLock = lock.writeLock();
        try {
            rLock.lock();
            // 打印log
            System.out.println("写锁加锁成功" + Thread.currentThread().getId());
            uuid = UUID.randomUUID().toString();
            TimeUnit.SECONDS.sleep(30);
            stringRedisTemplate.opsForValue().set("writeValue", uuid);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            rLock.unlock();
            // 打印log
            System.out.println("写锁释放" + Thread.currentThread().getId());
        }
        return uuid;
    }

    @ResponseBody
    @GetMapping("/read")
    public String read() {
        RReadWriteLock lock = redissonClient.getReadWriteLock("rwAnyLock");
        String uuid = "";
        RLock rLock = lock.readLock();
        try {
            rLock.lock();
            // 打印log
            System.out.println("读锁加锁成功" + Thread.currentThread().getId());
            uuid = stringRedisTemplate.opsForValue().get("writeValue");
            // 让读锁也等待30秒
            TimeUnit.SECONDS.sleep(30);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            rLock.unlock();
            // 打印log
            System.out.println("读锁释放" + Thread.currentThread().getId());
        }
        return uuid;
    }

测试三
先发送一个读请求，再发送一个写请求
结果：读加上锁了，读释放锁之后，写才加上锁

测试四
发送一个写请求，再发送四个读请求
结果：写请求释放的瞬间，四个读请求都加上锁了
写 + 写：阻塞方式

读 + 写：有读锁，写也需要等待

总结：只要有一个写存在，其它的读/写就必须等待

7、 闭锁（CountDownLatch）

P164集

跟JUC下的CountDownLatch是一样的用法

代码用例

    @ResponseBody
    @GetMapping("/go-home/{id}")
    public String goHome(@PathVariable Integer id) {
        RCountDownLatch home = redissonClient.getCountDownLatch("home");
        home.countDown();
        return id + "班已走";
    }

    @ResponseBody
    @GetMapping("/lock-door")
    public String lockDoor() throws InterruptedException {
        RCountDownLatch home = redissonClient.getCountDownLatch("home");
        home.await();
        return "锁门";
    }

测试过程：

​ 先为redis的home设置个3，先发送lock-door请求锁门，发现界面一直是加载中，此时发送3次go-home请求，让3个班回家，执行完第3次的时候，发现lock-door的界面刷新了，提示锁门。

8、Redisson信号量测试

p165集

用例一

以车库停车举例，如下：

    @ResponseBody
    @GetMapping("/park")// 停车的接口
    public String park() throws InterruptedException {
        RSemaphore park = redissonClient.getSemaphore("park");
        park.acquire(); // 获取一个
        return "空闲车位-1";
    }

    @ResponseBody
    @GetMapping("/go")
    public String go() {
        RSemaphore park = redissonClient.getSemaphore("park");
        park.release(); // 释放一个车位
        return "空闲车位+1";
    }

acquire()是一个阻塞方法，必须要获取成功，否则就一直阻塞

测试过程如下：

我们假设现在这个停车场有3个车位（在redis中新建一个key为‘park’，对应值为3），先发送3次go请求，添加3个空闲车位，在redis中，发现park的值为3，再发送park请求，每发送1次，redis的park就会-1，

执行第4次的时候，界面不动了，一直在加载

此时执行1次go请求，发现go请求刚执行完，空闲车位加了1个，park请求也执行完了，使空闲车位减了1个，最终park的值为0

用例二

使用tryAcquire()方法：

    @ResponseBody
    @GetMapping("/try-park")
    public String tryPark() {
        RSemaphore park = redissonClient.getSemaphore("park");
        boolean result = park.tryAcquire();// tryAcquire()尝试获取一下，不行就算了
        if (result) {
            // 执行复杂业务代码。。。
            return "空闲车位-1";
        } else {
            // 错误提示。。。
            return "空闲车位已满";
        }
    }

tryAcquire()尝试获取一下，不行就算了

还是用例一的那种情况，当try-park请求，执行到4次的时候，直接提示了空闲车位已满

结论：
信号量也可以用作分布式限流，在做分布式限流的时候，可以判断信号量是否为true，为true则执行业务，否则直接返回错误，告诉它当前流量过大

三、缓存一致性解决

初始方案

锁的粒度，越细越快 ("categoryJsonLock"这个名称关系到锁的粒度)

    /**
     * 使用RedissonLock解决缓存击穿
     *
     * @return
     */
    private Map<Long, List<Category2VO>> getCategoryJsonWithRedissonLock() {
        // 锁的粒度，越细越快("categoryJsonLock"这个名称关系到锁的粒度)
        RLock rLock = redissonClient.getReadWriteLock("categoryJsonLock").readLock();
        Map<Long, List<Category2VO>> categoryMap;
        try{
            rLock.lock();
            categoryMap = getCategoryJsonFromDB();
        } finally {
            rLock.unlock();
        }
        return categoryMap;
    }

缓存和数据库一致性常用解决方案

双写模式图示：

数据库改过的值，到我们最终看到的值，中间有一个比较大的延迟时间，无论怎么延迟，最终都会看到数据库最新修改的值。

失效模式图示：

1号机器写数据，删缓存

2号机器，它想把1号数据改成2，但是它操作比较慢，花的时间比较长，

1号机器刚删除完缓存，

3号机器就进来了，它去读缓存，结果发现没数据，然后就去读数据库，此时因为2号请求还没改完数据库，所以3号请求读到了1号机器写入的数据，

然后，3号机器要更新缓存了，如果它执行的快还好，顶多刚更新，然后碰上2号机器删缓存，相当于什么也没干，缓存没更新

如果它执行的慢，让2号机器把数据写完，再把缓存删了，那就没有人能阻拦3号机器更新缓存了，最终会把1号机器的数据更新到缓存，导致缓存不一致。

高并发时存在的问题：

• 双写模式：写数据库后，再写缓存
  • 问题：并发时，线程2写进入，写完DB后都去写缓存，如果1先对字段修改为a，然后2线程将值修改成b，然后线程1接着去写a到缓存的时候被2线程抢先了，就会导致最后缓存里的值是a。
    • 问题是：有暂时的脏数据
• 失效模式：写完数据库后，删缓存
  • 问题：还没存入数据库呢，线程2又读到旧的DB了
  • 解决：缓存设置过期时间，定期更新
  • 解决：写数据写时，加分布式的读写锁。

解决方案：

• 如果是用户纬度数据（订单数据、用户数据），这种并发几率非常小，不用考虑这个问题，缓存数据加上过期时间，每隔一段时间触发读的主动更新即可
• 如果是菜单，商品介绍等基础数据，也可以去使用canal订阅binlog的方式
• 缓存数据+过期时间也足够解决大部分业务对于缓存的要求。
• 通过加锁保证并发读写，写写的时候按顺序排好队。读读无所谓。所以适合使用读写锁。（业务不关心脏数据，允许临时脏数据可忽略）；

总结：

• 我们能放入缓存的数据本就不应该是实时性、一致性要求超高的。所以缓存数据的时候加上过期时间，保证每天拿到当前最新数据即可。
• 我们不应该过度设计，增加系统的复杂性
• 遇到实时性、一致性要求高的数据，就应该查数据库，即使慢点。

3、缓存数据一致性-解决框架-Canal

Canal是阿里开源的一个中间件，可以模拟成数据库的从服务器

比如说MySQL有一个数据库，如果我们装了Canal，Canal就会将自己伪装成MySQL的从服务器，

从服务器的特点就是，MySQL里面只要有什么变化，它都会同步过来

正好利用Canal的这个特性，只要我们的业务代码更新了数据库，我们的MySQL数据库肯定得开启Binlog（二进制日志），这个日志里面有MySQL每一次的更新变化，Canal就假装成MySQL的从库，把MySQL每一次的更新变化都拿过来，相当于MySQL只要有更新，Canal就知道了。

比如说它看到MySQL分类数据更新了，那它就去把Redis里边所有跟分类有关的数据更新了就行

用这种的好处就是，我们在编码期间改数据库就行，不需要管缓存的任何操作，Canel在后台只要看数据改了就自动都更新了。

缺点就是，又加了一个中间件，又得开发一些自定义的功能。

最终方案

我们谷粒商城最终采用的方案是如下：（失效模式+分布式读写锁）

1. 选择失效模式
2. 缓存的所有数据都有过期时间，即使有脏数据，下一次查询数据库也能触发主动更新
3. 读写数据的时候，加上分布式的读写锁，
   • 如果经常写还经常读，肯定会对系统的性能产生极大的影响
   • 如果偶尔写一次还经常读，那对系统性能一点也不影响

四、谷粒商城Spring Cache

1、介绍

可以随便找篇cache文章阅读：https://blog.csdn.net/er_ving/article/details/105421572---

• 每次都那样写缓存太麻烦了，Spring 从 3.1 开始定义了 org.springframework.cache.Cache 和 org.springframework.cache.CacheManager 接口来统一不同的缓存技术；并支持使用 JCache（JSR-107）注解简化我们开发；

• Cache 接口为缓存的组件规范定义，包含缓存的各种操作集合；Cache 接 口 下 Spring 提 供 了 各 种 xxxCache 的 实 现 ； 如 RedisCache、 EhCacheCache 、ConcurrentMapCache 等；

• 每次调用需要缓存功能的方法时，Spring 会检查指定参数指定的目标方法是否已经被调用过；如果有就直接从缓存中获取方法调用后的结果，如果没有就调用方法并缓存结果后返回给用户。下次调用直接从缓存中获取。

• 使用 Spring 缓存抽象时我们需要关注以下两点：

  • 确定方法需要被缓存以及他们的缓存策略
  • 从缓存中读取之前缓存存储的数据

2、基础概念

缓存管理器CacheManager定义规则， 真正实现缓存CRUD的是缓存组件，如ConcurrentHashMap、Redis

3、整合SpringCache开发

引入依赖

引入spring-boot-starter-cache、spring-boot-starter-data-redis依赖

        <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-cache</artifactId>
        </dependency>
        
        <!-- redis的starter已经引过，此处就不写了-->

写配置

自动配置：

• CacheAutoConfiguration 会导入 RedisCacheConfiguration;

• 会自动装配缓存管理器 RedisCacheManager;

• 观察源码如下：

  • // 缓存自动配置源码
    @Configuration(proxyBeanMethods = false)
    @ConditionalOnClass(CacheManager.class)
    @ConditionalOnBean(CacheAspectSupport.class)
    @ConditionalOnMissingBean(value = CacheManager.class, name = "cacheResolver")
    @EnableConfigurationProperties(CacheProperties.class)
    @AutoConfigureAfter({ CouchbaseAutoConfiguration.class, HazelcastAutoConfiguration.class,
                         HibernateJpaAutoConfiguration.class, RedisAutoConfiguration.class })
    @Import({ CacheConfigurationImportSelector.class, // 看导入什么CacheConfiguration
             CacheManagerEntityManagerFactoryDependsOnPostProcessor.class })
    public class CacheAutoConfiguration {
    
        @Bean
        @ConditionalOnMissingBean
        public CacheManagerCustomizers cacheManagerCustomizers(ObjectProvider<CacheManagerCustomizer<?>> customizers) {
            return new CacheManagerCustomizers(customizers.orderedStream().collect(Collectors.toList()));
        }
    
        @Bean
        public CacheManagerValidator cacheAutoConfigurationValidator(CacheProperties cacheProperties,
                                                                     ObjectProvider<CacheManager> cacheManager) {
            return new CacheManagerValidator(cacheProperties, cacheManager);
        }
    
        @ConditionalOnClass(LocalContainerEntityManagerFactoryBean.class)
        @ConditionalOnBean(AbstractEntityManagerFactoryBean.class)
        static class CacheManagerEntityManagerFactoryDependsOnPostProcessor
            extends EntityManagerFactoryDependsOnPostProcessor {
    
            CacheManagerEntityManagerFactoryDependsOnPostProcessor() {
                super("cacheManager");
            }
    
        }
    
    

  • @Configuration(proxyBeanMethods = false)
    @ConditionalOnClass(RedisConnectionFactory.class)
    @AutoConfigureAfter(RedisAutoConfiguration.class)
    @ConditionalOnBean(RedisConnectionFactory.class)
    @ConditionalOnMissingBean(CacheManager.class)
    @Conditional(CacheCondition.class)
    class RedisCacheConfiguration {
    
        @Bean // 放入缓存管理器
        RedisCacheManager cacheManager(CacheProperties cacheProperties, 
                                       CacheManagerCustomizers cacheManagerCustomizers,
                                       ObjectProvider<org.springframework.data.redis.cache.RedisCacheConfiguration> redisCacheConfiguration,
                                       ObjectProvider<RedisCacheManagerBuilderCustomizer> redisCacheManagerBuilderCustomizers,
                                       RedisConnectionFactory redisConnectionFactory, ResourceLoader resourceLoader) {
            RedisCacheManagerBuilder builder = RedisCacheManager.builder(redisConnectionFactory).cacheDefaults(
                determineConfiguration(cacheProperties, redisCacheConfiguration, resourceLoader.getClassLoader()));
            List<String> cacheNames = cacheProperties.getCacheNames();
            if (!cacheNames.isEmpty()) {
                builder.initialCacheNames(new LinkedHashSet<>(cacheNames));
            }
            redisCacheManagerBuilderCustomizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
            return cacheManagerCustomizers.customize(builder.build());
        }
    
    

常用注解：

@Caching	相当于组合@Cachable、@CacheEvice、@CachePut
@CacheConfig	在类级别共享缓存的相同配置

yml配置：

#spring
  cache:
    type: redis	# 配置使用redis作为缓存

测试使用缓存

开启缓存功能 : 启动类上添加@EnableCaching注解

使用直接完成缓存操作
为需要使用缓存功能的方法添加@Cacheable({"category"})注解，并指定分区（最好按照业务进行分区）

它实现的效果就是，每次调用需要缓存功能的方法时，Spring 会检查指定参数指定的目标方法是否已经被调用过；如果有就直接从缓存中获取方法调用后的结果，如果没有就调用方法并缓存结果后返回给用户。下次调用直接从缓存中获取。

现在配置就完成了，直接可以在想缓存的方法上加注解@Cacheable({"category"})即可让缓存生效，如果仅配置默认的配置（上面的这些）则SpringCache的默认行为如下：

• 如果缓存中有数据，则被调用的方法不执行，而是直接去缓存中拿

• redis中的key是默认自动生成的：

  • key：SimpleKey

  • value：默认使用jdk进行序列化（可读性差），默认ttl为-1永不过期（如果自定义序列化方式需要编写配置类）

    • value的取值有很多，也支持SPEL表达式，如下：
    • 

自定义配置

如果我们对默认配置不满意，如序列化不满意、设置过期时间不满意等则需要如下配置：

1. 指定生成的缓存使用的key：

   key 属性，接收一个SpEL表达式

	// 使用指定的值作为key
	@Cacheable(value = {"categories"}, key = "'level1Categories'")
	
	// 使用方法名作为key
    @Cacheable(value = {"categories"}, key = "#root.method.name")	

2. 指定缓存数据的存活时间

   配置文件中修改

   spring
     cache:
       redis:
         time-to-live: 3600000	# 单位ms
   

自定义缓存配置

​ 我们希望将数据保存为json格式，需要修改SpringCache的自定义配置，这个比较麻烦，我们先看一下它的加载过程，如下：

原理

配置类：

package com.atguigu.gulimall.product.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
* <p>Title: MyCacheConfig</p>
* Description：redis配置
* date：2020/6/12 16:38
*/
@EnableConfigurationProperties(CacheProperties.class)
@EnableCaching
@Configuration
public class MyCacheConfig {



   //@Autowired
   //CacheProperties cacheProperties;  这样注入不优雅，不如直接写在下面的方法形参里，这样也会获取到容器中的CacheProperties
   /**
    * 配置文件中 TTL设置没用上
    *
    * 原来:
    * @ConfigurationProperties(prefix = "spring.cache")
    * public class CacheProperties
    *
    * 现在要让这个配置文件生效	: @EnableConfigurationProperties(CacheProperties.class) 将文件绑定到这个配置类？
    *
    */
   // fixme 此处加入的形参cacheProperties 是可以从spring容器中bean自动获取的；如果不写在这里，也可以直接从上边用@Autowired注入
   @Bean
   RedisCacheConfiguration redisCacheConfiguration(CacheProperties cacheProperties){

   	RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();

   	// 设置kv的序列化机制
   	config = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()));
   	config = config.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
   	CacheProperties.Redis redisproperties = cacheProperties.getRedis();

   	// 设置配置 因为源码中配置了，这里需要绑定我们自己写的配置文件
   	if(redisproperties.getTimeToLive() != null){
   		config = config.entryTtl(redisproperties.getTimeToLive());
   	}
   	if(redisproperties.getKeyPrefix() != null){
   		config = config.prefixKeysWith(redisproperties.getKeyPrefix());
   	}
   	if(!redisproperties.isCacheNullValues()){
   		config = config.disableCachingNullValues();
   	}
   	if(!redisproperties.isUseKeyPrefix()){
   		config = config.disableKeyPrefix();
   	}
   	return config;
   }

}

测试其它配置

spring
  cache:
    redis:      
      key-prefix: CACHE_	# 指定key的前缀，不指定则使用缓存的名字作为前缀
      use-key-prefix: true	# 是否使用前缀
      cache-null-values: true	# 是否缓存空值，可以防止缓存穿透

缓存使用@Cacheable@CacheEvict

下面代码第一个方法存放缓存（使用@Cacheable注解），第二个方法清空缓存（使用@CacheEvict注解）

// 调用该方法时会将结果缓存，缓存名为category，key为方法名
// sync表示该方法的缓存被读取时会加锁 // value等同于cacheNames // key如果是字符串"''"
@Cacheable(value = {"category"},key = "#root.methodName",sync = true)
public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithSpringCache() {
    return getCategoriesDb();
}

//调用updateCascade()方法会删除缓存category下的所有cache，如果要删除某个具体，用key="''"
/**
 *  todo 级联更新所有关联的数据
 * @param category
 */
@CacheEvict(value = "category", key = "'getLevel1Category'") // 注意：这里是SPEL，要加''
// 触发从缓存中删除
@Transactional
@Override
public void updateCascade(CategoryEntity category) {
    this.updateById(category);
        categoryBrandRelationService.updateCategory(category.getCatId(),category.getName());
}
如果要清空多个缓存，用@Caching(evict={@CacheEvict(value="")})

@CacheEvict 相当于缓存”失效模式“

@CacheEvict(value = "category", key = "'getLevel1Category'") // 注意：这里是SPEL，要加''

删除一个分区的缓存数据

    // 删除一个分区的缓存数据
    @CacheEvict(value = "category", key = "'level1Categories'")
    public void updateDetail(CategoryEntity category) {
    }

删除多个分区的缓存数据

	// 删除多个分区的缓存数据
    @Caching(evict = {
            @CacheEvict(value = "category", key = "'level1Categories'"),
            @CacheEvict(value = "category", key = "'categoryJson'"),
    })    

删除指定分区的缓存数据

	// 删除category分区下的缓存数据，也就是清除模式
	@CacheEvict(value = "category",allEntries = true)
	// 如果希望修改完数据，再往缓存里放一份，也就是双写模式，可以使用这个注解
	@CachePut

双写模式，可以使用这个注解 @CachePut （方法需要有返回值）

推荐配置

spring
  cache:
    redis:      
     #  key-prefix: CACHE_	# 不指定key的前缀，不指定则使用缓存的名字作为前缀
      use-key-prefix: true	# 是否使用前缀
      cache-null-values: true	# 是否缓存空值，可以防止缓存穿透

不指定key的前缀，不指定则使用缓存的名字作为前缀，这样就会如下图示，看着非常清晰明了

4、 SpringCache原理与不足

1）、读模式

缓存穿透：查询一个null数据。

• SpringCache解决方案：缓存空数据，可通过spring.cache.redis.cache-null-values=true

缓存击穿：大量并发进来同时查询一个正好过期的数据。

• SpringCache解决方案：加锁 ? 默认是无加锁的;

  • 加锁，防止高并发，限制同一时间只能有一个服务获取锁，访问数据库
    在实现类上加注解（此处加的是本地锁）

    @Cacheable(value = {"category"},key = "#root.method.name",sync = true)
    

缓存雪崩：大量的key同时过期。

​ 其实只要不是超大型并发，比如十几W的key同时过期，正好碰上十几W的请求过来查询，不用考虑这个问题

​ 原来的解决方案是加随机时间，但是很容易弄巧成拙，假设数据不是同一时间放进去的，比如第1个数据是3秒过期，然后加了个随机数1秒，4秒过期，第2个数据是2秒过期，然后加了个随机数2秒，还是4秒过期，本来什么都不加的时候，它们过期时间不会冲撞在一起，结果有可能一加，倒还让他们撞在一起了，当时每一个数据存储的时间节点其实是不一样的，所以，只要指定过期时间就行

• SpringCache解决方案：加随机时间或者加不同的过期时间，防止key同时过期

spring
  cache:
    redis:
      time-to-live: 3600000	# 设置缓存过期时间，单位ms

2)、写模式：（缓存与数据库一致）

1. 读写加锁。
2. 引入Canal，感知到MySQL的更新去更新Redis
3. 读多写多，直接去数据库查询就行

3）、 原理

CacheManager -> Cache -> Cache负责缓存的读写

4）、总结：

常规数据（读多写少，即时性，一致性要求不高的数据，完全可以使用Spring-Cache（只要缓存的数据有过期时间即可））：

写模式(只要缓存的数据有过期时间就足够了)

特殊数据：需要特殊设计

原理部分也可参考文章：

1、 Spring-Cache的基本使用和缺点

​ https://blog.csdn.net/m0_56466015/article/details/123428624

五、谷粒商城搭建检索页面

1. 修改 hosts 配置文件

路径：C:\Windows\System32\drivers\etc

在后面追加以下内容：

# gulimall #
你的ip		search.gulimall.com

2. 配置 nginx 动静分离

Xftp 进入 /mydata/nginx/html/static 目录，创建 search 文件夹，将搜索页相关的静态资源放进去

3. 修改 nginx 配置文件

vi /mydata/nginx/conf/conf.d/gulimall.conf

修改配置如下：

4、配置网关

增加配置如下

        # 搜索服务路由
        - id: gulimall_search_route
          uri: lb://gulimall-search
          predicates:
            - Host=search.gulimall.com

原来的如下：需要去掉**.gulimall.com,

        - id: gulimall_host_route
          uri: lb://gulimall-product
          predicates:
            - Host=**.gulimall.com,gulimall.com # 需要去掉**.gulimall.com,

确认已经将域名转发携带上：（确认有如下图配置）

4. 其它

参考：gulimall_product_dsl.json

其他ElasticSearch相关参考文章： (137条消息) 【谷粒商城】ElasticSearch、上架与检索_hancoder的博客-CSDN博客

六、异步&线程池

1、线程回顾

1）、初始化线程的 4 种方式

1）、继承 Thread
2）、实现 Runnable 接口
3）、实现 Callable 接口 + FutureTask （可以拿到返回结果，可以处理异常）
4）、线程池
方式 1 和方式 2：主进程无法获取线程的运算结果。不适合当前场景
方式 3：主进程可以获取线程的运算结果，但是不利于控制服务器中的线程资源。可以导致
服务器资源耗尽。
方式 4：通过如下两种方式初始化线程池

Executors.newFiexedThreadPool(3);
//或者
new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit unit, workQueue, threadFactory, 

​ 通过线程池性能稳定，也可以获取执行结果，并捕获异常。但是，在业务复杂情况下，一个异步调用可能会依赖于另一个异步调用的执行结果。

2)、线程池的七大参数

运行流程：

1、线程池创建，准备好 core 数量的核心线程，准备接受任务

2、新的任务进来，用 core 准备好的空闲线程执行。
(1) 、core 满了，就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队列获取任务执行
(2) 、阻塞队列满了，就直接开新线程执行，最大只能开到 max 指定的数量
(3) 、max 都执行好了。Max-core 数量空闲的线程会在 keepAliveTime 指定的时间后自动销毁。最终保持到 core 大小
(4) 、如果线程数开到了 max 的数量，还有新任务进来，就会使用 reject 指定的拒绝策略进行处理

3、所有的线程创建都是由指定的 factory 创建的。

面试：

一个线程池 core 7； max 20 ，queue：50，100 并发进来怎么分配的；
先有 7 个能直接得到执行，接下来 50 个进入队列排队，在多开 13 个继续执行。现在 70 个被安排上了。剩下 30 个默认拒绝策略来执行。

3）、常见的 4 种线程池

• newCachedThreadPool

  •  创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

• newFixedThreadPool

  •  创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

• newScheduledThreadPool

  •  创建一个定长线程池，支持定时及周期性任务执行。

• newSingleThreadExecutor

  •  创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。（会从队列中挨个拿任务顺序执行）

4）、开发中为什么使用线程池

• 降低资源的消耗
  •  通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗
• 提高响应速度
•  因为线程池中的线程数没有超过线程池的最大上限时，有的线程处于等待分配任务的状态，当任务来时无需创建新的线程就能执行
• 提高线程的可管理性
  • 线程池会根据当前系统特点对池内的线程进行优化处理，减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源，还降低系统的稳定性，使用线程池进行统一分配

2、CompletableFuture 异步编排

业务场景：
查询商品详情页的逻辑比较复杂，有些数据还需要远程调用，必然需要花费更多的时间。

假如商品详情页的每个查询，需要如下标注的时间才能完成
那么，用户需要 5.5s 后才能看到商品详情页的内容。很显然是不能接受的。
如果有多个线程同时完成这 6 步操作，也许只需要 1.5s 即可完成响应。

Future 是 Java 5 添加的类，用来描述一个异步计算的结果。你可以使用isDone方法检查计算是否完成，或者使用get阻塞住调用线程，直到计算完成返回结果，你也可以使用cancel 方法停止任务的执行。

虽然Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的 CPU 资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？

很多语言，比如 Node.js，采用回调的方式实现异步编程。Java 的一些框架，比如 Netty，自己扩展了 Java 的 Future接口，提供了addListener等多个扩展方法；Google guava 也提供了通用的扩展 Future；Scala 也提供了简单易用且功能强大的 Future/Promise 异步编程模式。

作为正统的 Java 类库，是不是应该做点什么，加强一下自身库的功能呢？

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合 CompletableFuture 的方法。CompletableFuture 类实现了 Future 接口，所以你还是可以像以前一样通过get方法阻塞或者轮询的方式获得结果，但是这种方式不推荐使用。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果。

1）、创建异步对象

CompletableFuture 提供了四个静态方法来创建一个异步操作。

static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。

1、runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的
2、可以传入自定义的线程池（第二个参数executor），否则就用默认的线程池；

package com.atguigu.gulimall.product.thread;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadTest1 {

     public static ExecutorService executorService = Executors.newFixedThreadPool(8); // 放到成员变量位置 加上static
    /**
     * 测试使用CompletableFuture
     */
    public static void main(String[] args) throws ExecutionException, InterruptedException {


        // ExecutorService executorService = Executors.newFixedThreadPool(8); 放到成员变量位置
        System.out.println("main()线程开始执行。。。。。。");

        // 创建异步对象CompletableFuture
        // 1.1我们使用我们自己的线程池来执行异步任务

       /* // 1.1 使用无返回值的runAsync()
        CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());

            int i = 10 / 2;
            System.out.println("当前线程" + Thread.currentThread().getName() + "运行结果：" + i);
        }, executorService);
*/
        // 1.2 使用有返回值的supplyAsync()
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());

            int i = 10 / 2;
            System.out.println("当前线程" + Thread.currentThread().getName() + "运行结果：" + i);
            return i;
        }, executorService);
        Integer integer = future.get();
        System.out.println("main()线程执行完毕。。。返回结果：" + integer);
    }
}

2）、计算完成时回调方法

public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor);

public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn);

whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。

whenComplete 和 whenCompleteAsync 的区别：
whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture future = CompletableFuture.supplyAsync(new Supplier<Object>() {
            @Override
            public Object get() {
                System.out.println(Thread.currentThread().getName() + "\t
                        completableFuture");
                int i = 10 / 0;
                return 1024;
            }
        }).whenComplete(new BiConsumer<Object, Throwable>() {// 链式调用whenComplete
            @Override
            public void accept(Object o, Throwable throwable) {
                System.out.println("-------o=" + o.toString());
                System.out.println("-------throwable=" + throwable);
            }
        }).exceptionally(new Function<Throwable, Object>() {// 链式调用exceptionally
            @Override
            public Object apply(Throwable throwable) {
                System.out.println("throwable=" + throwable);
                return 6666;// exceptionally()可以自定义返回的东西（即：修改返回结果）
            }
        });
        System.out.println(future.get());
    }
}

exceptionally(new Function<Throwable, Object>() {// 链式调用exceptionally 。。。 也可以写成上面的（） -> {} 的lambda形式

3）、handle 方法

public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

和 complete 一样，可对结果做最后的处理（可处理异常，无论执行成功还是失败），可改变返回值。

4）、线程串行化方法

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值（可以让下一步的执行任务获取这一步的结果）。

thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果（当前方法无返回值）。

thenRun 方法：只要上面的任务执行完成，就开始执行 thenRun，只是处理完任务后，执行thenRun 的后续操作

带有 Async 默认是异步执行的。同之前。

以上都要前置任务成功完成。

Function<?super T, ? extends U>
T：上一个任务返回结果的类型
U：当前任务的返回值类型

5）、两任务组合 - 都要完成

指定的两个任务必须都完成之后才能触发该任务。

thenCombine：组合两个 future，并可以获取两个 future 的返回结果，并返回当前任务的返回值（3个任务都能有返回值，）

thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值。

runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后，处理该任务

6）、两任务组合 - 一个完成

当两个任务中，任意一个 future 任务完成的时候，就可以执行该任务。

applyToEither：两个任务有一个执行完成，获取它的返回值，处理任务并有新的返回值。

acceptEither：两个任务有一个执行完成，获取它的返回值，处理任务，没有新的返回值。

runAfterEither：两个任务有一个执行完成，不需要获取 future 的结果，处理任务，也没有返回值。

7）、多任务组合

allOf：等待所有任务完成

• 在使用allof(cfs...)方法的时候有个get()方法，这个方法的意思是阻塞等待所有的任务完成之后main线程才结束

anyOf：只要有一个任务完成

如果上面的get()方法注掉则会导致main线程结束后才有“查询商品介绍”任务的结果，这样是不合理的，所以应该使用get()方法，这个方法的意思是阻塞等待所有的任务完成之后main线程才结束，如下：

七、商品详情

1、配置环境

1）、hsot文件

添加hosts内容 192.168.56.10 item.gulimall.com

# 谷粒商城主页
192.168.56.10 gulimall.com
# 检索页面
192.168.56.10 search.gulimall.com
# 商品详情页面
192.168.56.10 item.gulimall.com

2）、nginx配置：

3）、gateway：

修改网关 使item路由到produc

        - id: gulimall_host_route
          uri: lb://gulimall-product
          predicates:
            - Host=gulimall.com,item.gulimall.com # 注意这里是以请求中的Host主机信息配置的断言路由规则

4）、页面相关资源

从我们的资料中复制详情页的html到product服务下的src/main/resources/templates文件夹下，静态资源文件则放到nginx的/mydata/nginx/html/static/item文件夹下

这样以后有访问路径为“/static/”的则去如下配置的文件夹中找（即上面/mydata/nginx/html/static/item文件夹下）

修改详情html文件中跳转路径位置

1、 商品详情VO

观察我们要建立怎样的VO

标签：缓存,return,--,lock,redis,压测,线程,public
From： https://www.cnblogs.com/qipaoxian/p/17852802.html