当前位置：首页 > news >正文

Redis应用场景（黑马点评快速复习）

news 2025/10/6 9:28:25

Redis基于Session实现登录

Redis商户查询缓存

实现思路：

缓存更新策略：

双写一致性

缓存穿透

缓存击穿

缓存雪崩

Redis工具类

Redis优惠卷秒杀

全局唯一ID

库存超卖问题分析

Redis分布式锁

基本原理和实现方式

Redis分布式锁的实现核心思路

Redis分布式锁误删情况说明

分布式锁的原子性问题

Redis分布式锁redission

Redission的介绍

分布式锁-redission可重入锁原理

分布式锁-redission锁重试和WatchDog机制

Redis消息队列

认识消息队列

Redis的消息队列

Redis达人探店

点赞功能

点赞排行榜

Redis好友关注

共同关注

Redis附近商户

Redis用户签到

实现签到功能

签到天数统计

Redis用户统计

Redis基于Session实现登录

每个tomcat中都有一份属于自己的session,假设用户第一次访问第一台tomcat，并且把自己的信息存放到第一台服务器的session中，但是第二次这个用户访问到了第二台tomcat，那么在第二台服务器上，肯定没有第一台服务器存放的session，所以此时整个登录拦截功能就会出现问题，我们能如何解决这个问题呢？早期的方案是session拷贝，就是说虽然每个tomcat上都有不同的session，但是每当任意一台服务器的session修改时，都会同步给其他的Tomcat服务器的session，这样的话，就可以实现session的共享了后来采用的方案都是基于redis来完成，我们把session换成redis，redis数据本身就是共享的，就可以避免session共享的问题了。

具体是实现思路是：当注册完成后，用户去登录会去校验用户提交的手机号和验证码，是否一致，如果一致，则根据手机号查询用户信息，不存在则新建，最后将用户数据保存到redis，并且生成token作为redis的key，当我们校验用户是否登录时，会去携带着token进行访问，从redis中取出token对应的value，判断是否存在这个数据，如果没有则拦截，如果存在则将其保存到threadLocal中，并且放行。因为我们的用户是保存在Redis中的，作为一个中间件，所有的tomcat都可以访问，这就实现了在所有的tomcat中共享session的功能了，每一个tomcat直接对这个公共的session操作即可。在数据结构的选择上，我们可以使用String，也可以使用hash结构，hash结构对内存有一定的占用，但是好处是可以对小key进行操作。因为在登录的时候我们只需要存储用户的信息，我们最后选择了用String来存储用户的一些基本信息。

在基于Redis实现共享session功能的时候，我们遇到了一个问题：状态登录刷新问题不及时。在这个方案中，他确实可以使用对应路径的拦截，同时刷新登录token令牌的存活时间，但是现在这个拦截器他只是拦截需要被拦截的路径，假设当前用户访问了一些不需要拦截的路径，那么这个拦截器就不会生效，所以此时令牌刷新的动作实际上就不会执行，所以这个方案他是存在问题的。关于这个问题，我们的解决方案是：既然之前的拦截器无法对不需要拦截的路径生效，那么我们可以添加一个拦截器，在第一个拦截器中拦截所有的路径，把第二个拦截器做的事情放入到第一个拦截器中，同时刷新令牌，因为第一个拦截器有了threadLocal的数据，所以此时第二个拦截器只需要判断拦截器中的user对象是否存在即可，完成整体刷新功能。简单来说就是，我们添加两个拦截器，第一个拦截器拦截所有路径，用来获取token查询Redis的用户以及刷新token的有效期；第二个拦截器专门用来判断用户的Threadlocal的用户是否存在，不存在表示用户未登录，我们直接拦截就行了；如果存在，就放行，正常去执行业务逻辑。如果有需要，我们可以生成多个拦截器来细化拦截器的具体功能。

Redis商户查询缓存

在我们日常开发的过程中，缓存是经常被使用到的，细化下来无论是前端还是后端都有自己的缓存工具，在平常的开发过程中，为了加快计算机的处理速度，提高用户的体验，我们通常会使用内存来缓存一些数据，以此来大大降低用户并发访问带来的服务器读写压力。在大多数情况下，内存的读写速度是远远高于磁盘的，这些是为什么内存会被用来当做缓存的来使用的一种重要原因。对于Java开发来说，最频繁被使用的缓存中间件就是Redis了，缓存也是Redis的一种重要的功能。在我们项目的开发过程中，所使用到的缓存中间件就是Redis。

实现思路：

标准的操作方式就是查询数据库之前先查询缓存，如果缓存数据存在，则直接从缓存中返回，如果缓存数据不存在，再查询数据库，然后将数据存入redis。如下图所示：

缓存更新策略：

缓存更新是redis为了节约内存而设计出来的一个东西，主要是因为内存数据宝贵，当我们向redis插入太多数据，此时就可能会导致缓存中的数据过多，所以redis会对部分数据进行更新，或者把他叫为淘汰更合适。关于Redis的常见的淘汰策略如下：

双写一致性

由于我们的缓存的数据源来自于数据库,而数据库的数据是会发生变化的,因此,如果当数据库中数据发生变化,而缓存却没有同步,这时的缓存就存在问题，这就是典型的Redis和数据库中的数据不一致问题。关于Redis和数据库的双写一致性，场景的方案无非就两个：一个先删缓存，再更新数据库；另一个就是先更新数据库，再删除缓存，具体解决逻辑图如下：

第一种方式：先删除缓存再更新数据库

假设线程1先把Redis中的缓存删除了，此时线程2进来了，线程2查询缓存，发现未命中，就会直接去查询数据库，但是此时从数据库中查询的数据还是旧数据，然后线程2把这个旧数据重新写入到Redis缓存中了；此时回到线程1，线程1接着完成剩下的更新数据库的操作，即将数据库中的数据换成新值，此时就会存在问题：数据库中的数据是线程1写入到数据库中的新值；而Redis中缓存的数据是线程2从数据库中读取的旧值。

第二种方式：先更新数据库再删除缓存

线程1查询缓存，发现没命中，就直接去查询数据库，线程1读取的值是旧值；此时，线程2进入，将数据库中的数据更新为新值，然后直接删除缓存；此时又回到线程1，线程1将之前的旧值重新写入到缓存中，此时就会存在问题：数据库中的数据是线程2写入的新值，但是缓存中的数据却是线程1读取到的旧值。

其实，无论是先更新数据库再删除缓存还是先删除缓存再更新数据库，这两种方式在高并发的情况下都会存在问题；但是综合对比下来，我们会选择方式二：核心原因在于它能最大程度降低 “缓存与数据库数据不一致” 的风险，且异常场景下的问题影响更小。理由如下：

数据一致性风险更低：前者（先删缓存）会因并发请求直接写入旧数据到缓存，导致 “长期不一致”；后者（先更 DB）仅在 “更新后删缓存前崩溃” 时出现 “临时不一致”，且可通过缓存过期自愈。
异常影响范围更小：前者崩溃会导致缓存缺失 + 数据库旧数据，引发缓存穿透；后者崩溃仅导致缓存未删，旧缓存会随过期失效，不会压垮数据库。
适配分布式场景：在多服务、多数据库节点的分布式环境中，“先更 DB 再删缓存” 的逻辑更稳定 —— 数据库的事务特性（如 ACID）能保证更新的可靠性，而删除缓存作为 “最终一致性” 的补充，无需强事务支持。

但是在大多数情况下，我们有更好的选择，即延迟双删：

为彻底解决 “先更 DB 再删缓存” 中 “更新后崩溃未删缓存” 的极端情况，可引入延迟双删：

先更新数据库；
立即删除一次缓存；
延迟一段时间（如 500ms，根据业务耗时调整）后，再次删除缓存。

延迟第二次删除的目的是：若第一次删除缓存前线程崩溃，延迟删除可覆盖 “未删缓存” 的场景；若第一次删除成功，第二次删除属于 “空操作”，无额外成本。这进一步降低了不一致的概率，是生产环境的常用优化方案。

缓存穿透

缓存穿透：缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。

常见的解决方案有两种：

缓存空对象

- 优点：实现简单，维护方便
- 缺点：

- - 额外的内存消耗
  - 可能造成短期的不一致

布隆过滤

- 优点：内存占用较少，没有多余key
- 缺点：

- - 实现复杂
  - 存在误判可能

缓存空对象思路分析：当我们客户端访问不存在的数据时，先请求redis，但是此时redis中没有数据，此时会访问到数据库，但是数据库中也没有数据，这个数据穿透了缓存，直击数据库，我们都知道数据库能够承载的并发不如redis这么高，如果大量的请求同时过来访问这种不存在的数据，这些请求就都会访问到数据库，简单的解决方案就是哪怕这个数据在数据库中也不存在，我们也把这个数据存入到redis中去，这样，下次用户过来访问这个不存在的数据，那么在redis中也能找到这个数据就不会进入到缓存了
布隆过滤：布隆过滤器其实采用的是哈希思想来解决这个问题，通过一个庞大的二进制数组，走哈希思想去判断当前这个要查询的这个数据是否存在，如果布隆过滤器判断存在，则放行，这个请求会去访问redis，哪怕此时redis中的数据过期了，但是数据库中一定存在这个数据，在数据库中查询出来这个数据后，再将其放入到redis中，假设布隆过滤器判断这个数据不存在，则直接返回这种方式优点在于节约内存空间，存在误判，误判原因在于：布隆过滤器走的是哈希思想，只要哈希思想，就可能存在哈希冲突

总结：

缓存击穿

缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

常见的解决方案有两种：

互斥锁
逻辑过期

逻辑分析：假设线程1在查询缓存之后，本来应该去查询数据库，然后把这个数据重新加载到缓存的，此时只要线程1走完这个逻辑，其他线程就都能从缓存中加载这些数据了，但是假设在线程1没有走完的时候，后续的线程2，线程3，线程4同时过来访问当前这个方法，那么这些线程都不能从缓存中查询到数据，那么他们就会同一时刻来访问查询缓存，都没查到，接着同一时间去访问数据库，同时的去执行数据库代码，对数据库访问压力过大。

解决方案一、使用锁来解决：

因为锁能实现互斥性。假设线程过来，只能一个人一个人的来访问数据库，从而避免对于数据库访问压力过大，但这也会影响查询的性能，因为此时会让查询的性能从并行变成了串行，我们可以采用tryLock方法 + double check来解决这样的问题。假设现在线程1过来访问，他查询缓存没有命中，但是此时他获得到了锁的资源，那么线程1就会一个人去执行逻辑，假设现在线程2过来，线程2在执行过程中，并没有获得到锁，那么线程2就可以进行到休眠，直到线程1把锁释放后，线程2获得到锁，然后再来执行逻辑，此时就能够从缓存中拿到数据了。

解决方案二、逻辑过期方案

方案分析：我们之所以会出现这个缓存击穿问题，主要原因是在于我们对key设置了过期时间，假设我们不设置过期时间，其实就不会有缓存击穿的问题，但是不设置过期时间，这样数据不就一直占用我们内存了吗，我们可以采用逻辑过期方案。我们把过期时间设置在 redis的value中，注意：这个过期时间并不会直接作用于redis，而是我们后续通过逻辑去处理。假设线程1去查询缓存，然后从value中判断出来当前的数据已经过期了，此时线程1去获得互斥锁，那么其他线程会进行阻塞，获得了锁的线程他会开启一个线程去进行以前的重构数据的逻辑，直到新开的线程完成这个逻辑后，才释放锁，而线程1直接进行返回，假设现在线程3过来访问，由于线程线程2持有着锁，所以线程3无法获得锁，线程3也直接返回数据，只有等到新开的线程2把重建数据构建完后，其他线程才能走返回正确的数据。这种方案巧妙在于，异步的构建缓存，缺点在于在构建完缓存之前，返回的都是脏数据。

方案对比：

互斥锁方案：由于保证了互斥性，所以数据一致，且实现简单，因为仅仅只需要加一把锁而已，也没其他的事情需要操心，所以没有额外的内存消耗，缺点在于有锁就有死锁问题的发生，且只能串行执行性能肯定受到影响

逻辑过期方案：线程读取过程中不需要等待，性能好，有一个额外的线程持有锁去进行重构数据，但是在重构数据完成前，其他的线程只能返回之前的数据，且实现起来麻烦。

缓存雪崩

缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

解决方案：

给不同的Key的TTL添加随机值
利用Redis集群提高服务的可用性
给缓存业务添加降级限流策略
给业务添加多级缓存

Redis工具类

基于StringRedisTemplate封装一个缓存工具类，满足下列需求：

方法1：将任意Java对象序列化为json并存储在string类型的key中，并且可以设置TTL过期时间
方法2：将任意Java对象序列化为json并存储在string类型的key中，并且可以设置逻辑过期时间，用于处理缓

存击穿问题

方法3：根据指定的key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题
方法4：根据指定的key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题

将逻辑进行封装：

import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;import static com.hmdp.utils.RedisConstants.CACHE_NULL_TTL;
import static com.hmdp.utils.RedisConstants.LOCK_SHOP_KEY;@Slf4j
@Component
@AllArgsConstructor
//这是封装的Redis的缓存工具类，可以用来解决缓存穿透、缓存击穿问题
public class CacheClient {private final StringRedisTemplate stringRedisTemplate;//开启线程池private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);//方法1：将任意Java对象序列化为json并存储在string类型的key中，并且可以设置TTL过期时间public void set(String key, Object value, Long time, TimeUnit unit) {//对于value而言，用户可以传递任意类型，我们需要将value转为json格式，用JSONUtil工具类转为jsonstringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, unit);}//方法2：将任意Java对象序列化为json并存储在string类型的key中，并且可以设置逻辑过期时间，用于处理缓存击穿问题public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {// 设置逻辑过期RedisData redisData = new RedisData();redisData.setData(value);//因为不确定前端传递的time的单位，所以需要统一转换为秒redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));// 写入RedisstringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));}//方法3：根据指定的key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题public <R,ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit){String key = keyPrefix + id;// 1.从redis查询商铺缓存String json = stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isNotBlank(json)) {// 3.存在，直接返回return JSONUtil.toBean(json, type);}// 判断命中的是否是空值if (json != null) {// 返回一个错误信息return null;}// 4.不存在，根据id查询数据库R r = dbFallback.apply(id);// 5.不存在，返回错误if (r == null) {// 将空值写入redisstringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回错误信息return null;}// 6.存在，写入redisthis.set(key, r, time, unit);return r;}//方法4：根据指定的key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题public <R, ID> R queryWithLogicalExpire(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {String key = keyPrefix + id;// 1.从redis查询商铺缓存String json = stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isBlank(json)) {// 3.存在，直接返回return null;}// 4.命中，需要先把json反序列化为对象RedisData redisData = JSONUtil.toBean(json, RedisData.class);R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);LocalDateTime expireTime = redisData.getExpireTime();// 5.判断是否过期if(expireTime.isAfter(LocalDateTime.now())) {// 5.1.未过期，直接返回店铺信息return r;}// 5.2.已过期，需要缓存重建// 6.缓存重建// 6.1.获取互斥锁String lockKey = LOCK_SHOP_KEY + id;boolean isLock = tryLock(lockKey);// 6.2.判断是否获取锁成功if (isLock){// 6.3.成功，开启独立线程，实现缓存重建CACHE_REBUILD_EXECUTOR.submit(() -> {try {// 查询数据库R newR = dbFallback.apply(id);// 重建缓存this.setWithLogicalExpire(key, newR, time, unit);} catch (Exception e) {throw new RuntimeException(e);}finally {// 释放锁unlock(lockKey);}});}// 6.4.返回过期的商铺信息return r;}public <R, ID> R queryWithMutex(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {String key = keyPrefix + id;// 1.从redis查询商铺缓存String shopJson = stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isNotBlank(shopJson)) {// 3.存在，直接返回return JSONUtil.toBean(shopJson, type);}// 判断命中的是否是空值if (shopJson != null) {// 返回一个错误信息return null;}// 4.实现缓存重建// 4.1.获取互斥锁String lockKey = LOCK_SHOP_KEY + id;R r = null;try {boolean isLock = tryLock(lockKey);// 4.2.判断是否获取成功if (!isLock) {// 4.3.获取锁失败，休眠并重试Thread.sleep(50);return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);}// 4.4.获取锁成功，根据id查询数据库r = dbFallback.apply(id);// 5.不存在，返回错误if (r == null) {// 将空值写入redisstringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回错误信息return null;}// 6.存在，写入redisthis.set(key, r, time, unit);} catch (InterruptedException e) {throw new RuntimeException(e);}finally {// 7.释放锁unlock(lockKey);}// 8.返回return r;}private boolean tryLock(String key) {Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);}private void unlock(String key) {stringRedisTemplate.delete(key);}
}

Redis优惠卷秒杀

全局唯一ID

在我们订单生成的过程中，我们订单id应该要满足以下条件：第一，订单id不能具有明显的规则，容易泄露隐私；第二，当我们系统的数据量起来后，可能会需要分库分表，因为存储在不同的表里，可能会存在id一样的主键，这是不合理的。要解决这些问题，我们就要需要使用全局ID生成器，这是一种在分布式系统下用来生成全局唯一ID的工具。为了增加ID的安全性，我们可以不直接使用Redis自增的数值，而是拼接一些其它信息：

在代码中，我们就是将id放到Redis上面，因为Redis是单线程的，并且有唯一的自增方式，这就可以很好的解决分布式主键具有明显规律和可以生成全局唯一的id。

基于Redis的工具类如下：

@Component
public class RedisIdWorker {/*** 开始时间戳*/private static final long BEGIN_TIMESTAMP = 1640995200L;/*** 序列号的位数*/private static final int COUNT_BITS = 32;private StringRedisTemplate stringRedisTemplate;public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}public long nextId(String keyPrefix) {// 1.生成时间戳LocalDateTime now = LocalDateTime.now();long nowSecond = now.toEpochSecond(ZoneOffset.UTC);long timestamp = nowSecond - BEGIN_TIMESTAMP;// 2.生成序列号// 2.1.获取当前日期，精确到天String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));// 2.2.自增长long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);// 3.拼接并返回return timestamp << COUNT_BITS | count;}
}

库存超卖问题分析

对于买卖问题，我们需要考虑到多线程的环境下的并发问题：假设线程1过来查询库存，判断出来库存大于1，正准备去扣减库存，但是还没有来得及去扣减，此时线程2过来，线程2也去查询库存，发现这个数量一定也大于1，那么这两个线程都会去扣减库存，最终多个线程相当于一起去扣减库存，此时就会出现库存的超卖问题。

也就是说，在多线程的环境下，我们的买卖问题是会存在问题的，问题造成的结果就叫做超卖问题；

解决超卖问题的方案有很多，归纳一下可以分为一下两点：

悲观锁：

悲观锁可以实现对于数据的串行化执行，比如syn，和lock都是悲观锁的代表，同时，悲观锁中又可以再细分为公平锁，非公平锁，可重入锁，等等

乐观锁：

乐观锁：会有一个版本号，每次操作数据会对版本号+1，再提交回数据时，会去校验是否比之前的版本大1 ，如果大1 ，则进行操作成功，这套机制的核心逻辑在于，如果在操作过程中，版本号只比原来大1 ，那么就意味着操作过程中没有人对他进行过修改，他的操作就是安全的，如果不大1，则数据被修改过，当然乐观锁还有一些变种的处理方式比如cas。观锁的典型代表：就是cas，利用cas进行无锁化机制加锁，var5 是操作前读取的内存值，while中的var1+var2 是预估值，如果预估值 == 内存值，则代表中间没有被人修改过，此时就将新值去替换内存值

课程中的使用方式：

课程中的使用方式是没有像cas一样带自旋的操作，也没有对version的版本号+1 ，他的操作逻辑是在操作时，对版本号进行+1 操作，然后要求version 如果是1 的情况下，才能操作，那么第一个线程在操作后，数据库中的version变成了2，但是他自己满足version=1 ，所以没有问题，此时线程2执行，线程2 最后也需要加上条件version =1 ，但是现在由于线程1已经操作过了，所以线程2，操作时就不满足version=1 的条件了，所以线程2无法执行成功。

Redis分布式锁

基本原理和实现方式

原理分析：分布式锁也是Redis常用的一个场景之一，所谓分布式锁，是指满足分布式系统或集群模式下多进程可见并且互斥的锁。分布式锁的核心思想就是让大家都使用同一把锁，只要大家使用的是同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路。

JVM的锁依赖于锁监视器，在每个JVM内部都有一个锁监视器，这个监视器最多只能绑定一个线程，那么在分布式的项目中，因为在每个节点上都有属于自己的tomcat，即在每个节点上都存在自己的JVM对象，这就导致了锁监视器锁住的对象不是同一个，这也是为什么一般的锁对分布式系统无效果的原因；对于分布式系统，我们只需要人所有的节点都共享一个锁监视器就可以达到互斥的目的。

根据上面的分析，分布式锁应该满足以下条件

可见性：多个线程都能看到相同的结果，注意：这个地方说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思，所以所有的进程都需要建立联系；

互斥性：互斥是分布式锁的最基本的条件，使得程序串行执行了；

高可用：程序不易崩溃，时时刻刻都保证较高的可用性；

高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能；

安全性：安全也是程序中必不可少的一环。

常见的分布式锁有一下三种：

常见的分布式锁有三种

Mysql：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少见；

Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用redis或者zookeeper作为分布式锁，利用setnx这个方法，如果插入key成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁；

Zookeeper：zookeeper也是企业级开发中较好的一个实现分布式锁的方案。

其中，最常见的分布式锁的方案就是Redis了。

Redis分布式锁的实现核心思路

核心思路：实现分布式锁时需要实现的两个基本方法，分别是获取锁和释放锁。核心思路其实很简单，简单来说就是，我们利用redis 的setNx 方法，当有多个线程进入时，我们就利用该方法，第一个线程进入时，redis 中就有这个key 了，返回了1，如果结果是1，则表示他抢到了锁，那么他去执行业务，然后再删除锁，退出锁逻辑，没有抢到锁的哥们，等待一定时间后重试即可。

Redis分布式锁误删情况说明

逻辑说明：

持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明

解决方案：

解决方案就是在每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁。

要解决这种情况，我们可以在存入锁时，放入自己线程的标识（如，UUID），在删除锁时，判断当前这把锁的标识是不是自己存入的，如果是，则进行删除，如果不是，则不进行删除。

分布式锁的原子性问题

另一种极端的误删逻辑说明：

线程1现在持有锁之后，在执行业务逻辑过程中，他正准备删除锁，而且已经走到了条件判断的过程中，比如他已经拿到了当前这把锁确实是属于他自己的，正准备删除锁，但是此时他的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，我们要防止刚才的情况发生。

我们知道了造成这种情况的原因是因为拿锁，比锁，删锁这三个操作在代码中是三个步骤，并不是原子性的，那么有没有一种方法，让这三步合并为一步呢?如果可以，那这三个操作就是原子性的了。其实，我们一般会使用Lua脚本解决多条命令原子性问题。Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Redis提供的调用函数，我们可以使用lua去操作redis，又能保证他的原子性，这样就可以实现拿锁比锁删锁是一个原子性动作了。

接下来我们来回一下我们释放锁的逻辑：

释放锁的业务流程是这样的

1、获取锁中的线程标示

2、判断是否与指定的标示（当前线程标示）一致

3、如果一致则释放锁（删除）

4、如果不一致则什么都不做

小总结：

基于Redis的分布式锁实现思路：

利用set nx ex获取锁，并设置过期时间，保存线程标示
释放锁时先判断线程标示是否与自己一致，一致则删除锁

- 特性：

- - 利用set nx满足互斥性
  - 利用set ex保证故障时锁依然能释放，避免死锁，提高安全性
  - 利用Redis集群保证高可用和高并发特性

笔者总结：我们一路走来，利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题我们开始是利用删之前通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是我们没法保证拿锁比锁删锁是一个原子性的动作，最后通过lua表达式来解决这个问题。

Redis分布式锁redission

Redission的介绍

之前基于Redis实现分布式锁的方法只是能保证在大多数情况下有效，其实这种方法还存在以下不足：

重入问题：重入问题是指获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如HashTable这样的代码中，他的方法都是使用synchronized修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

不可重试：是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。

超时释放：我们在加锁时增加了过期时间，这样的我们可以防止死锁，但是如果卡顿的时间超长，虽然我们采用了lua表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患

主从一致性： 如果Redis提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

其实，对于基于Redis实现分布式锁的实现，Redis已经给我们封装好了框架，这个框架可以很好的解决上述问题，这个框架就是Redission。Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。Redission提供了分布式锁的多种多样的功能，是一种相对来说比较成熟的分布式锁解决方案：

分布式锁-redission可重入锁原理

在Lock锁中，他是借助于底层的一个voaltile的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state=0，假如有人持有这把锁，那么state=1，如果持有这把锁的人再次持有这把锁，那么state就会+1 ，如果是对于synchronized而言，他在c语言代码中会有一个count，原理和state类似，也是重入一次就加一，释放一次就-1 ，直到减少成0 时，表示当前这把锁没有被人持有。

在redission中，我们的也支持支持可重入锁，在分布式锁中，他采用hash结构用来存储锁，其中大key表示表示这把锁是否存在，用小key表示当前这把锁被哪个线程持有，所以接下来我们一起分析一下当前的这个lua表达式

这个地方一共有3个参数

KEYS[1] ：锁名称

ARGV[1]：锁失效时间

ARGV[2]： id + ":" + threadId; 锁的小key

exists: 判断数据是否存在 name：是lock是否存在,如果==0，就表示当前这把锁不存在；

redis.call('hset', KEYS[1], ARGV[2], 1);此时他就开始往redis里边去写数据，写成一个hash结构；

Lock{

id + ":" + threadId : 1

}

如果当前这把锁存在，则第一个条件不满足，再判断

redis.call('hexists', KEYS[1], ARGV[2]) == 1

此时需要通过大key+小key判断当前这把锁是否是属于自己的，如果是自己的，则进行

redis.call('hincrby', KEYS[1], ARGV[2], 1)

将当前这个锁的value进行+1 ，redis.call('pexpire', KEYS[1], ARGV[1]); 然后再对其设置过期时间，如果以上两个条件都不满足，则表示当前这把锁抢锁失败，最后返回pttl，即为当前这把锁的失效时间如果小伙帮们看了前边的源码，你会发现他会去判断当前这个方法的返回值是否为null，如果是null，则对应则前两个if对应的条件，退出抢锁逻辑，如果返回的不是null，即走了第三个分支，在源码处会进行while(true)的自旋抢锁。

Redission框架可重入锁原理总结：

Redisson 的可重入锁（RLock）基于 Redis 的 Hash 结构存储锁信息（key 为锁名，field 为线程唯一标识，value 为重入次数），通过 Lua 脚本保证原子性；当线程首次获取锁时，在 Redis 中创建 Hash 并设置过期时间；若同一线程再次获取锁，则将重入次数加 1 并重置过期时间；释放锁时，重入次数减 1，只有当次数为 0 时才真正删除锁键，从而实现分布式环境下的可重入特性，并避免死锁。

分布式锁-redission锁重试和WatchDog机制

分析：

Redisson 锁重试机制：Redisson 在获取分布式锁时，如果锁已被其他线程占用，并不会立即返回失败，而是会进入一个重试逻辑。它会通过循环尝试再次获取锁，期间会根据配置的等待时间（waitTime）和重试间隔（retryInterval）进行多次尝试。这种重试机制可以有效避免因瞬时锁竞争导致的获取失败，提高锁获取的成功率。当等待时间耗尽仍未获取到锁时，才会返回获取失败的结果。（单体中的synchronized锁重试原理一样）

Redisson WatchDog 机制：WatchDog 是 Redisson 提供的自动续期机制，用于防止在持有锁的线程执行任务期间锁因过期时间到了而被自动释放。当客户端获取锁成功后，如果没有指定锁的过期时间，Redisson 会默认启用 WatchDog 线程，每隔一定时间（默认 30 秒的 1/3，即 10 秒）自动延长锁的有效期。这样可以保证只要持有锁的线程还在运行，锁就不会被其他线程提前抢占，直到任务完成并显式释放锁。

Redis消息队列

认识消息队列

什么是消息队列：字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

消息队列：存储和管理消息，也被称为消息代理（Message Broker）
生产者：发送消息到消息队列
消费者：从消息队列获取消息并处理消息

使用队列的好处在于 **解耦：**所谓解耦，举一个生活中的例子就是：快递员(生产者)把快递放到快递柜里边(Message Queue)去，我们(消费者)从快递柜里边去拿东西，这就是一个异步，如果耦合，那么这个快递员相当于直接把快递交给你，这事固然好，但是万一你不在家，那么快递员就会一直等你，这就浪费了快递员的时间，所以这种思想在我们日常开发中，是非常有必要的。这种场景在我们秒杀中就变成了：我们下单之后，利用redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快我们的响应速度。

Redis的消息队列

我们的Redis的消息队列，只要有三种：基于List结构模拟消息队列；基于PubSub的点对点消息模型以及比较完善的消息队列模型，这三者的区别如下图所示：

如上列图所示，在Redis中，基于Stream的消息队列性能最好，也最安全，所以如果用Redis来作为消息队列来使用，我们一般都会选择Stream。这种方式的主要特征如下：

Redis达人探店

点赞功能

需求：

同一个用户只能点赞一次，再次点击则取消点赞
如果当前用户已经点赞，则点赞按钮高亮显示（前端实现，判断字段Blog类的isLike属性）

实现步骤：

给Blog类中添加一个isLike字段，标示是否被当前用户点赞
修改点赞功能，利用Redis的set集合判断是否点赞过，未点赞过则点赞数+1，已点赞过则点赞数-1
修改根据id查询Blog的业务，判断当前登录用户是否点赞过，赋值给isLike字段
修改分页查询Blog业务，判断当前登录用户是否点赞过，赋值给isLike字段

分析：

对于点赞功能而言，我们后端应该去记录它的点赞数以及存储点赞人的信息，MySQL也可以实现这一需求，但是频繁的点赞会对数据库造成读写压力，影响性能，所以对于点赞功能而言，我们一般会选择性能更好的Redis来实现这一逻辑。因为一个人一个账号最多只能点赞一次，所以Redis的set集合就非常适合这个场景，因为set集合具有去重的功能，可以保证最多只有一个人在已点赞的列表中。具体的逻辑是：当有用户点赞，我们就去Redis的set集合中判断当前key（由点赞的固定前缀+用户id生成唯一标识）在Redis的set集合中是否存在，如果不存在，表示当前用户未点赞，那么我们就在MySQL数据库中将当前文章的点赞数+1，然后在给当前用户生成唯一的key，再存放到Redis中，表示当前用户已经点赞；相反，如果已点赞的用户再次点赞，我们就需要走相反的逻辑，就是将数据库中的当前文章的点赞数量-1，再从Redis中将当前用户的key从set集合中移除即可。需要注意的是，我们以上思想是对Redis进行查找和判断，而对于依赖点赞数量的页面中，就比如主页分页查找和根据博客id查找博客详情中，都需要查询点赞数，我们需要对这两个接口进行改造，才能让点赞数和对应的博客一致。

点赞排行榜

在探店笔记的详情页面，应该把给该笔记点赞的人显示出来，比如最早点赞的TOP5，形成点赞排行榜；之前的点赞是放到set集合，但是set集合是不能排序的，无法实现排行榜的功能；所以这个时候，咱们可以采用一个可以排序的set集合，就是咱们的sortedSet。Redis中三种集合的对比如下：

List有序但是值不具有唯一性；Set具有唯一性但是无序；SortedSet有序又具有唯一性，而且SortedSet的顺序可以由我们自己制定，很适合用来作为排行榜来使用。在点赞排行榜中，我们采用的就是Redis中的SortedSet。

思路如下：

点赞功能重构：要实现点赞排行榜，那我们的点赞功能的实现需要由Set集合转为SortedSet集合，所以我们需要对点赞功能进行改造：思路不变，就是在实现存和取的过程中要使用的是SortedSet的方法来实现，首先判断当前登录的用户是否已经点赞，如果未点赞，则数据库的点赞数+1，同时需要将点赞的用户信息保存到Redis当中去，需要注意的是，我们使用的是SortedSet的方法，需要加上时间戳作为SortedSet的score值，因为我们排序的时候需要使用这个score属性；相反，如果当前用户已点赞，我们就要执行取消点赞的逻辑，即数据库的点赞数需要-1，同时需要将当前用户从Redis的SortedSet集合中移除；
点赞排行榜实现：假设我们要选出一个博客中点赞速度前5的人的名单，即top5用户，我们就需要根据key去Redis的SortedSet集合中查找下标为0-4的用户id名单，再解析用户id后去数据库中查找用户信息，然后返回给前端即可。需要注意的是，MySQL默认是按索引排序的，即id更小的排在id更大的前面，但是我们点赞的时候的顺序是随机的，即假设我们查出点赞速度前五的id为1,5,3,7,9;但是MySQL中查出的顺序为1，3，5，7，9；这就改变了用户点赞顺序了，就造成排序失效，我们的解决方案是：从Redis中查出用户id列表（假设为ids）之后，通过StrUtil.join(",", ids)方法将这个集合转换为字符串；然后再根据用户id去数据库中查找用户的时候，通过“.in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list()”来手动追加SQL的排序部分，以此来保证数据库查询和我们从Redis中查出来的顺序一致。这样，我们的点赞排行榜功能就实现了。

Redis好友关注

共同关注

所谓的共同关注，其实就是求两个人的关注列表的交集部分。在Redis当中，我们可以使用之前学习过的set集合，在set集合中，有交集并集补集的api，我们可以把两人的关注的人分别放入到一个set集合中，然后再通过api去查看这两个set集合中的交集数据，这个交集就是这两个人的共同关注了。要实现这一效果，我们首先需要在当用户关注了某个用户的时候，需要将这个用户的信息放到Set集合当中，方便后续进行共同关注，同时当取消关注的时候，也需要将用户从set集合中进行删除，这就需要我们对关注列表进行改造了；然后，在求两个set集合的共同关注的时候，我们首先得为这两个用户各自创建一个key，然后调用Redis的intersect（）方法，这个方法我们将这两个set集合的key传递过去后，它会给我们返回这两个key锁对应的set集合的交集部分，这个交集部分就是这两个人的共同关注，这返回的是用户的id，我们拿着这些id就可以查找出共同关注的人的全部信息了。

Redis附近商户

Redis中有一种叫做GEO数据结构，专门用来对地理位置信息进行操作，GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。在我们的项目中，当我们点击美食之后，会出现一系列的商家，商家中可以按照多种排序方式，我们此时关注的是距离，这个地方就需要使用到我们的GEO，向后台传入当前app收集的地址，以当前坐标作为圆心，同时绑定相同的店家类型type，以及分页信息，把这几个条件传入后台，后台查询出对应的数据再返回。我们要做的事情是：将数据库表中的数据导入到redis中去，redis中的GEO，GEO在redis中就一个menber和一个经纬度，我们把x和y轴传入到redis做的经纬度位置去，但我们不能把所有的数据都放入到menber中去，毕竟作为redis是一个内存级数据库，如果存海量数据，redis还是力不从心，所以我们在这个地方存储他的id即可。但是这个时候还有一个问题，就是在redis中并没有存储type，所以我们无法根据type来对数据进行筛选，所以我们可以按照商户类型做分组，类型相同的商户作为同一组，以typeId为key存入同一个GEO集合中即可。

总结：

这一逻辑的实现思路是：首先，从数据库中查询出所有店铺信息，然后使用 Java Stream API 按店铺类型 ID（typeId）进行分组，将同一类型的店铺数据放入一个列表中。接着，遍历每个类型分组，为每个类型构造一个 Redis GEO 结构的 key（由固定前缀 SHOP_GEO_KEY 与 typeId 拼接而成），再将该类型下所有店铺的 ID 和经纬度封装成 RedisGeoCommands.GeoLocation 对象集合，最后通过 Spring Data Redis 的批量添加方法一次性将这些地理位置信息写入 Redis 对应的 GEO key 中，从而实现按类型存储店铺地理位置数据，为后续基于 Redis GEO 命令的附近店铺搜索提供数据基础。在这里，因为Redis是基于内存的，不适合缓存太多数据，所以我们只缓存店铺类型，id以及位置的经纬度，key是店铺类型，value是由店铺的id和经纬度组成的对象集合；当我们去查找数据的时候，我们只需要根据店铺类型就可以查找出当前店铺类型下的所有数据，包含店铺id，店铺经纬度；我们根据店铺id又可以去数据库中查找出店铺详情；而GEO中的经纬度是用来做过滤的，具体来说就是在Redis中根据经纬度查找指定范围内的所有店铺的id，所以我们需要所有店铺的经纬度信息。

具体实现思路：

我们代码的实现思路是：首先判断是否传入经纬度参数，如果没有则直接通过数据库按类型分页查询店铺信息返回；如果传入了经纬度，则先计算分页的起止位置，然后以给定坐标为中心、5000米为半径，从Redis对应类型的GEO集合中搜索店铺，获取包含距离的结果并限制返回数量，接着跳过前from条记录，截取需要的分页数据，提取店铺ID和距离信息，再到数据库中批量查询这些ID对应的店铺详情，并通过ORDER BY FIELD保持Redis返回的距离排序，最后将距离信息设置到对应的店铺对象中返回，从而实现了基于地理位置的附近店铺分页查询功能，兼顾了查询效率和结果顺序。

Redis用户签到

实现签到功能

要实现签到功能，我们只需要用数据库去记录某个用户某天是否已签到即可，但是对于MySQL数据库而言，如果对于每个用户每天的签到都去记录的话，那么当用户以及天数达到一定的数量的时候，我们的MySQL的性能是会受影响的，很显然此时再用MySQL去记录这一签到行为就有点不太合适了。那么有没有更好的方式去记录用户的签到行为呢？我们的Redis就提供了很好的实现，在Redis中，有一种思想：用0和1标示业务状态，这种思路就称为位图（BitMap）。这样我们就用极小的空间，来实现了大量数据的表示，把每一个bit位对应当月的每一天，形成了映射关系。Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是 2^32个bit位。在具体的实现过程中，我们把年和月作为bitMap的key，然后保存到一个bitMap中，每次签到就到对应的位上把数字从0变成1，只要对应是1，就表明说明这一天已经签到了，反之则没有签到。

签到天数统计

问题1：什么叫做连续签到天数？

从最后一次签到开始向前统计，直到遇到第一次未签到为止，计算总的签到次数，就是连续签到天数。

Java逻辑代码：获得当前这个月的最后一次签到数据，定义一个计数器，然后不停的向前统计，直到获得第一个非0的数字即可，每得到一个非0的数字计数器+1，直到遍历完所有的数据，就可以获得当前月的签到总天数了。

问题2：如何得到本月到今天为止的所有签到数据？

假设今天是10号，那么我们就可以从当前月的第一天开始，获得到当前这一天的位数，是10号，那么就是10位，去拿这段时间的数据，就能拿到所有的数据了，那么这10天里边签到了多少次呢？统计有多少个1即可。

问题3：如何从后向前遍历每个bit位？

bitMap返回的数据是10进制，假如说返回一个数字8，那么我哪里知道到底哪些是0，哪些是1呢？我们只需要让得到的10进制数字和1做与运算就可以了，因为在与运算中，1只有遇见1 才是1，其他数字都是0 ，我们把签到结果和1进行与操作，每与一次，就把签到结果向右移动一位，依次内推，我们就能完成从当天开始逐个往前遍历的效果了。

实现思路总结：

实现思路是，首先获取当前登录用户的 ID 以及当前的日期时间。接着，按照用户 ID 和当前年月拼接出 Redis 中存储签到信息的键。然后，获取当前是本月的第几天，通过 Redis 的 `bitField` 方法获取本月截止到当天的签到记录（以十进制数字形式返回）。若没有签到结果或签到记录对应的数字为 0，直接返回签到次数 0。如果有有效的签到数字，就通过循环遍历该数字的每一个二进制位，从最后一位开始，判断每一位是否为 1（为 1 表示当天签到），若是则签到次数加 1，然后将数字右移一位，继续判断下一位，直到遇到为 0 的位（表示未签到）就结束循环，最后返回统计得到的签到次数。

Redis用户统计

首先我们搞懂两个概念：

UV：全称Unique Visitor，也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次，就比如说QQ，抖音的访客。

PV：全称Page View，也叫页面访问量或点击量，用户每访问网站的一个页面，记录1次PV，用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。类似于抖音的点赞量。

通常情况下，PV（页面访问量）的数量会大于 UV（独立访客量），所以衡量同一个网站的访问量，我们需要综合考虑很多因素，所以我们只是单纯的把这两个值作为一个参考值

UV统计在服务端做会比较麻烦，因为要判断该用户是否已经统计过了，需要将统计过的用户信息保存。但是如果每个访问的用户都保存到Redis中，数据量会非常恐怖，那怎么处理呢？

Hyperloglog(HLL)是从Loglog算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。Redis中的HLL是基于string结构实现的，单个HLL的内存永远小于16kb，内存占用低的令人发指！作为代价，其测量结果是概率性的，有小于0.81％的误差。不过对于UV统计来说，这完全可以忽略。

实现思路：

在点评软件中，统计一个账号或商品的访客量，可基于Redis的HyperLogLog数据结构来实现。首先明确UV（独立访客量）和PV（页面访问量）的概念，UV统计的是访问的自然人数量，1天内同一用户多次访问只记1次，PV则是用户每访问一个页面记录1次，通常PV数量大于UV 。直接在服务端统计UV较为麻烦，需保存已统计用户信息，若每个用户信息都存到Redis中数据量会很庞大。而HyperLogLog是从Loglog算法派生的概率算法，用于确定大集合基数，不需要存储集合中所有值，在Redis中基于string结构实现，单个HyperLogLog内存占用永远小于16kb，虽然测量结果具有概率性，存在小于0.81%的误差，但对于UV统计而言该误差可忽略不计，所以可以利用Redis的HyperLogLog数据结构来高效地统计点评软件中账号或商品的访客量（即UV ），减少内存占用且能满足精度要求。

在实际开发中，借助 Redis 的 HyperLogLog 实现点评软件里账号或商品的访客量（UV）统计时，首先为每个目标（账号或商品）按规则生成唯一 HyperLogLog 键，当有用户访问对应页面，就用客户端库将用户标识添加到该键，利用其自动去重特性记录访客；查询时，通过客户端库查询对应键的基数得到访客量。同时，可在键名中加入时间信息实现跨时间维度统计，还能为店铺、类别等维度创建 HyperLogLog 键来支持多维度统计，对于误差，在高要求场景下可定期全量精确统计并对比调整，以此高效且低内存占用地完成访客量统计。

查看全文

http://www.dtcms.com/a/446553.html