Java打怪之路谷粒商城Nginx与Redis

　　一、商品上架流程
　　product
　　1、根据spuid查询所有sku信息
　　2、将sku信息封装成skuModel（向es中存储的是skuModelVO）
　　不一致信息：
　　库存信息
　　发送远程调用查询系统中是否存在库存
　　热度评分
　　品牌和分类的名字
　　检索属性
　　查出当前sku中所有可以被检索的属性
　　3、将数据发给es进行保存，调用gulimallsearch微服务进行保存
　　（1）在es中建立索引，建立号映射关系（docjsonproductmapping。json）
　　（2）在es中保存这些数据
　　（3）设置成功失败状态二、Nginx反向代理配置
　　1、什么是nginx反向代理
　　正向代理：我们要访问谷歌，但是被墙，所以将请求发送给代理服务器，由代理服务器来访问，这叫正向代理
　　反向代理：服务器部署在内网中，客户端访问服务器不能直接访问服务器的内网地址，通常需要根据域名来访问（www。baidu。com）。所以需要反向代理服务器来将外网的ip转换成内网ip来访问。
　　2、反向代理如何实现
　　只代理到商品服务：客户端发送gulimall。com请求，在windows中映射gulimall。com是虚拟机ip（192。168。56。10）所以会访问虚拟机。虚拟机中的nginx监听80端口。在配置文件中，会设置nginx代理到的路径http：192。168。56。1：10000。这样就访问到了商品服务里的首页（首页在商品服务中）location｛proxypasshttp：192。168。56。1：10000｝
　　代理到网关：客户端发送gulimall。com请求，在windows中映射gulimall。com是虚拟机ip（192。168。56。10）所以会访问虚拟机。虚拟机中的nginx监听80端口。在配置文件中nginx。config中配置上游服务器gulimallhttp：192。168。56。1：88。然后继续在配置文件gulimall。config中nginx代理到的路径http：gulimall。
　　需要注意的是，nginx代理给网关会丢失一部分信息。解决如下
　　三、压力测试
　　1、JVM堆栈垃圾回收
　　2、优化方式
　　加内存（提升JVM堆中eden区和old区的内存空间，避免频繁的gc）
　　数据库优化，加索引
　　打开模板缓存，设置日志为error时记录
　　静态资源快速返回动静分离
　　优化代码减少循环查表
　　3、Nginx实现动静分离
　　（1）将所有的静态资源移动到nginx中
　　（2）static下的所有内容由nginx返回，在nginx中进行相关配置，gulimallconf中设置locationstatic｛rootusrsharenginxhtml｝
　　4、优化三级分类数据获取
　　直接查询所有分类的结果，只查一次表，将其他循环查表的过程抽取为一个函数四、缓存与分布式锁为什么使用缓存如何使用redis
　　引入dataredisstarterdependencygroupIdorg。springframework。bootgroupIdspringbootstarterdataredisartifactIddependency
　　简单配置redis的host
　　使用自动配置好的StringRedisTemplateredis
　　StringRedisTemplateredis中存储键值对键与值都是String类型将数据放入缓存OverridepublicMapLong，ListCatelog2VogetCatelogJson（）｛StringcatelogJsonredisTemplate。opsForValue（）。get（catelogJson）；if（StringUtils。isEmpty（catelogJson））｛缓存中没有MapLong，ListCatelog2VocatelogJsonFromDBgetCatelogJsonFromDB（）；redisTemplate。opsForValue（）。set（catelogJson，JSON。toJSONString（catelogJsonFromDB））；returncatelogJsonFromDB；｝因为转化的对象是复杂对象，所以通过TypeReferenceMapLong，ListCatelog2VocatelogJsonFromDBJSON。parseObject（catelogJson，newTypeReferenceMapLong，ListCatelog2Vo（）｛｝）；returncatelogJsonFromDB；｝从数据库查询数据publicMapLong，ListCatelog2VogetCatelogJsonFromDB（）｛将数据库的多次交互，转为一次，一次性查询所有数据ListCategoryEntityallListbaseMapper。selectList（null）；查出所有分类ListCategoryEntitylevel1CategorysgetParentcid（allList，0L）；分装数据MapLong，ListCatelog2VoresultMaplevel1Categorys。stream（）。collect（Collectors。toMap（CategoryEntity：：getCatId，v｛每一个的一级分类，查到这个一级分类的二级分类ListCategoryEntitylistgetParentcid（allList，v。getCatId（））；ListCatelog2Vocatelog2VoLif（！StringUtils。isEmpty（list））｛catelog2VoListlist。stream（）。map（item｛Catelog2Vocatelog2VonewCatelog2Vo（v。getCatId（）。toString（），null，item。getCatId（）。toString（），item。getName（））；封装二级分类的三级分类ListCategoryEntityentityListgetParentcid（allList，item。getCatId（））；if（！StringUtils。isEmpty（entityList））｛ListCatelog2Vo。Catelog3Vocatelog3VosentityList。stream（）。map（m｛Catelog2Vo。Catelog3Vocatelog3VonewCatelog2Vo。Catelog3Vo（item。getCatId（）。toString（），m。getCatId（）。toString（），m。getName（））；returncatelog3Vo；｝）。collect（Collectors。toList（））；catelog2Vo。setCatalog3List（catelog3Vos）；｝returncatelog2Vo；｝）。collect（Collectors。toList（））；returncatelog2VoL｝returncatelog2VoL｝））；returnresultM｝privateListCategoryEntitygetParentcid（ListCategoryEntityallList，Longparentcid）｛ListCategoryEntitycollectallList。stream（）。filter（item｛returnitem。getParentCid（）。equals（parentcid）；｝）。collect（Collectors。toList（））；returnbaseMapper。selectList（newQueryWrapperCategoryEntity（）。eq（parentcid，v。getCatId（）））；｝内存溢出以及解决
　　TODO产生堆外内存溢出OutOfDirectMemoryError：
　　产生原因springboot2。0以后默认使用lettuce作为操作redis的客户端，它使用netty进行网络通信lettuce的bug导致netty堆外内存溢出。netty如果没有指定堆外内存，默认使
　　用Xms的值，可以使用Dio。netty。maxDirectMemory进行设置
　　解决方案：由于是lettuce的bug造成，不要直接使用Dio。netty。maxDirectMemory去调大虚拟机堆外内存，治标不治本。升级lettuce客户端。切换使用jedis
　　lettuce和jedis是操作redis的底层客户端，RedisTemplate是再次封装font缓存失效穿透、雪崩、击穿
　　缓存穿透：是指大量的并发请求访问一个数据库中从未出现的内容，由于缓存中没有这条内容，所以会并发查表导致数据库崩溃。
　　解决办法：将null也写入缓存中
　　缓存雪崩：是指设置缓存时，key使用了相同的过期时间，大量并发此时同时访问这些过期缓存，导致同时查表导致数据库崩溃。
　　解决办法：缓存过期时间在原来的基础上设置随机值。
　　缓存击穿：是指大量并发请求同时访问一个热点key，如果这个热点key刚好在缓存中过期，导致这个key的数据查询都落到数据库导致数据库崩溃。
　　解决办法：大量的并发只让一个人去查，其他人等待，查到之后释放锁，其他人回去到锁，查找缓存不会查找数据库。加锁方法使用synchronized（this），this表示当前对象。
　　注意：缓存时许问题，确认缓存、查数据库和结果放入缓存是一个原子操作，要在锁内实现。
　　OverridepublicMapLong，ListCatelog2VogetCatelogJson（）｛StringcatelogJsonredisTemplate。opsForValue（）。get（catelogJson）；缓存中没有if（StringUtils。isEmpty（catelogJson））｛获取数据返回MapLong，ListCatelog2VocatelogJsonFromDBgetCatelogJsonFromDB（）；returncatelogJsonFromDB；｝因为转化的对象是复杂对象，所以通过TypeReferenceMapLong，ListCatelog2VocatelogJsonFromDBJSON。parseObject（catelogJson，newTypeReferenceMapLong，ListCatelog2Vo（）｛｝）；returncatelogJsonFromDB；｝从数据库查询数据publicMapLong，ListCatelog2VogetCatelogJsonFromDB（）｛synchronized（this）｛判断缓存是否已经有数据，防止之前的线程已经放好数据StringcatelogJsonredisTemplate。opsForValue（）。get（catelogJsonFromDB）；if（！StringUtils。isEmpty（catelogJson））｛因为转化的对象是复杂对象，所以通过TypeReferenceMapLong，ListCatelog2VoresultMapJSON。parseObject（catelogJson，newTypeReferenceMapLong，ListCatelog2Vo（）｛｝）；returnresultM｝将数据库的多次交互，转为一次，一次性查询所有数据ListCategoryEntityallListbaseMapper。selectList（null）；查出所有分类ListCategoryEntitylevel1CategorysgetParentcid（allList，0L）；分装数据MapLong，ListCatelog2VoresultMaplevel1Categorys。stream（）。collect（Collectors。toMap（CategoryEntity：：getCatId，v｛每一个的一级分类，查到这个一级分类的二级分类ListCategoryEntitylistgetParentcid（allList，v。getCatId（））；ListCatelog2Vocatelog2VoLif（！StringUtils。isEmpty（list））｛catelog2VoListlist。stream（）。map（item｛Catelog2Vocatelog2VonewCatelog2Vo（v。getCatId（）。toString（），null，item。getCatId（）。toString（），item。getName（））；封装二级分类的三级分类ListCategoryEntityentityListgetParentcid（allList，item。getCatId（））；if（！StringUtils。isEmpty（entityList））｛ListCatelog2Vo。Catelog3Vocatelog3VosentityList。stream（）。map（m｛Catelog2Vo。Catelog3Vocatelog3VonewCatelog2Vo。Catelog3Vo（item。getCatId（）。toString（），m。getCatId（）。toString（），m。getName（））；returncatelog3Vo；｝）。collect（Collectors。toList（））；catelog2Vo。setCatalog3List（catelog3Vos）；｝returncatelog2Vo；｝）。collect（Collectors。toList（））；returncatelog2VoL｝returncatelog2VoL｝））；放入缓存redisTemplate。opsForValue（）。set（catelogJson，JSON。toJSONString（resultMap），1L，TimeUnit。DAYS）；returnresultM｝｝privateListCategoryEntitygetParentcid（ListCategoryEntityallList，Longparentcid）｛ListCategoryEntitycollectallList。stream（）。filter（item｛returnitem。getParentCid（）。equals（parentcid）；｝）。collect（Collectors。toList（））；returnbaseMapper。selectList（newQueryWrapperCategoryEntity（）。eq（parentcid，v。getCatId（）））；｝
　　上面的本地锁只能锁住当前进程，在分布式的情况下无法保证锁住整个集群服务
　　分布式锁
　　可以让商品服务来占坑，当一个商品服务拿到锁后，其他商品服务必须等待
　　设置分布式锁需要考虑的问题
　　一二阶段为设置锁需要考虑的问题，二三阶段为删除锁需要考虑的问题第一阶段setnx会获取锁，然后执行业务，最后删除锁出现问题：当执行业务时发生宕机，无法执行删除锁的操作，其他线程无法拿到锁。造成死锁。解决问题：将锁设置一个自动过期时间，来保证锁肯定会被释放
　　方法名：getCateLogJsonFromDbWithRedisLock占分布式锁BooleanlockredisTemplate。opsForValue（）。setIfAbsent（lock，111）；if（lock）｛如果获得锁，执行业务MapString，ListCatelog2VodataFromDbgetDataFromDb（）；执行完删除锁redisTemplate。delete（lock）returndataFromDb；｝else｛returngetCateLogJsonFromDbWithRedisLock（）；｝
　　改进后为：方法名：getCateLogJsonFromDbWithRedisLock占分布式锁BooleanlockredisTemplate。opsForValue（）。setIfAbsent（lock，111）；if（lock）｛如果获得锁，执行业务MapString，ListCatelog2VodataFromDbgetDataFromDb（）；设置过期时间redisTemplate。expire（key，30，TimeUnit。SECONDS）执行完删除锁redisTemplate。delete（lock）returndataFromDb；｝else｛returngetCateLogJsonFromDbWithRedisLock（）；｝第二阶段出现问题：虽然设置了过期时间，但是如果在设置过期时间之前发生了宕机，过期时间没有设置上，又会导致死锁问题。解决问题：设置过期时间和占位必须是原子操作占分布式锁setIfAbsent方法的第三个参数是设置过期时间BooleanlockredisTemplate。opsForValue（）。setIfAbsent（lock，111，300）；if（lock）｛如果获得锁，执行业务MapString，ListCatelog2VodataFromDbgetDataFromDb（）；设置过期时间redisTemplate。expire（key，30，TimeUnit。SECONDS）执行完删除锁redisTemplate。delete（lock）returndataFromDb；｝else｛returngetCateLogJsonFromDbWithRedisLock（）；｝第三阶段
　　出现问题：由于设置了自动过期时间，如果当前执行的业务很长，删除锁的时候，锁已经自动过期了，此时删除的可能是别人的锁。
　　解决问题：给每个锁添加唯一的uuid，删除的时候判断一下，匹配的是自己的锁才能删除第四阶段
　　出现问题：判断完成之后，需要删除锁前，锁刚好过期，别人设置新的值，那么我们会删除别人的锁
　　解决问题：删除锁必须保证原子操作（判断与删除同时进行）采用redislua脚本实现StringuuidUUID。randomUUID（）。toString（）；设置redis分布式锁，30s自动删除锁BooleanisLockredisTemplate。opsForValue（）。setIfAbsent（lock，uuid，300L，TimeUnit。SECONDS）；if（isLock）｛抢锁成功执行业务MapLong，ListCatelog2VoresultMtry｛resultMapgetLongListMap（）；｝finally｛lua脚本解锁：让获取数据对比数据成为原子操作Stringscriptifredis。call（get，KEYS〔1〕）ARGV〔1〕thenreturnredis。call（del，KEYS〔1〕）elsereturn0LonglockredisTemplate。execute（newDefaultRedisScriptLong（script，Long。class），Arrays。asList（lock），uuid）；｝returnresultM｝else｛抢锁失败重试try｛Thread。sleep（200）；｝catch（InterruptedExceptione）｛e。printStackTrace（）；｝睡眠0。2s后，重新调用自旋returngetCatelogJsonFromDBWithRedisLock（）；｝RedissonRedisson使用pom导入dependencygroupIdorg。redissongroupIdredissonartifactIdversion3。13。4versiondependency开启配置，写一个配置类ConfigurationpublicclassRedissonConfig｛Bean（destroyMethodshutdown）publicRedissonClientredissonClient（）｛ConfigconfignewConfig（）；config。setTransportMode（TransportMode。EPOLL）；config。useClusterServers（）可以用rediss：来启用SSL连接。addNodeAddress（redis：192。168。109。101：6379）；returnRedisson。create（config）；｝｝可重入锁
　　基于Redis的Redisson分布式可重入锁RLockJava对象实现了java。util。concurrent。locks。Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。什么是可重入锁：A中调用B，A拿到锁后，该锁可以由B直接使用，这个锁叫可重入锁，不可重入锁是A拿到锁后，A调用B，B需要等待A锁的释放，而A锁需要等待B执行完才释放，造成死锁。如何使用Redisson分布式可重入锁RequestBodyGetMapping（hello）publicStringhello（）｛1、获取锁锁名字一样就是同一把锁RLocklockredission。getLock（lock）；指定解锁时间的锁2、枷锁lock。lock（）；阻塞式等待指定解锁时间的锁lock。lock（10，TimeUnit。SECONDS）try｛3、执行业务System。out。println（执行业务）；Thread。sleep（30000）；｝catch（Exceptione）｛｝finally｛解锁lock。unlock（）；｝｝
　　Question1：这与上节redis中自己设置的锁有什么区别
　　这里是对自己设置的锁进行了封装：上节我们需要自己设置加锁和解锁时的原子操作。而Redisson直接封装了这些原子操作。由于这里加锁会自动设置过期时间，所以不会出现业务崩溃，锁释放不了导致死锁问题。
　　Question2：Redisson可重入锁，如何保证业务时间长而导致锁被自动删除。Redisson可重入锁提供了锁的续期：如果业务时间超长，运行期间会自动给锁续上新的30s（看门狗默认时间）。这里有个疑问，看门狗在业务执行的时候会不断设置新的定时任务来给锁续期，如果此时业务崩溃，看门狗是失效了吗？如果是失效的话，那就相当于不会再继续进行续期，锁会在默认事件之后自动删除。
　　Question3：看门狗如何进行锁的续期
　　只要占锁成功，就会启动一个定时任务，重新给锁设置过期时间，新的过期时间就是看门狗的默认时间。每隔（一个看门狗时间3）10s，会自动再次续期，续到30s
　　Question4：如果指定自动解锁时间会发生什么
　　如果指定自动过期时间，看门狗会失效，不会再有锁的续期。自动解锁时间一定要大于业务执行时间。（如果小于会出现业务没执行完，锁已经被释放，此时别的进程抢占锁，当前进程删除锁是报错）。
　　分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。
　　读锁被称为共享锁，写锁被称为互斥锁
　　读读共享读写写写都是互斥RReadWriteLockrwlockredisson。getReadWriteLock（anyRWLock）；最常见的使用方法rwlock。readLock（）。lock（）；或rwlock。writeLock（）。lock（）；10秒钟以后自动解锁无需调用unlock方法手动解锁rwlock。readLock（）。lock（10，TimeUnit。SECONDS）；或rwlock。writeLock（）。lock（10，TimeUnit。SECONDS）；尝试加锁，最多等待100秒，上锁以后10秒自动解锁booleanresrwlock。readLock（）。tryLock（100，10，TimeUnit。SECONDS）；或booleanresrwlock。writeLock（）。tryLock（100，10，TimeUnit。SECONDS）；。。。lock。unlock（）；
　　生产者消费者问题中出现。例如停车与离开
　　信号量为存储在redis中的一个数字，当这个数字大于0时，即可以调用acquire（）方法增加数量，也可以调用release（）方法减少数量，但是当调用release（）之后小于0的话方法就会阻塞，直到数字大于0
　　tryAcquire（）返回一个布尔型量，表示当前是否可以获取数量，如果数量为0，返回false。但是不会阻塞。常用于分布式限流GetMapping（park）ResponseBodypublicStringpark（）｛RSemaphoreparkredissonClient。getSemaphore（park）；try｛park。acquire（2）；｝catch（InterruptedExceptione）｛e。printStackTrace（）；｝return停进2；｝GetMapping（go）ResponseBodypublicStringgo（）｛RSemaphoreparkredissonClient。getSemaphore（park）；park。release（2）；return开走2；｝缓存与数据一致性
　　在对表进行修改时，缓存中的内容也需要修改，可以采用两种模式来进行修改双写模式：在写完数据库之后，再写缓存。存在问题：会出现脏数据的问题失效模式：再写完数据库，删除缓存中的数据，下次再读数据库存在问题：也会产生脏数据解决脏数据问题：设置过期时间，数据过期之后下一次重新查找数据库读写数据时，加上分布式锁的读写锁，经常写经常读会对性能有影响
　　如果想要保证强一致性可以使用Canal
　　Canal相当于一个数据库的从库，业务更新数据库后，它会及时的修改缓存。
　　注意：实时更新的数据、一致性要求高的数据本就不应该放到缓存中，缓存中加上过期时间保证每天拿到的数据是当前最新数据就可。遇到实时更新的数据、一致性要求高的数据就应该查找数据库。