Redis 实战篇

发表于2024-03-18更新于2024-06-21字数统计30k阅读时长215分阅读次数

author：SmallBoat

内容概述

当下主流模型介绍：
手机或者app端发起请求，请求我们的nginx服务器，nginx基于七层模型走的是HTTP协议，可以实现基于Lua直接绕开tomcat访问redis，也可以作为静态资源服务器，轻松扛下上万并发， 负载均衡到下游tomcat服务器，打散流量。我们都知道一台4核8G的tomcat，在优化和处理简单业务的加持下，最多能处理1000左右的并发， 经过nginx的负载均衡分流后，利用集群支撑起整个项目，同时nginx在部署了前端项目后，更是可以做到动静分离，进一步降低tomcat服务的压力，这些功能都得靠nginx起作用，所以nginx是整个项目中重要的一环。
在tomcat支撑起并发流量后，我们如果让tomcat直接去访问Mysql，根据经验Mysql企业级服务器一般是16或32核心cpu，32或64G内存，就算是企业级mysql加上固态硬盘能够支撑的并发，大概就是4000~7000左右，上万并发， 瞬间就会让Mysql服务器的cpu，硬盘全部打满。所以我们在高并发场景下，会选择使用mysql集群，同时为了进一步降低Mysql的压力，同时增加访问的性能，我们也会加入Redis，同时使用Redis集群使得Redis对外提供更好的服务。
短信登录：

  • 代码实现登录相关内容：如实现短信验证码登录、登录校验拦截、解决Session共享问题等
  • 通过Redis共享Session实现短信登录的验证

商户查询缓存：

  • 常见的缓存更新策略
  • 分析并解决缓存穿透、缓存雪崩、缓存击穿等问题
  • 封装Redis工具类

优惠券秒杀（4-8章）：

  • 通过乐观锁、悲观锁解决优惠券下单业务的线程安全问题
  • Lua脚本实现原子操作
  • Redis实现分布式锁及分布式解决框架Redisson
  • Redis实现消息队列

达人探店：

  • 利用Set对点赞用户判重
  • 利用SortedSet对点赞用户进行排名

好友关注：

  • 利用Set集合快速求出共同关注（交集）
  • 利用Feed流实现消息推送

附近商户：

  • 利用GEO数据结构实现附近商户的查询

用户签到：

  • 利用BitMap实现用户签到功能及连续签到天数统计
  • 简单分析布隆过滤器解决缓存穿透问题思路

UV统计：

  • 利用HyperLogLog数据结构实现百万UV统计

短信登录

基于Session实现登录流程

! 1.png
实现发送验证码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public Result sendCode(String phone, HttpSession session) {
    // 1.校验手机号
    if (RegexUtils.isPhoneInvalid(phone)) {
        // 2.如果不符合，返回错误信息
        return Result.fail("手机号格式错误！");
    }
    // 3.符合，生成验证码
    String code = RandomUtil.randomNumbers(6);

    // 4.保存验证码到 session
    session.setAttribute("code", code);

    // 5.发送验证码（并没有真正发送，只是存到日志里）
    log.debug("发送短信验证码成功，验证码：{}", code);
    // 返回ok
    return Result.ok();
}

实现验证码登录与注册：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
    // 1.校验手机号
    String phone = loginForm.getPhone();
    if (RegexUtils.isPhoneInvalid(phone)) {
        // 2.如果不符合，返回错误信息
        return Result.fail("手机号格式错误！");
    }
    // 3.从session获取验证码并校验
    Object cacheCode = session.getAttribute("code");
    String code = loginForm.getCode();
    if (cacheCode == null || !cacheCode.toString().equals(code)) {
        // 不一致，报错
        return Result.fail("验证码错误");
    }

    // 4.一致，根据手机号查询用户 select * from tb_user where phone = ?
    User user = query().eq("phone", phone).one();

    // 5.判断用户是否存在
    if (user == null) {
        // 6.不存在，创建新用户并保存
        user = createUserWithPhone(phone);
    }

    // 7.保存用户信息到 session中
    session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));
    return Result.ok();
}

private User createUserWithPhone(String phone) {
    // 1.创建用户
    User user = new User();
    user.setPhone(phone);
    user.setNickName(USER_NICK_NAME_PREFIX + RandomUtil.randomString(10));
    // 2.保存用户
    save(user);
    return user;
}

实现登录拦截功能

前置知识：

  • tomcat运行原理： 每次用户请求过来时，会访问我们在tomcat注册的端口，任何程序想要运行，都需要有一个线程对当前端口号进行监听，tomcat也不例外，当监听线程知道用户想要和tomcat连接时，就会由监听线程创建socket连接，socket都是成对出现的，用户通过socket传递数据，当tomcat端的socket接受到数据后，此时监听线程会从tomcat的线程池中取出一个线程执行用户请求，在我们的服务部署到tomcat后，线程会找到用户想要访问的工程，然后用这个线程转发到工程中的controller，service，dao中，并且访问对应的DB，在用户执行完请求后，再统一返回，再找到tomcat端的socket，再将数据写回到用户端的socket，完成请求和响应。

2.png

  • 线程隔离： 每个用户都是去找tomcat线程池中的一个线程来完成工作，使用完成后再进行回收，并且每个请求都是独立的，所以在每个用户去访问我们的工程时，我们可以使用threadlocal来做到线程隔离，使得每个线程操作自己的一份数据。

基于以上情况，我们可以在线程转发到对应的controller之前对用户身份进行校验，检查Session中的user是否为null。
3.png
实现拦截器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class LoginInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 1.获取session
        HttpSession session = request.getSession();
        // 2.获取session中的用户
        Object user = session.getAttribute("user");
        // 3.判断用户是否存在
        if(user == null) {
              // 4.不存在，拦截，返回401状态码
              response.setStatus(401);
              return false;
        }
        // 5.存在，保存用户信息到Threadlocal
        UserHolder.saveUser((User)user);
        // 6.放行
        return true;
    }
}

将拦截器配置到MvcConfig使其生效：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        // 登录拦截器
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/shop/**",
                        "/voucher/**",
                        "/shop-type/**",
                        "/upload/**",
                        "/blog/hot",
                        "/user/code",
                        "/user/login"
                ).order(1);
        // token刷新的拦截器
        registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);
    }
}

集群下的Session共享问题

Session共享问题

在集群模式下，nginx可能会将同一个用户的多次请求分配到不同的tomcat进行处理，每个tomcat中都有一份属于自己的session，假设用户第一次访问第一台tomcat，并且把自己的信息存放到第一台服务器的session中，但是第二次这个用户访问到了第二台tomcat，那么在第二台服务器上，肯定没有第一台服务器存放的session，所以此时整个登录拦截功能就会出现问题。这就是集群下的Session共享问题。

session拷贝

早期的方案是session拷贝：每当任意一台服务器的session修改时，都会同步给其他的Tomcat服务器的session，这样就能实现session的共享了。
但是这种方案会存在两大问题：

  • 每台服务器中都有完整的一份session数据，服务器压力过大；
  • session拷贝数据时，可能会出现延迟。

于是，强大的Redis登场了。↓

Redis实现Session共享

设计key结构

首先我们要思考一下利用redis来存储数据，那么到底使用哪种结构呢？由于存入的数据比较简单，我们可以考虑使用String，但是会多占用一点空间。
4.png
如果使用哈希，则他的value中只会存储他数据本身。如果不是特别在意内存，其实使用String就可以了。
5.png
所以我们选择简单的String结构。关于key的处理，每个用户有自己的Session，但是redis的key是共享的，所以需要为每个用户生成一个token，按照“前缀 + token”的方式作为key存储用户的信息。

引入Redis后的登录业务流程

6.png
实现发送验证码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public Result sendCode(String phone, HttpSession session) {
    // 1.校验手机号
    if (RegexUtils.isPhoneInvalid(phone)) {
        // 2.如果不符合，返回错误信息
        return Result.fail("手机号格式错误！");
    }
    // 3.符合，生成验证码
    String code = RandomUtil.randomNumbers(6);

    // 4.保存验证码到 redis
    stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY + phone, code, LOGIN_CODE_TTL, TimeUnit.MINUTES);

    // 5.发送验证码
    log.debug("发送短信验证码成功，验证码：{}", code);
    // 返回ok
    return Result.ok();
}

实现验证码登录与注册：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
    // 1.校验手机号
    String phone = loginForm.getPhone();
    if (RegexUtils.isPhoneInvalid(phone)) {
        // 2.如果不符合，返回错误信息
        return Result.fail("手机号格式错误！");
    }
    // 3.从redis获取验证码并校验
    String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);
    String code = loginForm.getCode();
    if (cacheCode == null || !cacheCode.equals(code)) {
        // 不一致，报错
        return Result.fail("验证码错误");
    }

    // 4.一致，根据手机号查询用户 select * from tb_user where phone = ?
    User user = query().eq("phone", phone).one();

    // 5.判断用户是否存在
    if (user == null) {
        // 6.不存在，创建新用户并保存
        user = createUserWithPhone(phone);
    }

    // 7.保存用户信息到 redis中
    // 7.1.随机生成token，作为登录令牌
    String token = UUID.randomUUID().toString(true);
    // 7.2.将User对象转为HashMap存储
    UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
    Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
            CopyOptions.create()
                    .setIgnoreNullValue(true)
                    .setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
    // 7.3.存储
    String tokenKey = LOGIN_USER_KEY + token;
    stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);
    // 7.4.设置token有效期
    stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);

    // 8.返回token
    return Result.ok(token);
}

Redis实现Session的刷新

现在有这样一个场景： 用户已经登录，然后过了一阵访问了一个不需要拦截的路径。结果再访问需要拦截的路径时发现登录过期了。
问题分析： 由于用户访问的是不需要拦截的路径，因此在访问时，token不会被刷新，于是下一次访问时，发现token已经过期。
7.png
解决方案： 既然已登录用户在访问无需拦截的路径不会刷新token，那么我们就在拦截器前面再加一个拦截器，这个拦截器专门用来刷新token或者放行未登录用户。再到第二个拦截器时再判断是否需要登录才能访问。
8.png
实现拦截器1：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public class RefreshTokenInterceptor implements HandlerInterceptor {

    private StringRedisTemplate stringRedisTemplate;

    public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 1.获取请求头中的token
        String token = request.getHeader("authorization");
        if (StrUtil.isBlank(token)) {
            return true;
        }
        // 2.基于TOKEN获取redis中的用户
        String key  = LOGIN_USER_KEY + token;
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(key);
        // 3.判断用户是否存在
        if (userMap.isEmpty()) {
            return true;
        }
        // 5.将查询到的hash数据转为UserDTO
        UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
        // 6.存在，保存用户信息到 ThreadLocal
        UserHolder.saveUser(userDTO);
        // 7.刷新token有效期
        stringRedisTemplate.expire(key, LOGIN_USER_TTL, TimeUnit.MINUTES);
        // 8.放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 移除用户
        UserHolder.removeUser();
    }
}

简化拦截器2：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class LoginInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 1.判断是否需要拦截（ThreadLocal中是否有用户）
        if (UserHolder.getUser() == null) {
            // 没有，需要拦截，设置状态码
            response.setStatus(401);
            // 拦截
            return false;
        }
        // 有用户，则放行
        return true;
    }
}

商户查询缓存

什么是缓存

生活中，与缓存相似的就好比越野车的“减震器”，它可以防止越野车因硬着陆导致车体损伤。
在我们的项目开发中，我们也需要类似的功能来解决大量的数据猛冲进系统，导致其操作线程无法处理过量信息而导致瘫痪的问题。
并且在企业中，如果这种情况处理不好，会极大的降低用户的口碑，所以企业十分重视缓存技术。
缓存： 即数据的交换缓冲区，平时我们所说的缓存其实就是缓冲区中的数据，一般来讲是从数据库中获取，并存储在本地。
例如：

1
2
3
4
5
6
7
8
// 本地用于高并发
static final ConcurrentHashMap<K,V> map = new ConcurrentHashMap<>();

// 用于redis等缓存
static final Cache<K,V> USER_CACHE = CacheBuilder.newBuilder().build();

// 本地缓存
static final Map<K,V> map = new HashMap();

由于其被static修饰，所以随着类的加载而被加载到内存之中,作为本地缓存；由于其又被final修饰，所以其引用(map)和对象(new HashMap())之间的关系是固定的，不能改变，因此不用担心赋值(=)导致缓存失效。

当然，使用缓存也是有一定的成本的：
9.png

缓存的使用场景

实际开发中，会构筑多级缓存来使系统运行速度进一步提升，例如：本地缓存与redis中的缓存并发使用。
常见的缓存主要有一下几种：

  • 浏览器缓存： 主要是存在于浏览器端的缓存；
  • 应用层缓存： 可以分为tomcat本地缓存，比如之前提到的map，或者是使用redis作为缓存；
  • 数据库缓存： 在数据库中有一片空间是 buffer pool，增改查数据都会先加载到mysql的缓存中；
  • CPU缓存： 当代计算机最大的问题是cpu性能提升了，但内存读写速度没有跟上，所以为了适应当下的情况，增加了cpu的L1，L2，L3级的缓存。

10.png

添加商户缓存

我们在查询商户信息的时候，如果直接去数据库里查修，速度就会比较慢，所以需要在这里添加商户的缓存。
标准的操作方式就是查询数据库之前先查询缓存，如果缓存数据存在，则直接从缓存中返回，如果缓存数据不存在，再查询数据库，然后将数据存入redis。
11.png
代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
public Result queryById(Long id) {
    // 先从Redis中查，这里的常量值是固定的前缀 + 店铺id
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 如果不为空（查询到了），则转为Shop类型直接返回
    if (StrUtil.isNotBlank(shopJson)) {
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    // 否则去数据库中查
    Shop shop = getById(id);
    // 查不到返回一个错误信息或者返回空都可以，根据自己的需求来
    if (shop == null) {
        return Result.fail("店铺不存在！！");
    }
    // 查到了则转为json字符串
    String jsonStr = JSONUtil.toJsonStr(shop);
    // 并存入redis
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, jsonStr);
    // 最终把查询到的商户信息返回给前端
    return Result.ok(shop);
}

缓存更新策略

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

  • 内存淘汰： redis自动进行，当redis内存达到咱们设定的max-memery的时候，会自动触发淘汰机制，淘汰掉一些不重要的数据(可以自己设置策略方式)；
  • 超时剔除： 当我们给redis设置了过期时间ttl之后，redis会将超时的数据进行删除，方便咱们继续使用缓存；
  • 主动更新： 我们可以手动调用方法把缓存删掉，通常用于解决缓存和数据库不一致问题。

12.png

数据库与缓存数据不一致问题

由于我们的缓存的数据源来自于数据库，而数据库的数据是会发生变化的。因此，如果当数据库中数据发生变化，而缓存却没有同步，此时就会有一致性问题存在，其后果是：用户使用缓存中的过时数据,就会产生类似多线程数据安全问题，从而影响业务，产品口碑等。
当前几种较为常见的解决方案（主动更新的几种方式）：

  • Cache Aside Pattern（旁路缓存模式）：先更新数据库，然后再更新缓存（双写方案）；
  • Read/Write Through Pattern（读写穿透）：将缓存和数据库整合为一个服务，由服务来维护数据一致性；
  • Write Behind Caching Pattern（异步缓存写入）：调用者只操作缓存，由其他线程异步将缓存持久化到数据库。

缓存不一致采用的最终方案

各个方案的特点：

  • Cache Aside Pattern：需要写代码，较为复杂，但可以人为控制；
  • Read/Write Through Pattern：调用者无需关心一致性问题，维护这样一个服务是比较复杂的，开发成本高；
  • Write Behind Caching Pattern：可以将多次对数据库的写操作合并为一次，但是可能会丢失数据，维护任务较为复杂，并且一致性不能及时保证。

综上，最终方案使用方案一：Cache Aside Pattern！
13.png
采用方案一衍生的问题：

  • 更新缓存时是更新缓存还是删除缓存？
    • 更新缓存：每次更新数据库都更新缓存，无效写操作较多
    • 删除缓存： 更新数据库时让缓存失效，查询时再更新缓存 √
  • 如何保证缓存与数据库的操作的同时成功或失败？
    • 对于单体系统，将缓存与数据库操作放在一个事务
    • 对于分布式系统，利用TCC等分布式事务方案
  • 先操作缓存还是先操作数据库？
    • 先删除缓存，再操作数据库
    • 先操作数据库，再删除缓存 √

我们应当是先操作数据库，再删除缓存，原因在于，如果两个线程并发来访问时，假设线程1先来，他先把缓存删了，此时线程2过来，他查询缓存数据并不存在，此时他写入缓存，当他写入缓存后，线程1再执行更新动作时，实际上写入的就是旧的数据，新的数据被旧数据覆盖了。

遇到线程安全问题时的“先删除缓存，再操作数据库”（导致将旧数据又写回缓存中）：
14.png
遇到线程安全问题时的“先操作数据库，再删除缓存”：
16.png
所以最终的方案是：先更新数据库，再删除缓存。

实现商户缓存与数据库双写一致

根据上述分析，最终我们选择在查询缓存时，如果缓存中没有数据，则查询数据库，并设置过期时间作为超时剔除方案兜底。更新缓存时，先删除数据库再删除缓存。

添加缓存时设置过期时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Override
public Result queryById(Long id) {
    // 先从Redis中查，这里的常量值是固定的前缀 + 店铺id
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 如果不为空（查询到了），则转为Shop类型直接返回
    if (StrUtil.isNotBlank(shopJson)) {
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    // 否则去数据库中查
    Shop shop = getById(id);
    if (shop == null) {
        return Result.fail("店铺不存在！");
    }
    // 查到了则转为json字符串
    String jsonStr = JSONUtil.toJsonStr(shop);
    // 并存入redis，设置TTL
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, jsonStr,CACHE_SHOP_TTL, TimeUnit.MINUTES);
    // 最终把查询到的商户信息返回给前端
    return Result.ok(shop);
}

更新缓存时先修改数据库再删除缓存

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
@Transactional
public Result update(Shop shop) {
    // 判空
    if (shop.getId() == null) {
        return Result.fail("店铺id不能为空！！");
    }
    // 先修改数据库
    updateById(shop);
    // 再删除缓存
    stringRedisTemplate.delete(CACHE_SHOP_KEY + shop.getId());
    return Result.ok();
}

缓存穿透问题解决方案及实现

缓存穿透及解决方案

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

  • 缓存空对象：实现简单，维护方便。但是会有额外的内存消耗，可能会造成短期的数据不一致。
  • 布隆过滤：内存占用少，没有多余的key。但是实现较为复杂，存在误判可能。
  • 增强id的复杂度，避免被猜测id规律
  • 做好数据的基础格式校验
  • 加强用户权限校验
  • 做好热点参数的限流

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

编码解决缓存穿透

在原来的逻辑中，我们如果发现这个数据在mysql中不存在，直接就返回404，这样会存在缓存穿透问题。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
public Result queryById(Long id) {
    // 先从Redis中查，这里的常量值是固定的前缀 + 店铺id
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 如果不为空（查询到了），则转为Shop类型直接返回
    if (StrUtil.isNotBlank(shopJson)) {
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    // 否则去数据库中查
    Shop shop = getById(id);
    // 查不到返回一个错误信息或者返回空都可以，根据自己的需求来
    if (shop == null) {
        return Result.fail("店铺不存在！！");
    }
    // 查到了则转为json字符串
    String jsonStr = JSONUtil.toJsonStr(shop);
    // 并存入redis，设置TTL
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, jsonStr,CACHE_SHOP_TTL, TimeUnit.MINUTES);
    // 最终把查询到的商户信息返回给前端
    return Result.ok(shop);
}

优化逻辑： 如果这个数据不存在，我们不会返回404 ，还是会把这个数据写入到Redis中，并且将value设置为空，当再次发起查询时，我们如果发现命中之后，判断这个value是否是null，如果是null，则是之前写入的数据，证明是缓存穿透数据，如果不是，则直接返回数据。
18.png 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Override
public Result queryById(Long id) {
    // 先从Redis中查，这里的常量值是固定的前缀 + 店铺id
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 如果不为空（查询到了），则转为Shop类型直接返回
    if (StrUtil.isNotBlank(shopJson)) {
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    // 如果查询到的是空字符串，则说明是我们缓存的空数据
    if (shopjson != null) {
        return Result.fail("店铺不存在！！");
    }
    // 否则去数据库中查
    Shop shop = getById(id);
    // 查不到，则将空字符串写入Redis
    if (shop == null) {
        // 这里的常量值是2分钟
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
        return Result.fail("店铺不存在！");
    }
    // 查到了则转为json字符串
    String jsonStr = JSONUtil.toJsonStr(shop);
    // 并存入redis，设置TTL
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, jsonStr, CACHE_SHOP_TTL, TimeUnit.MINUTES);
    // 最终把查询到的商户信息返回给前端
    return Result.ok(shop);
}

缓存雪崩问题解决方案

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

  • 给不同的Key的TTL添加随机值（针对数据同时失效）；
  • 利用Redis集群提高服务的可用性（针对Redis宕机，使用一个或者多个哨兵(Sentinel)实例组成的系统，对redis节点进行监控，在主节点出现故障的情况下，能将从节点中的一个升级为主节点，进行故障转义，保证系统的可用性。）；
  • 给缓存业务添加降级限流策略（主动限流保全数据库）；
  • 给业务添加多级缓存（增加缓存高可用，浏览器访问静态资源时，优先读取浏览器本地缓存；访问非静态资源（ajax查询数据）时，访问服务端；请求到达Nginx后，优先读取Nginx本地缓存；如果Nginx本地缓存未命中，则去直接查询Redis（不经过Tomcat）；如果Redis查询未命中，则查询Tomcat；请求进入Tomcat后，优先查询JVM进程缓存；如果JVM进程缓存未命中，则查询数据库）。

缓存击穿问题解决方案及实现

缓存击穿问题分析

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

  • 互斥锁
  • 逻辑过期

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

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

21.png
方案二：逻辑过期分析：

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

22.png
对比互斥锁与逻辑删除：

  • 互斥锁方案： 由于保证了互斥性，所以数据一致，且实现简单，只是加了一把锁而已，也没有其他的事情需要操心，所以没有额外的内存消耗，缺点在于有锁的情况，就可能死锁，所以只能串行执行，性能会受到影响。
  • 逻辑过期方案： 线程读取过程中不需要等待，性能好，有一个额外的线程持有锁去进行重构缓存数据，但是在重构数据完成之前，其他线程只能返回脏数据，且实现起来比较麻烦。

23.png

利用互斥锁解决缓存击穿问题

核心思路： 相较于原来从缓存中查询不到数据后直接查询数据库而言，现在的方案是进行查询之后，如果从缓存没有查询到数据，则进行互斥锁的获取，获取互斥锁后，判断是否获得到了锁，如果没有获得到，则休眠，过一会再进行尝试，直到获取到锁为止，才能进行查询。
24.png
操作锁的代码：
利用redis的setnx方法来表示获取锁，该方法含义是redis中如果没有这个key，则插入成功，返回1，在stringRedisTemplate中返回true， 如果有这个key则插入失败，则返回0，在stringRedisTemplate返回false，我们可以通过true，或者是false，来表示是否有线程成功插入key，成功插入的key的线程我们认为他就是获得到锁的线程。

1
2
3
4
5
6
7
8
9
10
// 获取锁
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);
}

实际查询代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 public Shop queryWithMutex(Long id)  {
    String key = CACHE_SHOP_KEY + id;
    // 1、从redis中查询商铺缓存
    String shopJson = stringRedisTemplate.opsForValue().get("key");
    // 2、判断是否存在
    if (StrUtil.isNotBlank(shopJson)) {
        // 存在,直接返回
        return JSONUtil.toBean(shopJson, Shop.class);
    }
    // 判断命中的值是否是空值
    if (shopJson != null) {
        // 返回一个错误信息
        return null;
    }
    // 4.实现缓存重构
    // 4.1 获取互斥锁
    String lockKey = "lock:shop:" + id;
    Shop shop = null;
    try {
        boolean isLock = tryLock(lockKey);
        // 4.2 判断否获取成功
        if(!isLock) {
            // 4.3 失败，则休眠重试
            Thread.sleep(50);
            return queryWithMutex(id);
        }
        // 4.4 成功，根据id查询数据库
         shop = getById(id);
        // 5.不存在，返回错误
        if(shop == null) {
             // 将空值写入redis
            stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
            // 返回错误信息
            return null;
        }
        // 6.写入redis
        stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),CACHE_NULL_TTL,TimeUnit.MINUTES);

    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    finally {
        // 7.释放互斥锁
        unlock(lockKey);
    }
    return shop;
}

利用逻辑过期解决缓存击穿

核心思路： 当用户开始查询redis时，判断是否命中，如果没有命中则直接返回空数据，不查询数据库，而一旦命中后，将value取出，判断value中的过期时间是否满足，如果没有过期，则直接返回redis中的数据，如果过期，则在开启独立线程后直接返回之前的数据，独立线程去重构数据，重构完成后释放互斥锁。
25.png
实现步骤：

  • 封装数据：因为现在redis中存储的数据的value需要带上过期时间，此时要么你去修改（或者继承）原来的实体类，要么新建一个类包含原有的数据和过期时间。
1
2
3
4
5
6
// 这里我们选择新建一个实体类，包含原有数据(Object)和过期时间，这样对原有的代码没有侵入性
@Data
public class RedisData {
    private LocalDateTime expireTime;
    private Object data;
}
  • 利用逻辑过期实现解决缓存击穿
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// 这里需要声明一个线程池，因为下面我们需要新建一个线程来完成重构缓存
private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

@Override
public Shop queryWithLogicalExpire(Long id) {
    // 1. 从redis中查询商铺缓存
    String json = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    // 2. 如果未命中，则返回空
    if (StrUtil.isBlank(json)) {
        return null;
    }
    // 3. 命中，将json反序列化为对象
    RedisData redisData = JSONUtil.toBean(json, RedisData.class);
    // 3.1 将data转为Shop对象
    JSONObject shopJson = (JSONObject) redisData.getData();
    Shop shop = JSONUtil.toBean(shopJson, Shop.class);
    // 3.2 获取过期时间
    LocalDateTime expireTime = redisData.getExpireTime();
    // 4. 判断是否过期
    if (LocalDateTime.now().isBefore(time)) {
        // 5. 未过期，直接返回商铺信息
        return shop;
    }
    // 6. 过期，尝试获取互斥锁
    boolean flag = tryLock(LOCK_SHOP_KEY + id);
    // 7. 获取到了锁
    if (flag) {
        // 8. 开启独立线程
        CACHE_REBUILD_EXECUTOR.submit(() -> {
            try {
                this.saveShop2Redis(id, LOCK_SHOP_TTL);
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                unlock(LOCK_SHOP_KEY + id);
            }
        });
        // 9. 直接返回商铺信息
        return shop;
    }
    // 10. 未获取到锁，直接返回商铺信息
    return shop;
}

封装Redis工具类

工具类的各方法说明

本节将基于StringRedisTemplate封装一个缓存工具类，该工具类的方法如下：

  • 方法1： 将任意Java对象序列化为JSON，并存储到String类型的Key中，并可以设置TTL过期时间
1
2
3
public void set(String key, Object value, Long time, TimeUnit timeUnit) {
    stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, timeUnit);
}
  • 方法2： 将任意Java对象序列化为JSON，并存储在String类型的Key中，并可以设置逻辑过期时间（用于处理缓存击穿问题）
1
2
3
4
5
6
7
8
public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {
    // 设置逻辑过期
    RedisData redisData = new RedisData();
    redisData.setData(value);
    redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
    // 写入Redis
    stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
}
  • 方法3： 根据指定的Key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题

注意：
（1）通用方法查询的就不一定是shop了，所以这里使用泛型代替，同时id也需要使用泛型。
（2）key的前缀会随着业务不同而变更，所以需要作为参数传入
（3）如果通过传入的id在redis中查询不到是需要去数据库查询的，所以需要传入查询数据库的逻辑函数
（4）对于TTL的设置需要传入时间长度和时间单位

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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) {
        // 将空值写入redis
        stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
        // 返回错误信息
        return null;
    }
    // 6.存在，写入redis
    this.set(key, r, time, unit);
    return r;
}
  • 方法4： 根据指定的Key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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;
}
  • 方法5： 根据指定的Key查询缓存，并反序列化为指定类型，需要利用互斥锁解决缓存击穿问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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) {
            // 将空值写入redis
            stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
            // 返回错误信息
            return null;
        }
        // 6.存在，写入redis
        this.set(key, r, time, unit);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }finally {
        // 7.释放锁
        unlock(lockKey);
    }
    // 8.返回
    return r;
}

工具类完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@Slf4j
@Component
public class CacheClient {

    private final StringRedisTemplate stringRedisTemplate;

    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

    public CacheClient(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    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);
    }

    public void set(String key, Object value, Long time, TimeUnit timeUnit) {
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, timeUnit);
    }

    public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {
    // 设置逻辑过期
        RedisData redisData = new RedisData();
        redisData.setData(value);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
        // 写入Redis
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
    }

    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) {
            // 将空值写入redis
            stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
            // 返回错误信息
            return null;
        }
        // 6.存在，写入redis
        this.set(key, r, time, unit);
        return r;
    }

    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) {
                // 将空值写入redis
                stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
                // 返回错误信息
                return null;
            }
            // 6.存在，写入redis
            this.set(key, r, time, unit);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }finally {
            // 7.释放锁
            unlock(lockKey);
        }
        // 8.返回
        return r;
    }
}

优惠券秒杀

Redis实现全局唯一ID

全局唯一ID的意义

  • 需要对订单做唯一标识（id）
  • id 的规律性不能太明显
  • 不能受到表单数量的限制（可能会进行分库分表）

全局唯一生成器需满足的特征

  • 唯一性
  • 安全性
  • 递增性
  • 高可用
  • 高性能

全局唯一ID的实现

为了增加ID的安全性，我们可以不直接使用Redis自增的数值，而是拼接一些其他信息。
26.png 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Component
public class RedisIdWorker {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    // 设置起始时间，我这里设定的是2022.01.01 00:00:00
    public static final Long BEGIN_TIMESTAMP = 1640995200L;
    // 序列号长度
    public static final Long COUNT_BIT = 32L;

    public long nextId(String keyPrefix) {
        // 1. 生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long currentSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timeStamp = currentSecond - BEGIN_TIMESTAMP;
        // 2. 生成序列号
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        long count = stringRedisTemplate.opsForValue().increment("inc:"+keyPrefix+":"+date);
        // 3. 拼接并返回，简单位运算
        return timeStamp << COUNT_BIT | count;
    }
}

实现秒杀下单基本流程

首先，根据分析，秒杀的下单流程如下：
27.png
Controller：

1
2
3
4
5
6
7
8
9
10
@RestController
@RequestMapping("/voucher-order")
public class VoucherOrderController {
    @Autowired
    private IVoucherOrderService voucherOrderService;
    @PostMapping("/seckill/{id}")
    public Result seckillVoucher(@PathVariable("id") Long voucherId) {
        return voucherOrderService.seckillVoucher(voucherId);
    }
}

Service：

1
2
3
public interface IVoucherOrderService extends IService<VoucherOrder> {
    Result seckillVoucher(Long voucherId);
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@Autowired
private ISeckillVoucherService seckillVoucherService;

@Autowired
private RedisIdWorker redisIdWorker;
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }
    // 5. 扣减库存
    boolean success = seckillVoucherService.update()
        .setSql("stock = stock - 1")
        .eq("voucher_id",voucherId)
        .update();
    if (!success) {
        return Result.fail("库存不足");
    }
    // 6. 创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1 设置订单id
    long orderId = redisIdWorker.nextId("order");
    // 6.2 设置用户id
    Long id = UserHolder.getUser().getId();
    // 6.3 设置代金券id
    voucherOrder.setVoucherId(voucherId);
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(id);
    // 7. 将订单数据保存到表中
    save(voucherOrder);
    // 8. 返回订单id
    return Result.ok(orderId);
}

解决超卖问题

超卖问题分析

在完成秒杀基本下单流程后，我们发现，在测试的时候会产生线程安全问题：

  • 线程1查询，发现商品数量大于0（商品数量为1）
  • 线程2查询，发现商品数量大于0（商品数量为1）
  • 线程1创建订单，并扣减库存
  • 线程1创建订单，并扣减库存（此时库存已经为0，但仍然创建了订单，并扣减库存，导致超卖问题）

28.png

超卖问题解决思路

超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是加锁：而对于加锁，我们通常有两种解决方案：

  1. 悲观锁
  • 悲观锁认为线程安全问题一定会发生，因此在操作数据之前先获取锁，确保线程串行执行
  • 例如Synchronized、Lock等，都是悲观锁
  1. 乐观锁
  • 乐观锁认为线程安全问题不一定会发生，因此不加锁，只是在更新数据的时候再去判断有没有其他线程对数据进行了修改
    • 如果没有修改，则认为自己是安全的，自己才可以更新数据
    • 如果已经被其他线程修改，则说明发生了安全问题，此时可以重试或者异常

悲观锁： 悲观锁可以实现对于数据的串行化执行，比如syn，和lock都是悲观锁的代表，同时，悲观锁中又可以再细分为公平锁，非公平锁，可重入锁等等。
乐观锁： 乐观锁则是增加版本号字段，每次操作数据会对版本号+1，提交操作时会校验当前版本号是否与刚才查询的版本号一致，如果不一致则说明已经被修改过。

此处采用针对当前业务优化的乐观锁解决超卖问题。我们在实现的时候并不需要增加一个版本号字段，而是利用库存是否发生变化进行版本的判断（更进一步可以只是校验库存是否大于0）

代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }
    // 5. 扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId)
            .gt("stock", 0)
            .update();
    if (!success) {
        return Result.fail("库存不足");
    }
    // 6. 创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1 设置订单id
    long orderId = redisIdWorker.nextId("order");
    // 6.2 设置用户id
    Long id = UserHolder.getUser().getId();
    // 6.3 设置代金券id
    voucherOrder.setVoucherId(voucherId);
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(id);
    // 7. 将订单数据保存到表中
    save(voucherOrder);
    // 8. 返回订单id
    return Result.ok(orderId);
}

一人一单限制

一人一单问题分析

在优惠券场景当中，往往会有同一个用户对于同一种优惠券只能购买一单的需求。
逻辑分析：
29.png

初步代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }

    // 新增一人一单逻辑
    Long userId = UserHolder.getUser().getId();
    int count = query().eq("voucher_id", voucherId).eq("user_id", userId).count();
    if (count > 0) {
        return Result.fail("你已经抢过优惠券了哦");
    }

    // 5. 扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1")
            .eq("voucher_id", voucherId)
            .gt("stock", 0)
            .update();
    if (!success) {
        return Result.fail("库存不足");
    }
    // 6. 创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 6.1 设置订单id
    long orderId = redisIdWorker.nextId("order");
    // 6.2 设置用户id
    Long id = UserHolder.getUser().getId();
    // 6.3 设置代金券id
    voucherOrder.setVoucherId(voucherId);
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(id);
    // 7. 将订单数据保存到表中
    save(voucherOrder);
    // 8. 返回订单id
    return Result.ok(orderId);
}

初步代码问题分析与二次代码实现

与超卖问题类似，一人一单也存在着同样的线程安全问题（当一个用户同时使用多个设备秒杀优惠券时）：

  • 线程1查询，发现当前用户未购买过该优惠券
  • 线程2查询，发现当前用户未购买过该优惠券
  • 线程1创建订单，并扣减库存
  • 线程2创建订单，并扣减库存（导致线程安全问题）

于是乎，我们只能被迫进行加锁，但是乐观锁比较适合更新数据，而现在是插入数据，所以我们需要使用悲观锁操作。
首先我们将创建订单部分抽取出来，并加上synchronized 锁，同时加上事务@Transactional。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@Transactional
public synchronized Result createVoucherOrder(Long voucherId) {

	Long userId = UserHolder.getUser().getId();
         // 5.1.查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        // 5.2.判断是否存在
        if (count > 0) {
            // 用户已经购买过了
            return Result.fail("用户已经购买过一次！");
        }

        // 6.扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
                .update();
        if (!success) {
            // 扣减失败
            return Result.fail("库存不足！");
        }

        // 7.创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 7.1.订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 7.2.用户id
        voucherOrder.setUserId(userId);
        // 7.3.代金券id
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);

        // 7.返回订单id
        return Result.ok(orderId);
}

二次代码问题分析与三次代码实现

我们发现，二次代码中加锁的粒度太大了（在使用锁过程中，控制锁粒度 是一个非常重要的事情，因为如果锁的粒度太大，会导致每个线程进来都会锁住）。
在当前业务中，我们只需要对同一个用户加锁即可，于是有了版本3。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@Transactional
public  Result createVoucherOrder(Long voucherId) {
	Long userId = UserHolder.getUser().getId();
    // intern() 这个方法是从常量池中拿到数据，如果我们直接使用userId.toString() 他拿到的对象实际上是不同的对象，new 出来的对象，我们使用锁必须保证锁必须是同一把，所以我们需要使用intern()方法
	synchronized(userId.toString().intern()) {
         // 5.1.查询订单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        // 5.2.判断是否存在
        if (count > 0) {
            // 用户已经购买过了
            return Result.fail("用户已经购买过一次！");
        }

        // 6.扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
                .update();
        if (!success) {
            // 扣减失败
            return Result.fail("库存不足！");
        }

        // 7.创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 7.1.订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 7.2.用户id
        voucherOrder.setUserId(userId);
        // 7.3.代金券id
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);

        // 7.返回订单id
        return Result.ok(orderId);
    }
}

三次代码问题分析与最终版本代码实现

在第三个版本中依然存在两个问题：

  • 问题1：当前方法被spring的事务控制，如果你在方法内部加锁，可能会导致当前方法事务还没有提交，但是锁已经释放，导致并发问题再次出现
  • 问题2：但是如果将锁加在调用createVoucherOrder的地方（问题1的解决办法），我们的事务是无法正常生效的，因为我们调用这个函数其实是this.的方式调用的，想要事务生效，必须通过代理来进行调用。

于是，我们这里可以将锁加在调用createVoucherOrder的地方，同时利用AopContext.currentProxy()来获取当前对象的代理对象，然后再用代理对象调用方法

需要在接口中创建createVoucherOrder方法。
需要引入aspectjweaver依赖
在启动类上加上@EnableAspectJAutoProxy(exposeProxy = true)注解

最终代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }

    Long userId = UserHolder.getUser().getId();

    // 在调用处加锁，同时利用代理使事务生效
    synchronized (userId.toString().intern()) {
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return proxy.createVoucherOrder(voucherId);
    }
}

@Transactional
public  Result createVoucherOrder(Long voucherId) {
    // 5.1.查询订单
    int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    // 5.2.判断是否存在
    if (count > 0) {
        // 用户已经购买过了
        return Result.fail("用户已经购买过一次！");
    }

    // 6.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1") // set stock = stock - 1
            .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
            .update();
    if (!success) {
        // 扣减失败
        return Result.fail("库存不足！");
    }

    // 7.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    // 7.1.订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    // 7.2.用户id
    voucherOrder.setUserId(userId);
    // 7.3.代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);

    // 7.返回订单id
    return Result.ok(orderId);
}

当前有更优的方案： 即将用户的购买信息加到Redis中，用户下单时直接从Redis中查询，校验其下单数量。

集群环境下的并发问题

通过上面的一人一单限制，解决了在单台tomcat服务器上的并发问题，但是当我们对tomcat进行集群部署时，又会出现并发安全问题：

  • jvm1中的线程1获取锁成功
  • jvm1中的线程2获取锁失败（单台tomcat并发问题成功抗住）
  • jvm2中的线程1获取锁成功（此时出现了集群环境下的并发安全问题）
  • jvm2中的线程2获取锁失败

30.png
那么，上面这个棘手的并发安全问题该如何解决呢？
这时候，就该分布式锁登场了！

分布式锁

分布式锁的基本原理

分布式锁：满足分布式系统或集群模式下多线程可见并且可以互斥的锁。
分布式锁的核心思想就是让大家共用同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路。
31.png
分布式锁应当满足如下条件：

  • 可见性：多个线程都能看到相同的结果

注意：这里说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思

  • 互斥：互斥是分布式锁的最基本条件，使得程序串行执行
  • 高可用：程序不易崩溃，时时刻刻都保证较高的可用性
  • 高性能：由于加锁本身就让性能降低，所以对于分布式锁需要他较高的加锁性能和释放锁性能
  • 安全性：安全也是程序中必不可少的一环

常见的分布式锁的实现：

  • MySQL：MySQL本身就带有锁机制，但是由于MySQL的性能一般，所以采用分布式锁的情况下，使用MySQL作为分布式锁比较少见
  • Redis：Redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都是用Redis或者Zookeeper作为分布式锁，利用SETNX这个方法，如果插入Key成功，则表示获得到了锁，如果有人插入成功，那么其他人就会插入失败，无法获取到锁，利用这套逻辑完成互斥，从而实现分布式锁
  • Zookeeper：其实Zookeeper也是企业级开发中较好的一种实现分布式锁的方案，本文主要讲解Redis的实现

32.png

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

实现分布式锁时需要实现两个基本方法

  • 获取锁
    • 互斥：确保只能有一个线程获取锁
    • 非阻塞：尝试一次，成功返回true，失败返回false
  • 释放锁
    • 手动释放
    • 超时释放：获取锁的时候添加一个超时时间（作为兜底）

核心思路：我们利用redis的SETNX方法，当有多个线程进入时，我们就利用该方法来获取锁。第一个线程进入时，redis 中就有这个key了，返回了1，如果结果是1，则表示他抢到了锁，那么他去执行业务，然后再删除锁，退出锁逻辑。没有抢到锁（返回了0）的线程，等待一定时间后重试即可（对于当前业务，说明用户的当前操作不合法，直接返回失败即可）。
33.png

实现分布式锁

创建锁的基本接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public interface ILock {
    /**
     * 尝试获取锁
     *
     * @param timeoutSec 锁持有的超时时间，过期自动释放
     * @return true表示获取锁成功，false表示获取锁失败
     */
    boolean tryLock(long timeoutSec);

    /**
     * 释放锁
     */
    void unlock();
}

实现锁的基本接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class SimpleRedisLock implements ILock {
    // 锁的前缀
    private static final String KEY_PREFIX = "lock:";
    // 具体业务名称，将前缀和业务名拼接之后当做Key
    private String name;
    // 这里不是@Autowired注入，采用的是构造器注入，在创建SimpleRedisLock时，将RedisTemplate作为参数传入
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标识
        long threadId = Thread.currentThread().getId();
        // 获取锁，使用SETNX方法进行加锁，同时设置过期时间，防止死锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
        // 自动拆箱可能会导致NPE问题，这样写更稳妥
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unlock() {
        // 通过DEL来删除锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

基于分布式锁修改秒杀业务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }
    Long userId = UserHolder.getUser().getId();
    // 创建锁对象
    SimpleRedisLock redisLock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
    // 获取锁对象
    boolean isLock = redisLock.tryLock(30);
    // 加锁失败，说明当前用户开了多个线程抢优惠券，但是由于key是SETNX的，所以不能创建key，得等key的TTL到期或释放锁（删除key）
    if (!isLock) {
        return Result.fail("你已经抢过一张啦，留给别人吧！");
    }
    try {
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return proxy.createVoucherOrder(voucherId);
    } finally {
        // 释放锁
        redisLock.unlock();
    }
}

Redis分布式锁错误释放问题

Redis分布式锁的错误释放问题

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

Redis分布式锁的错误释放问题解决方案

解决方案就是在每个线程释放锁的时候，都判断一下这个锁是不是自己的，如果不属于自己，则不进行删除操作。
35.png

解决后重新模拟错误释放问题

假设还是上面的情况，线程1阻塞，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1阻塞完了，继续往下执行，开始删除锁，但是线程1发现这把锁不是自己的，所以不进行删除锁的逻辑，当线程2执行到删除锁的逻辑时，如果TTL还未到期，则判断当前这把锁是自己的，于是删除这把锁。
36.png

解决Redis分布式锁误删问题代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 加锁
private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
@Override
public boolean tryLock(long timeoutSec) {
   // 获取线程标示
   String threadId = ID_PREFIX + Thread.currentThread().getId();
   // 获取锁
   Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
   return Boolean.TRUE.equals(success);
}

// 释放锁
public void unlock() {
    // 获取线程标示
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    // 获取锁中的标示
    String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    // 判断标示是否一致
    if(threadId.equals(id)) {
        // 释放锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

分布式锁的原子性问题

什么是分布式锁的原子性问题

所谓分布式锁的原子性问题，就是在上面的误删问题的基础上，即使判断了是自己的线程标识，但是由于判断和释放锁的这两个步骤并不具有原子性，在一些极端情况，可能会在判断成功后，阻塞住了，然后被超时释放，结果这个锁被其他的线程获取，这时阻塞住的线程恢复后直接进行了锁的释放，又导致了误删问题。

利用Lua脚本解决分布式锁的原子性问题

Lua脚本：

  • Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性
  • Lua是一种编程语言，其基本语法可在菜鸟教程查看
  • 这里重点介绍Redis提供的调用函数，我们可以使用Lua去操作Redis，而且还能保证它的原子性，这样就可以实现拿锁，判断标识，删锁是一个原子性动作了

Redis调用Lua脚本：

  • Lua脚本格式
1
redis.call('命令名称', 'key', '其它参数', ...)
  • 示例
1
2
3
4
5
6
-- 先执行 set name jack
redis.call('set', 'name', 'Rose')
-- 再执行 get name
local name = redis.call('get', 'name')
-- 返回
return name
  • Redis调用Lua脚本

37.png
Lua脚本实现判断标识与释放锁：

1
2
3
4
5
6
7
8
-- 这里的KEYS[1]就是传入锁的key
-- 这里的ARGV[1]就是线程标识
-- 比较锁中的线程标识与线程标识是否一致
if (redis.call('get', KEYS[1]) == ARGV[1]) then
    -- 一致则释放锁
    return redis.call('del', KEYS[1])
end
return 0

利用Java调用Lua脚本解决分布式锁的原子性问题

在RedisTemplate中，可以利用excute方法执行Lua脚本

1
2
3
public <T> T execute(RedisScript<T> script, List<K> keys, Object... args) {
    return this.scriptExecutor.execute(script, keys, args);
}

代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

static {
    UNLOCK_SCRIPT = new DefaultRedisScript();
    UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    UNLOCK_SCRIPT.setResultType(Long.class);
}

@Override
public void unlock() {
    stringRedisTemplate.execute(UNLOCK_SCRIPT,
            Collections.singletonList(KEY_PREFIX + name),
            ID_PREFIX + Thread.currentThread().getId());
}

基于Redis实现分布式锁的不足

基于Redis中的setnx命令实现分布式锁具有以下几个问题：
38.png
如果没有以上几种需求，基于Redis实现的分布式锁是完全没有问题的，但是，如果业务中的确有以上需求，那么该怎么解决呢？
于是，强大的Redisson诞生了！
在以后需要使用分布式锁的时候，直接使用Redisson的分布式解决方案即可。

分布式锁-Redisson

Redisson基础

Redisson是一个在Redis的基础上实现的Java驻内存数据网格(In-Memory Data Grid)。它不仅提供了一系列的分布式Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。
Redisson实现了分布式锁的各种功能：

  • 可重入锁(Reentrant Lock)
  • 公平锁(Fair Lock)
  • 联锁(MultiLock)
  • 红锁(RedLock)
  • 读写锁(ReadWriteLock)
  • 信号量(Semaphore)
  • 可过期性信号量(PermitExpirableSemaphore)
  • 闭锁(CountDownLatch)

项目引入Redisson

  • 引入依赖
1
2
3
4
5
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>
  • 配置RedissonConfig类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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 {
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
            .setAddress("redis:// 127.0.0.1:6379")
            .setPassword("123456");
        return Redisson.create(config);
    }
}
  • 使用Redisson的分布式锁修改秒杀业务
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@Resource
private RedissonClient redissonClient;
@Override
public Result seckillVoucher(Long voucherId) {
    LambdaQueryWrapper<SeckillVoucher> queryWrapper = new LambdaQueryWrapper<>();
    // 1. 查询优惠券
    queryWrapper.eq(SeckillVoucher::getVoucherId, voucherId);
    SeckillVoucher seckillVoucher = seckillVoucherService.getOne(queryWrapper);
    // 2. 判断秒杀时间是否开始
    if (LocalDateTime.now().isBefore(seckillVoucher.getBeginTime())) {
        return Result.fail("秒杀还未开始，请耐心等待");
    }
    // 3. 判断秒杀时间是否结束
    if (LocalDateTime.now().isAfter(seckillVoucher.getEndTime())) {
        return Result.fail("秒杀已经结束！");
    }
    // 4. 判断库存是否充足
    if (seckillVoucher.getStock() < 1) {
        return Result.fail("优惠券已被抢光了哦，下次记得手速快点");
    }
    Long userId = UserHolder.getUser().getId();
    RLock redisLock = redissonClient.getLock("order:" + userId);
    boolean isLock = redisLock.tryLock();
    if (!isLock) {
        return Result.fail("不允许抢多张优惠券");
    }
    try {
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return proxy.createVoucherOrder(voucherId);
    } finally {
        redisLock.unlock();
    }
}

Redisson实现可重入锁原理

参考JDK实现锁的原理，Redisson实现可重入也是基于计数机制，即有人获取当前锁时，count = 1，再次获取该锁时，count ++，当释放该锁时，count --，直到count == 0时表示当前锁未被持有。

能再次获取锁的必要条件是处于同一线程。

Redisson实现计数机制使用的是Redis的Hash结构，用外层的“大”key表示当前锁是否被持有，用Hash结构的“小”Key来记录被哪个线程持有。
39.png
尝试自己实现获取锁和释放锁：

由于需要保证原子性，所以也需要使用Lua实现。

  • 获取锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 锁不存在
if (redis.call('exists', key) == 0) then
    -- 获取锁并添加线程标识，state设为1
    redis.call('hset', key, threadId, '1');
    -- 设置锁有效期
    redis.call('expire', key, releaseTime);
    return 1; -- 返回结果
end;
-- 锁存在，判断threadId是否为自己
if (redis.call('hexists', key, threadId) == 1) then
    -- 锁存在，重入次数 +1，这里用的是hash结构的incrby增长
    redis.call('hincrby', key, thread, 1);
    -- 设置锁的有效期
    redis.call('expire', key, releaseTime);
    return 1; -- 返回结果
end;
return 0; -- 代码走到这里，说明获取锁的不是自己，获取锁失败
  • 释放锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
local key = KEYS[1];
local threadId = ARGV[1];
local releaseTime = ARGV[2];
-- 如果锁不是自己的
if (redis.call('HEXISTS', key, threadId) == 0) then
    return nil; -- 直接返回
end;
-- 锁是自己的，锁计数-1，还是用hincrby，不过自增长的值为-1
local count = redis.call('hincrby', key, threadId, -1);
-- 判断重入次数为多少
if (count > 0) then
    -- 大于0，重置有效期
    redis.call('expire', key, releaseTime);
    return nil;
else
    -- 否则直接释放锁
    redis.call('del', key);
    return nil;
end;

查看Redisson实现获取锁与释放锁的源码：

  • 获取锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    this.internalLockLeaseTime = unit.toMillis(leaseTime);
    return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command, 
                                "if (redis.call('exists', KEYS[1]) == 0) then " +
                                    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                    "return nil; " +
                                "end; " +
                                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                                    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                    "return nil; " +
                                 "end; " +
                                "return redis.call('pttl', KEYS[1]);", 
                                Collections.singletonList(this.getName()), 
                                this.internalLockLeaseTime, 
                                this.getLockName(threadId));
}
  • 释放锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, 
                                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                                    "return nil;" +
                                "end; " +
                                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                                "if (counter > 0) then " +
                                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                                    "return 0; " +
                                "else " +
                                    "redis.call('del', KEYS[1]); " +
                                    "redis.call('publish', KEYS[2], ARGV[1]); return 1; " +
                                "end; " +
                                "return nil;", 
                                Arrays.asList(this.getName(), 
                                this.getChannelName()), 
                                LockPubSub.UNLOCK_MESSAGE, 
                                this.internalLockLeaseTime, 
                                this.getLockName(threadId));
}

我们发现，源码实现和我们的预期几乎一样。

Redisson锁重试与WatchDog机制

通过对源码的解读，我们发现Redisson锁重试的实现如下图所示：
40.png

Redisson的MultiLock锁实现原理

原理分析

为了提高redis的可用性，我们会搭建集群或者主从，现在以主从为例。
此时我们去写命令，写在主机上， 主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。
41.png
为了解决这个问题，Redisson提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁加锁的逻辑需要写入到每一个主从节点上，只有所有的服务器都写入成功，此时才是加锁成功。
假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。
42.png
Redisson的MultiLock锁源码分析图：
43.png

实践

  • 使用虚拟机额外搭建两个Redis节点，然后重新配置RedissonConfig类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis:// 192.168.137.130:6379")
                .setPassword("root");
        return Redisson.create(config);
    }

    @Bean
    public RedissonClient redissonClient2() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis:// 92.168.137.131:6379")
                .setPassword("root");
        return Redisson.create(config);
    }

    @Bean
    public RedissonClient redissonClient3() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis:// 92.168.137.132:6379")
                .setPassword("root");
        return Redisson.create(config);
    }
}
  • 要使用联锁，我们需要注入三个RedissonClient对象
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@Resource
private RedissonClient redissonClient;
@Resource
private RedissonClient redissonClient2;
@Resource
private RedissonClient redissonClient3;

private RLock lock;

@BeforeEach
void setUp() {
    RLock lock1 = redissonClient.getLock("lock");
    RLock lock2 = redissonClient2.getLock("lock");
    RLock lock3 = redissonClient3.getLock("lock");
    lock = redissonClient.getMultiLock(lock1, lock2, lock3);
}

@Test
void method1() {
    boolean success = lock.tryLock();
    redissonClient.getMultiLock();
    if (!success) {
        log.error("获取锁失败，1");
        return;
    }
    try {
        log.info("获取锁成功");
        method2();
    } finally {
        log.info("释放锁，1");
        lock.unlock();
    }
}

void method2() {
    RLock lock = redissonClient.getLock("lock");
    boolean success = lock.tryLock();
    if (!success) {
        log.error("获取锁失败，2");
        return;
    }
    try {
        log.info("获取锁成功，2");
    } finally {
        log.info("释放锁，2");
        lock.unlock();
    }
}

Redisson分布式锁小结

  • 不可重入Redis分布式锁
    • 原理：利用SETNX的互斥性；利用EX避免死锁；释放锁时判断线程标识
    • 缺陷：不可重入、无法重试、锁超时失效
  • 可重入Redis分布式锁
    • 原理：利用Hash结构，记录线程标识与重入次数；利用WatchDog延续锁时间；利用信号量控制锁重试等待
    • 缺陷：Redis宕机引起锁失效问题
  • Redisson的multiLock
    • 原理：多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功
    • 缺陷：运维成本高、实现较为复杂

秒杀优化

异步秒杀思路

我们先来回顾一下秒杀下单流程：
当用户发起请求，此时会先请求Nginx，Nginx反向代理到Tomcat，而Tomcat中的程序，会进行串行操作，分为如下几个步骤：

  • 查询优惠券
  • 判断秒杀库存是否足够
  • 查询订单
  • 校验是否一人一单
  • 扣减库存
  • 创建订单

44.png
在这六个步骤中，有很多操作都是要去操作数据库的，而且还是一个线程串行执行，这样就会导致我们的程序执行很慢，所以我们需要异步程序执行，那么如何加速呢？
优化方案： 我们将耗时较短的逻辑判断放到Redis中，例如：库存是否充足，是否一人一单这样的操作，只要满足这两条操作，那我们是一定可以下单成功的，不用等数据真的写进数据库，我们直接告诉用户下单成功就好了。然后后台再开一个线程，后台线程再去慢慢执行队列里的消息，这样我们就能很快的完成下单业务。
但是这里存在两个难点：

  • 我们怎么在Redis中快速校验是否一人一单，还有库存判断
  • 我们校验一人一单和将下单数据写入数据库，这是两个线程，我们怎么知道下单是否完成

解决难点： 为了完成这件事我们在redis操作完（执行Lua脚本以保证原子性）之后，我们会将一些信息返回给前端，同时也会把这些信息丢到异步queue中去，后续操作中，可以通过这个id来查询我们tomcat中的下单逻辑是否完成了。
45.png
我们现在来看整体思路： 当用户下单之后，判断库存是否充足，只需要取Redis中根据key找对应的value是否大于0即可，如果不充足，则直接结束。如果充足，则在Redis中判断用户是否可以下单，如果set集合中没有该用户的下单数据，则可以下单，并将userId和优惠券存入到Redis中，并且返回0，整个过程需要保证是原子性的，所以我们要用Lua来操作。同时，由于我们需要在Redis中查询优惠券信息，所以在我们新增秒杀优惠券的同时，需要将优惠券信息保存到Redis中。
完成以上逻辑判断时，我们只需要判断当前Redis中的返回值是否为0，如果是0，则表示可以下单，将信息保存到queue中去，然后返回，开一个线程来异步下单，前端可以通过返回订单的id来查询是否下单成功
46.png

Redis完成用户秒杀资格校验

本节需求：

  • 新增秒杀优惠券的同时，将优惠券信息保存到Redis中
  • 基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功

保存优惠券代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {
    // 保存优惠券
    save(voucher);
    // 保存秒杀信息
    SeckillVoucher seckillVoucher = new SeckillVoucher();
    seckillVoucher.setVoucherId(voucher.getId());
    seckillVoucher.setStock(voucher.getStock());
    seckillVoucher.setBeginTime(voucher.getBeginTime());
    seckillVoucher.setEndTime(voucher.getEndTime());
    seckillVoucherService.save(seckillVoucher);
    // 保存秒杀优惠券信息到Reids，Key名中包含优惠券ID，Value为优惠券的剩余数量
    stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString()); 
}

编写Lua脚本：

Lua的字符串拼接使用 …，字符串转数字是 tonumber()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
-- 订单id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]
-- 优惠券key
local stockKey = 'seckill:stock:' .. voucherId
-- 订单key
local orderKey = 'seckill:order:' .. voucherId
-- 判断库存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0) then
    return 1
end
-- 判断用户是否下单
if (redis.call('sismember', orderKey, userId) == 1) then
    return 2
end
-- 扣减库存
redis.call('incrby', stockKey, -1)
-- 将userId存入当前优惠券的set集合
redis.call('sadd', orderKey, userId)
return 0

修改秒杀下单逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override
public Result seckillVoucher(Long voucherId) {
    // 1. 执行lua脚本
    Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
            Collections.emptyList(), voucherId.toString(),
            UserHolder.getUser().getId().toString());
    // 2. 判断返回值，并返回错误信息
    if (result.intValue() != 0) {
        return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");
    }
    long orderId = redisIdWorker.nextId("order");
    // TODO 保存阻塞队列

    // 3. 返回订单id
    return Result.ok(orderId);
}

基于阻塞队列实现秒杀优化

在上一小节中，我们在判断用户具有秒杀资格后还未创建订单并放进异步队列实现异步下单。
本节任务：

  • 创建订单并放进异步队列
  • 开启线程任务对异步队列中的订单进行消费

创建异步队列

创建阻塞队列：

1
private final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

封装优惠券订单并放进异步队列：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Override
public Result seckillVoucher(Long voucherId) {
    Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
            Collections.emptyList(), voucherId.toString(),
            UserHolder.getUser().getId().toString());
    if (result.intValue() != 0) {
        return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");
    }
    long orderId = redisIdWorker.nextId("order");
    // 封装到voucherOrder中
    VoucherOrder voucherOrder = new VoucherOrder();
    voucherOrder.setVoucherId(voucherId);
    voucherOrder.setUserId(UserHolder.getUser().getId());
    voucherOrder.setId(orderId);
    // 加入到阻塞队列
    orderTasks.add(voucherOrder);
    return Result.ok(orderId);
}

实现异步下单

创建线程池：

1
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

创建线程任务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 秒杀业务需要在类初始化之后，就立即执行，所以这里需要用到@PostConstruct注解
@PostConstruct
private void init() {
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}

private class VoucherOrderHandler implements Runnable {
    @Override
    public void run() {
        while (true) {
            try {
                // 1. 获取队列中的订单信息
                VoucherOrder voucherOrder = orderTasks.take();
                // 2. 创建订单
                handleVoucherOrder(voucherOrder);
            } catch (Exception e) {
                log.error("订单处理异常", e);
            }
        }
    }
}

创建订单业务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private IVoucherOrderService proxy;
private void handleVoucherOrder(VoucherOrder voucherOrder) {
    // 1. 获取用户
    Long userId = voucherOrder.getUserId();
    // 2. 创建锁对象，作为兜底方案
    RLock redisLock = redissonClient.getLock("order:" + userId);
    // 3. 获取锁
    boolean isLock = redisLock.tryLock();
    // 4. 判断是否获取锁成功         
    if (!isLock) {
        log.error("不允许重复下单!");
        return;
    }
    try {
        // 5. 使用代理对象，由于这里是另外一个线程，
        proxy.createVoucherOrder(voucherOrder);
    } finally {
        redisLock.unlock();
    }
}

这里有一个细节的地方，我们通过查看AopContext的源码发现，其获取代理对象的方式是通过ThreadLocal，由于我们现在是异步下单，已经不在主线程中了，所以直接获取是不行的。

1
private static final ThreadLocal<Object> currentProxy = new NamedThreadLocal("Current AOP proxy");

修改秒杀业务（在主线程中获取代理对象并存放在成员变量中）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Override
public Result seckillVoucher(Long voucherId) {
    Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
            Collections.emptyList(), voucherId.toString(),
            UserHolder.getUser().getId().toString());
    if (result.intValue() != 0) {
        return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");
    }
    long orderId = redisIdWorker.nextId("order");
    // 封装到voucherOrder中
    VoucherOrder voucherOrder = new VoucherOrder();
    voucherOrder.setVoucherId(voucherId);
    voucherOrder.setUserId(UserHolder.getUser().getId());
    voucherOrder.setId(orderId);
    // 加入到阻塞队列
    orderTasks.add(voucherOrder);
    // 主线程获取代理对象
    proxy = (IVoucherOrderService) AopContext.currentProxy();
    return Result.ok(orderId);
}

基于阻塞队列实现秒杀优化完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
package com.hmdp.service.impl;

import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Collections;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    @Autowired
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedissonClient redissonClient;

    private IVoucherOrderService proxy;


    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    private final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 1. 获取用户
        Long userId = voucherOrder.getUserId();
        // 2. 创建锁对象，作为兜底方案
        RLock redisLock = redissonClient.getLock("order:" + userId);
        // 3. 获取锁
        boolean isLock = redisLock.tryLock();
        // 4. 判断是否获取锁成功 
        if (!isLock) {
            log.error("不允许重复下单!");
            return;
        }
        try {
            // 5. 使用代理对象，由于这里是另外一个线程，
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            redisLock.unlock();
        }
    }

    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 1. 获取队列中的订单信息
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 2. 创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("订单处理异常", e);
                }
            }
        }
    }

    @Override
    public Result seckillVoucher(Long voucherId) {
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                Collections.emptyList(), voucherId.toString(),
                UserHolder.getUser().getId().toString());
        if (result.intValue() != 0) {
            return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");
        }
        long orderId = redisIdWorker.nextId("order");
        // 封装到voucherOrder中
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(UserHolder.getUser().getId());
        voucherOrder.setId(orderId);
        // 加入到阻塞队列
        orderTasks.add(voucherOrder);
        // 主线程获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        return Result.ok(orderId);
    }


    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 一人一单逻辑
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        synchronized (userId.toString().intern()) {
            int count = query().eq("voucher_id", voucherId).eq("user_id", userId).count();
            if (count > 0) {
                log.error("你已经抢过优惠券了哦");
                return;
            }
            // 5. 扣减库存
            boolean success = seckillVoucherService.update()
                    .setSql("stock = stock - 1")
                    .eq("voucher_id", voucherId)
                    .gt("stock", 0)
                    .update();
            if (!success) {
                log.error("库存不足");
            }
            // 7. 将订单数据保存到表中
            save(voucherOrder);
        }
    }
}

秒杀优化小结

秒杀业务的优化思路

  • 先利用Redis完成库存容量、一人一单的判断，完成抢单业务
  • 再将下单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题

内存限制问题： 我们现在使用的是JDK自带的阻塞队列，它使用的是JVM的内存，如果在高并发的条件下，大量的订单都会放进阻塞队列里，可能就会造成内存溢出，所以我们在创建阻塞队列时，设置了一个长度，但是如果真的存满了，再有新的订单来往里塞，那就塞不进去了，存在内存限制问题
数据安全问题：

  • 如果服务器宕机，内存中的信息全部丢失，导致已付款却无订单
  • 从异步队列中拿出了订单却因为异常没有消费，那就永远也不会被消费了

Redis消息队列

认识消息队列

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

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

使用队列的好处在于解耦：举个例子，快递员(生产者)把快递放到驿站/快递柜里去(Message Queue)去，我们(消费者)从快递柜/驿站去拿快递，这就是一个异步，如果耦合，那么快递员必须亲自上楼把快递递到你手里，服务当然好，但是万一我不在家，快递员就得一直等我，浪费了快递员的时间。所以解耦还是非常有必要的
那么在这种场景下我们的秒杀就变成了：在我们下单之后，利用Redis去进行校验下单的结果，然后在通过队列把消息发送出去，然后在启动一个线程去拿到这个消息，完成解耦，同时也加快我们的响应速度
对于小中型项目完全可以使用Redis中的Stream作为消息队列，对于一些对消息队列有较高要求的项目，可以建议使用现有的消息队列框架，如RabitMq，RocketMq，Kafka。

基于Redis-List实现消息队列

基于List结构模拟消息队列
消息队列(Message Queue)，字面意思就是存放消息的队列，而Redis的list数据结构是一个双向链表，很容易模拟出队列的效果
队列的入口和出口不在同一边，所以我们可以利用：LPUSH结合RPOP或者RPUSH结合LPOP来实现消息队列。
不过需要注意的是，当队列中没有消息时，RPOP和LPOP操作会返回NULL，而不像JVM阻塞队列那样会阻塞，并等待消息，所以我们这里应该使用BRPOP或者BLPOP来实现阻塞效果
47.png
基于List的消息队列有哪些优缺点？
优点：

  • 利用Redis存储，不受限于JVM内存上限
  • 基于Redis的持久化机制，数据安全性有保障
  • 可以满足消息有序性

缺点：

  • 只支持单消费者(一个消息只能被读一次)
  • 无法避免消息丢失(如果读了消息之后服务器宕机，又由于只能被读一次，所以直接寄)

基于PubSub的消息队列

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。顾名思义，消费和可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息

  • SUBSCRIBE channel [channel]：订阅一个或多个频道
  • PUBLISH channel msg：向一个频道发送消息
  • PSUBSCRIBE pattern [pattern]：订阅与pattern格式匹配的所有频道

官网对于partern的解释：
Subscribes the client to the given patterns.
Supported glob-style patterns:

  • h?flo subscribes to hello, hallo and hxllo
  • h*llo subscribes to hllo and heeeello
  • h[ae]llo subscribes to hello and hallo, but not hillo

Use \ to escape special characters if you want to match them verbatim.

基于PubSub的消息队列有哪些优缺点
优点：

  • 采用发布订阅模型，支持多生产，多消费

缺点：

  • 不支持数据持久化
  • 无法避免消息丢失（如果向频道发送了消息，却没有人订阅该频道，那发送的这条消息就丢失了）
  • 消息堆积有上限，超出时数据丢失（消费者拿到数据的时候处理的太慢，而发送消息发的太快）

基于Stream的单消费消息队列

Stream是Redis 5.0引入的一种新数据类型，可以实现一个功能非常完善的消息队列。
发送消息的命令：XADD
48.png

例如：

49.png
读取消息的方式之一：XREAD
50.png

例如使用XREAD读取第一个消息：

51.png

例如让XREAD阻塞方式读取最新消息：

52.png
在业务开发中，我们可以使用循环调用的XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下：

1
2
3
4
5
6
7
8
9
while (true) {
    // 尝试读取队列中的消息，最多阻塞2秒
    Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $");
    if (msg == null) {
        continue;
    }
    // 处理消息
    handleMessage(msg);
}

注意：当我们指定其实ID为$时，代表只能读取到最新消息，如果当我们在处理一条消息的过程中，又有超过1条以上的消息到达队列，那么下次获取的时候，也只能获取到最新的一条，会出现漏读消息的问题

STREAM类型消息队列的XREAD命令特点：

  • 消息可回溯
  • 一个消息可以被多个消费者读取
  • 可以阻塞读取
  • 有漏读消息的风险

基于Stream的消费者组消息队列

消费者组(Consumer Group)：将多个消费者划分到一个组中，监听同一个队列，具备以下特点：
53.png
创建消费者组的命令：

1
XGROUP CREATE key groupName ID [MKSTREAM]

key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列

其他常见命令：

1
2
3
4
5
6
7
8
# 删除指定消费组
XGROUP DESTORY key groupName

# 给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupName consumerName

# 删除消费者组中指定的消费者
XGROUP DELCONSUMER key groupName consumerName

从消费者组中读取消息：

1
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

group：消费组名称
consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
count：本次查询的最大数量
BLOCK milliseconds：当没有消息时最长等待时间
NOACK：无需手动ACK，获取到消息后自动确认（不建议）
STREAMS key：指定队列名称
ID：获取消息的起始ID，ID只有两种：

  • >：从下一个未消费的消息开始（非pending-list中的）
  • 其他：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

消费者组监听消息的基本思路（伪代码）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
while(true) {
    // 尝试监听队列，使用阻塞模式，最大等待时长为2000ms
    Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >")
    if(msg == null) {
        // 没监听到消息，重试
        continue;
    }
    try{
        // 处理消息，完成后要手动确认ACK，ACK代码在handleMessage中编写
        handleMessage(msg);
    } catch(Exception e) {
        while(true) {
            // 0表示从pending-list中的第一个消息开始，如果前面都ACK了，那么这里就不会监听到消息
            Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0");
            if(msg == null) {
                // null表示没有异常消息，所有消息均已确认，结束循环
                break;
            }
            try{
                // 说明有异常消息，再次处理
                handleMessage(msg);
            } catch(Exception e) {
                // 再次出现异常，记录日志，继续循环
                log.error("..");
                continue;
            }
        }
    }
}

STREAM类型消息队列的XREADGROUP命令的特点：

  • 消息可回溯
  • 可以多消费者争抢消息，加快消费速度
  • 可以阻塞读取
  • 没有消息漏读风险
  • 有消息确认机制，保证消息至少被消费一次

Redis实现消息队列的三种方式对比：
54.png

基于Stream的消费者组优化秒杀

本小节需求：

  • 创建一个Stream类型的消息队列，名为stream.orders
  • 修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
  • 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

在redis中创建名为stream.orders的消息队列：

1
XGROUP CREATE stream.orders g1 0 MKSTREAM

修改Lua脚本，新增orderId参数，并将订单信息加入到消息队列中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
-- 订单id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]
-- 新增orderId，但是变量名用id就好，因为VoucherOrder实体类中的orderId就是用id表示的
local id = ARGV[3]
-- 优惠券key
local stockKey = 'seckill:stock:' .. voucherId
-- 订单key
local orderKey = 'seckill:order:' .. voucherId
-- 判断库存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0) then
    return 1
end
-- 判断用户是否下单
if (redis.call('sismember', orderKey, userId) == 1) then
    return 2
end
-- 扣减库存
redis.call('incrby', stockKey, -1)
-- 将userId存入当前优惠券的set集合
redis.call('sadd', orderKey, userId)
-- 将下单数据保存到消息队列中
redis.call("sadd", 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', id)
return 0

修改秒杀逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
public Result seckillVoucher(Long voucherId) {
    long orderId = redisIdWorker.nextId("order");
    Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
            Collections.emptyList(), voucherId.toString(),
            UserHolder.getUser().getId().toString(), String.valueOf(orderId));
    if (result.intValue() != 0) {
        return Result.fail(result.intValue() == 1 ? "库存不足" : "不能重复下单");
    }
    // 主线程获取代理对象
    proxy = (IVoucherOrderService) AopContext.currentProxy();
    return Result.ok(orderId);
}

修改VoucherOrderHandler：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
String queueName = "stream.orders";

private class VoucherOrderHandler implements Runnable {

    @Override
    public void run() {
        while (true) {
            try {
                // 1. 获取队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        // ReadOffset.lastConsumed()底层就是 '>'
                        StreamOffset.create(queueName, ReadOffset.lastConsumed()));
                // 2. 判断消息是否获取成功
                if (records == null || records.isEmpty()) {
                    continue;
                }
                // 3. 消息获取成功之后，我们需要将其转为对象
                MapRecord<String, Object, Object> record = records.get(0);
                Map<Object, Object> values = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                // 4. 获取成功，执行下单逻辑，将数据保存到数据库中
                handleVoucherOrder(voucherOrder);
                // 5. 手动ACK，SACK stream.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("订单处理异常", e);
                // 订单异常的处理方式我们封装成一个函数，避免代码太臃肿
                handlePendingList();
            }
        }
    }
}

private void handlePendingList() {
    while (true) {
        try {
            // 1. 获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders 0
            List<MapRecord<String, Object, Object>> records = stringRedisTemplate.opsForStream().read(
                    Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1),
                    StreamOffset.create(queueName, ReadOffset.from("0")));
            // 2. 判断pending-list中是否有未处理消息
            if (records == null || records.isEmpty()) {
                // 如果没有就说明没有异常消息，直接结束循环
                break;
            }
            // 3. 消息获取成功之后，我们需要将其转为对象
            MapRecord<String, Object, Object> record = records.get(0);
            Map<Object, Object> values = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
            // 4. 获取成功，执行下单逻辑，将数据保存到数据库中
            handleVoucherOrder(voucherOrder);
            // 5. 手动ACK，SACK stream.orders g1 id
            stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
        } catch (Exception e) {
            log.info("处理pending-list异常");
            // 如果怕异常多次出现，可以在这里休眠一会儿
            try {
                Thread.sleep(50);
            } catch (InterruptedException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}

达人探店

在本小节，我们主要需要实现探店笔记的点赞功能和点赞排行榜功能。

假设在此之前已经实现了发布探店笔记和查看探店笔记排行榜功能。

点赞功能

需求背景： 我们对于一篇笔记，如果不校验用户是否点赞，用户就可以一直点赞，显然不符合预期。
需求：

  • 当用户未点赞时，点赞成功，点赞数 + 1
  • 当用户已点赞时，取消点赞，点赞数 - 1

解决方案： 我们可以利用Redis中的Set数据类型实现快速校验用户是否已经点赞。
Controller：

1
2
3
4
@PutMapping("/like/{id}")
public Result likeBlog(@PathVariable("id") Long id) {
    return blogService.likeBlog(id);
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Override
public Result likeBlog(Long id) {
    // 1. 获取当前用户信息
    Long userId = UserHolder.getUser().getId();
    // 2. 如果当前用户未点赞，则点赞数 +1，同时将用户加入set集合
    String key = BLOG_LIKED_KEY + id;
    Boolean isLiked = stringRedisTemplate.opsForSet().isMember(key, userId.toString());
    if (BooleanUtil.isFalse(isLiked)) {
        // 点赞数 +1
        boolean success = update().setSql("liked = liked + 1").eq("id", id).update();
        // 将用户加入set集合
        if (success) {
            stringRedisTemplate.opsForSet().add(key, userId.toString());
        }
    // 3. 如果当前用户已点赞，则取消点赞，将用户从set集合中移除
    } else {
        // 点赞数 -1
        boolean success = update().setSql("liked = liked - 1").eq("id", id).update();
        if (success) {
            // 从set集合移除
            stringRedisTemplate.opsForSet().remove(key, userId.toString());
        }
    }
    return Result.ok();
}

点赞排行榜

需求背景： 在探店笔记的详情页面，应该把给该条笔记点赞的人显示出来，例如显示最早点赞的5个人，形成点赞排行榜。
55.png
解决方案： 我们之前是将点赞的人放到Set集合中的，但是Set并不支持排序，所以现在需要修改成SortedSet(Zset)，在此，我们顺便对比一下几种集合的区别：
56.png

这里有一个小细节，如果我们使用Zset，那么点赞功能就要做相应的修改，由于Zset中没有SISMEMBER方法，所以需要通过查询分数来判断元素是否在集合中，如果查询到分数，则说明有该元素，否则该元素不存在。

Controller：

1
2
3
4
@GetMapping("/likes/{id}")
public Result queryBlogLikes(@PathVariable Integer id) {
    return blogService.queryBlogLikes(id);
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Override
public Result queryBlogLikes(Integer id) {
    String key = BLOG_LIKED_KEY + id;
    // zrange key 0 4  查询zset中前5个元素
    Set<String> top5 = stringRedisTemplate.opsForZSet().range(key, 0, 4);
    // 如果是空的(可能没人点赞)，直接返回一个空集合
    if (top5 == null || top5.isEmpty()) {
        return Result.ok(Collections.emptyList());
    }
    List<Long> ids = top5.stream().map(Long::valueOf).collect(Collectors.toList());
    // 将ids使用`,`拼接，SQL语句查询出来的结果并不是按照我们期望的方式进行排
    // 所以我们需要用order by field来指定排序方式，期望的排序方式就是按照查询出来的id进行排序
    String idsStr = StrUtil.join(",", ids);
    // SELECT * FROM tb_user WHERE id IN (ids[0], ids[1] ...) ORDER BY FIELD(id, ids[0], ids[1] ...)
    List<UserDTO> userDTOS = userService.query().in("id", ids)
            .last("order by field(id," + idsStr + ")")
            .list().stream()
            .map(user -> BeanUtil.copyProperties(user, UserDTO.class))
            .collect(Collectors.toList());
    return Result.ok(userDTOS);
}

好友关注

查看用户笔记

当我们对某个博主感兴趣时，会点进其主页看看她/他发布过哪些笔记，以便深入了解。

1
2
3
4
5
6
7
8
9
10
11
12
// 逻辑较为简单，就没写Service，不太建议这样实现
@GetMapping("/of/user")
public Result queryBlogByUserId(
		@RequestParam(value = "current", defaultValue = "1") Integer current,
		@RequestParam("id") Long id) {
	// 根据用户查询
	Page<Blog> page = blogService.query()
			.eq("user_id", id).page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
	// 获取当前页数据
	List<Blog> records = page.getRecords();
	return Result.ok(records);
}

关注与取消关注

在生活中有许多场景涉及到关注与取消关注，所以该功能用处广泛，本小节将实现用户的关注与取消关注功能。
首先我们抽象出每一次关注行为，并记录到follow表中。follow表结构如下表所示：

Field Type Collation Null Key Default Extra Comment
id bigint (NULL) NO PRI (NULL) auto_increment 主键
user_id bigint unsigned (NULL) NO (NULL) 用户id
follow_user_id bigint unsigned (NULL) NO (NULL) 关联的用户id
create_time timestamp (NULL) NO CURRENT_TIMESTAMP DEFAULT_GENERATED 创建时间

follow表对应的实体类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@TableName("tb_follow")
public class Follow implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 主键
     */
    @TableId(value = "id", type = IdType.AUTO)
    private Long id;

    /**
     * 用户id
     */
    private Long userId;

    /**
     * 关联的用户id
     */
    private Long followUserId;

    /**
     * 创建时间
     */
    private LocalDateTime createTime;
}

Controller：

1
2
3
4
5
6
7
8
9
10
// 关注
@PutMapping("/{id}/{isFollow}")
public Result follow(@PathVariable("id") Long followUserId, @PathVariable("isFollow") Boolean isFollow) {
    return followService.follow(followUserId, isFollow);
}
// 取消关注
@GetMapping("/or/not/{id}")
public Result isFollow(@PathVariable("id") Long followUserId) {
      return followService.isFollow(followUserId);
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// 查询是否关注
@Override
public Result isFollow(Long followUserId) {
    // 1.获取登录用户
    Long userId = UserHolder.getUser().getId();
    // 2.查询是否关注 select count(*) from tb_follow where user_id = ? and follow_user_id = ?
    Integer count = query().eq("user_id", userId).eq("follow_user_id", followUserId).count();
    // 3.判断
    return Result.ok(count > 0);
}

// 关注与取消关注
@Override
public Result follow(Long followUserId, Boolean isFollow) {
    // 1.获取登录用户
    Long userId = UserHolder.getUser().getId();
    String key = "follows:" + userId;
    // 1.判断到底是关注还是取关
    if (isFollow) {
        // 2.关注，新增数据
        Follow follow = new Follow();
        follow.setUserId(userId);
        follow.setFollowUserId(followUserId);
        save(follow);
    } else {
        // 3.取关，删除 delete from tb_follow where user_id = ? and follow_user_id = ?
        remove(new QueryWrapper<Follow>()
                .eq("user_id", userId).eq("follow_user_id", followUserId));
    }
    return Result.ok();
}

共同关注

当我们有了关注功能后，我们去访问别人主页时，我们还希望看到共同关注的人有哪些，这个功能在很多平台也都有广泛应用。那么我们该怎么去实现这个功能呢？
共同关注实现方案：
在Redis中有Set这样的数据结构，我们可以通过INTERSECT命令快速求出两个集合的交集。所以，我们将采用这个数据结构实现共同关注功能。

上一小节中，我们实现关注与取消关注时并没有使用Redis，所以我们需要修改一下代码，维护好Set集。

修改关注与取消关注：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@Resource
private StringRedisTemplate stringRedisTemplate;

@Override
public Result follow(Long followUserId, Boolean isFellow) {
    // 获取当前用户id
    Long userId = UserHolder.getUser().getId();
    String key = "follows:" + userId;
    // 判断是否关注
    if (isFellow) {
        // 关注，则将信息保存到数据库
        Follow follow = new Follow();
        follow.setUserId(userId);
        follow.setFollowUserId(followUserId);
        // 如果保存成功
        boolean success = save(follow);
        // 则将数据也写入Redis
        if (success) {
            stringRedisTemplate.opsForSet().add(key, followUserId.toString());
        }
    } else {
        // 取关，则将数据从数据库中移除
        LambdaQueryWrapper<Follow> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(Follow::getUserId, userId).eq(Follow::getFollowUserId, followUserId);
        // 如果取关成功
        boolean success = remove(queryWrapper);
        // 则将数据也从Redis中移除
        if (success) {
            stringRedisTemplate.opsForSet().remove(key,followUserId.toString());
        }
    }
    return Result.ok();
}

共同关注Controller：

1
2
3
4
@GetMapping("/common/{id}")
public Result followCommons(@PathVariable Long id) {
    return followService.followCommons(id);
}

共同关注ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Override
public Result followCommons(Long id) {
    // 获取当前用户id
    Long userId = UserHolder.getUser().getId();
    String key1 = "follows:" + id;
    String key2 = "follows:" + userId;
    // 对当前用户和博主用户的关注列表取交集
    Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key1, key2);
    if (intersect == null || intersect.isEmpty()) {
        // 无交集就返回个空集合
        return Result.ok(Collections.emptyList());
    }
    // 将结果转为list
    List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());
    // 之后根据ids去查询共同关注的用户，封装成UserDto再返回
    List<UserDTO> userDTOS = userService.listByIds(ids).stream().map(user ->
            BeanUtil.copyProperties(user, UserDTO.class)).collect(Collectors.toList());
    return Result.ok(userDTOS);
}

Feed流

Feed流简介

对于传统的模式内容检索：用户需要主动通过搜索引擎或者是其他方式去查找想看的内容
当我们关注了用户之后，这个用户发布了动态，那我们应该把这些数据推送给用户，这个需求，我们又称其为Feed流，关注推送也叫作Feed流，直译为投喂，为用户提供沉浸式体验，通过无限下拉刷新获取新的信息。
对于新型Feed流的效果：系统分析用户到底想看什么，然后直接把内容推送给用户，从而使用户能更加节约时间，不用去主动搜素。

Feed流实现方案

Feed流的实现有两种模式：
Timeline： 不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注(朋友圈就是基于此种方式实现)

  • 优点：信息全面，不会有缺失，并且实现也相对简单
  • 缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低

智能排序： 利用智能算法屏蔽掉违规的、用户不感兴趣的内容，推送用户感兴趣的信息来吸引用户

  • 优点：投喂用户感兴趣的信息，用户粘度很高，容易沉迷
  • 缺点：如果算法不精准，可能会起到反作用

由于我们针对的是好友关注关系的推送，因此采用Timeline方式：拿到关注的用户的博客，然后按照时间排序。
Timeline模式的三种实现方案：

  • 拉模式（读扩散）
  • 推模式（写扩散）
  • 推拉结合

拉模式：

  • 该模式的核心含义是：当张三和李四、王五发了消息之后，都会保存到自己的发件箱中，如果赵六要读取消息，那么他会读取他自己的收件箱，此时系统会从他关注的人群中，将他关注人的信息全都进行拉取，然后进行排序
  • 优点：比较节约空间，因为赵六在读取信息时，并没有重复读取，并且读取完之后，可以将他的收件箱清除
  • 缺点：有延迟，当用户读取数据时，才会去关注的人的时发件箱中拉取信息，假设该用户关注了海量用户，那么此时就会拉取很多信息，对服务器压力巨大

57.png
推模式：

  • 推模式是没有写邮箱的，当张三写了一个内容，此时会主动把张三写的内容发送到它粉丝的收件箱中，假设此时李四再来读取，就不用再去临时拉取了
  • 优点：时效快，不用临时拉取
  • 缺点：内存压力大，假设一个大V发了一个动态，很多人关注他，那么就会写很多份数据到粉丝那边去

58.png
推拉结合：
推拉模式是一个折中的方案，站在发件人这一边，如果是普通人，那么我们采用写扩散的方式，直接把数据写入到他的粉丝收件箱中，因为普通人的粉丝数量较少，所以这样不会产生太大压力。但如果是大V，那么他是直接将数据写入一份到发件箱中去，在直接写一份到活跃粉丝的收件箱中，站在收件人这边来看，如果是活跃粉丝，那么大V和普通人发的都会写到自己的收件箱里，但如果是普通粉丝，由于上线不是很频繁，所以等他们上线的时候，再从发件箱中去拉取信息。
59.png
Timeline的三种实现方式对比：
60.png
综上，我们将采用Timeline方式的推模式实现已关注推送功能。

基于推模式实现推送功能

需求及思路分析

本小节需求：

  • 修改新增探店笔记的业务，在保存blog到数据库的同时，推送到粉丝的收件箱（已关注用户的博客）
  • 收件箱满足可以根据时间戳排序（使用Redis实现）
  • 用户查看已关注收件箱时，做到分页查询

有一个需要注意的地方，我们不能采用传统的分页查询模式。例如，在第一次查询用户收件箱之后，可能某一个关注的用户突然更新一条博客，我们如果继续按照下标查询第二页，就会重复查询到第一页的最后一条记录。
61.png
那么怎么解决这个问题呢？我们使用滚动分页查询即可。

  • 我们需要记录每次操作的最后一条，然后从这个位置去开始读数据
  • 举个例子：我们从t1时刻开始，拿到第一页数据，拿到了106，然后记录下当前最后一次读取的记录，就是6，t2时刻发布了新纪录，此时这个11在最上面，但不会影响我们之前拿到的6，此时t3时刻来读取第二页，第二页读数据的时候，从6-1=5开始读，这样就拿到了51的记录。我们在这个地方可以使用SortedSet来做，使用时间戳来充当表中的1~10

62.png

补充发布博客代码

由于之前未介绍到发送博客功能，本小节结合推送功能，对发送博客代码进行书写和完善。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Override
public Result saveBlog(Blog blog) {
    // 获取登录用户
    UserDTO user = UserHolder.getUser();
    blog.setUserId(user.getId());
    // 保存探店博文
    save(blog);
    // 条件构造器
    LambdaQueryWrapper<Follow> queryWrapper = new LambdaQueryWrapper<>();
    // 从follow表最中，查找当前用户的粉丝  select * from follow where follow_user_id = user_id
    queryWrapper.eq(Follow::getFollowUserId, user.getId());
    // 获取当前用户的粉丝
    List<Follow> follows = followService.list(queryWrapper);
    for (Follow follow : follows) {
        Long userId = follow.getUserId();
        String key = FEED_KEY + userId;
        // 推送数据
        stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());
    }
    // 返回id
    return Result.ok(blog.getId());
}

实现分页查询收件箱

本小节将实现在个人主页的关注栏中查询并展示推送的博客信息。
实现步骤：

  • 编写一个通用的分页类
  • 根据参数进行查询
  • 每次查询完成之后，我们要分析出查询出的最小时间戳，这个值会作为下一次的查询条件（计算lastId）
  • 我们需要找到与上一次查询相同的查询个数，并作为偏移量，下次查询的时候，跳过这些查询过的数据，拿到我们需要的数据（计算offset）

我们的请求参数中需要携带lastId和offset，即上一次查询时的最小时间戳和偏移量，除了第一次，后面的每一次查询的参数会由前一次查询计算出

编写通用的分页查询结果类：

1
2
3
4
5
6
7
@Data
public class ScrollResult {
    // 为了达到通用的效果，我们使用泛型，这样也能用于查询博客意外的数据
    private List<?> list;
    private Long minTime;
    private Integer offset;
}

Controller：

1
2
3
4
@GetMapping("/of/follow")
public Result queryBlogOfFollow(@RequestParam("lastId") Long max, @RequestParam(value = "offset",defaultValue = "0") Integer offset) {
    return blogService.queryBlogOfFollow(max, offset);
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@Override
public Result queryBlogOfFollow(Long max, Integer offset) {
    // 1. 获取当前用户
    Long userId = UserHolder.getUser().getId();
    // 2. 查询该用户收件箱（之前我们存的key是固定前缀 + 粉丝id），所以根据当前用户id就可以查询是否有关注的人发了笔记
    String key = FEED_KEY + userId;
    Set<ZSetOperations.TypedTuple<String>> typeTuples = stringRedisTemplate.opsForZSet()
            .reverseRangeByScoreWithScores(key, 0, max, offset, 2);
    // 3. 非空判断
    if (typeTuples == null || typeTuples.isEmpty()) {
        return Result.ok(Collections.emptyList());
    }
    // 4. 解析数据，blogId、minTime（时间戳）、offset，这里指定创建的list大小，可以略微提高效率，因为我们知道这个list就得是这么大
    ArrayList<Long> ids = new ArrayList<>(typeTuples.size());
    long minTime = 0;
    int os = 1;
    for (ZSetOperations.TypedTuple<String> typeTuple : typeTuples) {
        // 4.1 获取id
        String id = typeTuple.getValue();
        ids.add(Long.valueOf(id));
        // 4.2 获取score（时间戳）
        long time = typeTuple.getScore().longValue();
        if (time == minTime) {
            os ++;
        }else {
            minTime = time;
            os = 1;
        }
    }
    // 解决SQL的in不能排序问题，手动指定排序为传入的ids
    String idsStr = StrUtil.join(",");
    // 5. 根据id查询blog
    List<Blog> blogs = query().in("id", ids).last("ORDER BY FIELD(id, " + idsStr + ")").list()
    for (Blog blog : blogs) {
        // 5.1 查询发布该blog的用户信息
        queryBlogUser(blog);
        // 5.2 查询当前用户是否给该blog点过赞
        isBlogLiked(blog);
    }
    // 6. 封装结果并返回
    ScrollResult scrollResult = new ScrollResult();
    scrollResult.setList(blogs);
    scrollResult.setOffset(os);
    scrollResult.setMinTime(minTime);
    return Result.ok(scrollResult);
}

附近商户

GEO数据结构

GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据，常见的命令有：

  • GEOADD：添加一个地理空间信息，包含：经度（longitude）、纬度（latitude）、值（member）
  • GEODIST：计算指定的两个点之间的距离并返回
  • GEOHASH：将指定member的坐标转为hash字符串形式并返回
  • GEOPOS：返回指定member的坐标
  • GEORADIUS（6.2以后已废弃）：指定圆心、半径，找到该圆内包含的所有member，并按照与圆心之间的距离排序后返回。
  • GEOSEARCH（6.2.新功能）：在指定范围内搜索member，并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。
  • GEOSEARCHSTORE（6.2.新功能）：与GEOSEARCH功能一致，不过可以把结果存储到一个指定的key。

导入店铺数据到Redis-GEO

具体场景说明，例如美团/饿了么这种外卖App，你是可以看到商家离你有多远的，那我们现在也要实现这个功能。
我们可以使用GEO来实现该功能，以当前坐标为圆心，同时绑定相同的店家类型type，以及分页信息，把这几个条件插入后台，后台查询出对应的数据再返回。
那现在我们要做的就是：将数据库中的数据导入到Redis中去，GEO在Redis中就是一个member和一个经纬度，经纬度对应的就是tb_shop中的x和y，而member，我们用shop_id来存，因为Redis只是一个内存级数据库，如果存海量的数据，还是力不从心，所以我们只存一个id，用的时候再拿id去SQL数据库中查询shop信息。
但是此时还有一个问题，我们在redis中没有存储shop_type，无法根据店铺类型来对数据进行筛选，解决办法就是将type_id作为key，存入同一个GEO集合即可
63.png
代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Test
public void loadShopData() {
    List<Shop> shopList = shopService.list();
    Map<Long, List<Shop>> map = shopList.stream().collect(Collectors.groupingBy(Shop::getTypeId));
    // 处理每一个类别
    for (Map.Entry<Long, List<Shop>> entry : map.entrySet()) {
        Long typeId = entry.getKey();
        List<Shop> shops = entry.getValue();
        String key = SHOP_GEO_KEY + typeId;
        List<RedisGeoCommands.GeoLocation<String>> locations = new ArrayList<>(shops.size());
        // 处理每一个类别中的所有店铺
        for (Shop shop : shops) {
            // 将当前type的商铺都添加到locations集合中
            locations.add(new RedisGeoCommands.GeoLocation<>(shop.getId().toString(), new Point(shop.getX(), shop.getY())));
        }
        // 批量写入
        stringRedisTemplate.opsForGeo().add(key, locations);
    }
}

实现查询附近商户功能

在实现之前还有一个注意的地方，由于GEOSEARCH是Redis 6.2才提供的，因此如果SpringDataRedis版本较低，则需要修改pom.xml文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <exclusions>
        <exclusion>
            <artifactId>spring-data-redis</artifactId>
            <groupId>org.springframework.data</groupId>
        </exclusion>
        <exclusion>
            <artifactId>lettuce-core</artifactId>
            <groupId>io.lettuce</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-redis</artifactId>
    <version>2.6.2</version>
</dependency>
<dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.1.6.RELEASE</version>
</dependency>

Controller：

1
2
3
4
5
6
7
8
9
@GetMapping("/of/type")
public Result queryShopByType(
        @RequestParam("typeId") Integer typeId,
        @RequestParam(value = "current", defaultValue = "1") Integer current,
        @RequestParam(value = "x", required = false) Double x,
        @RequestParam(value = "y", required = false) Double y
) {
   return shopService.queryShopByType(typeId,current,x,y);
}

ServiceImpl（细节超多）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@Override
public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {
    // 1. 判断是否需要根据距离查询
    if (x == null || y == null) {
        // 根据类型分页查询
        Page<Shop> page = query()
                .eq("type_id", typeId)
                .page(new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE));
        // 返回数据
        return Result.ok(page.getRecords());
    }
    // 2. 计算分页查询参数
    int from = (current - 1) * SystemConstants.MAX_PAGE_SIZE;
    int end = current * SystemConstants.MAX_PAGE_SIZE;
    String key = SHOP_GEO_KEY + typeId;
    // 3. 查询redis、按照距离排序、分页; 结果：shopId、distance
    // GEOSEARCH key FROMLONLAT x y BYRADIUS 5000 m WITHDIST
    GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo().search(key,
            GeoReference.fromCoordinate(x, y),
            new Distance(5000),
            RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end));
    if (results == null) {
        return Result.ok(Collections.emptyList());
    }
    // 4. 解析出id
    List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();
    if (list.size() < from) {
        // 起始查询位置大于数据总量，则说明没数据了，返回空集合
        return Result.ok(Collections.emptyList());
    }
    ArrayList<Long> ids = new ArrayList<>(list.size());
    HashMap<String, Distance> distanceMap = new HashMap<>(list.size());
    list.stream().skip(from).forEach(result -> {
        String shopIdStr = result.getContent().getName();
        ids.add(Long.valueOf(shopIdStr));
        Distance distance = result.getDistance();
        distanceMap.put(shopIdStr, distance);
    });
    // 5. 根据id查询shop
    String idsStr = StrUtil.join(",", ids);
    List<Shop> shops = query().in("id", ids).last("ORDER BY FIELD( id," + idsStr + ")").list();
    for (Shop shop : shops) {
        // 设置shop的举例属性，从distanceMap中根据shopId查询
        shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());
    }
    // 6. 返回
    return Result.ok(shops);
}

用户签到

实现思路及BitMap介绍

对于用户签到的功能，应用场景也可谓是广泛，如果我们使用数据库来完成会是什么样子呢？
如果把用户的每次签到当成一条记录，假设1000万用户，平均每人一年签到10次（不过分吧），那一年下来这张表就有1亿条数据，这是多么恐怖的数据量，并且每条记录就算10多字节，这存储空间压力也山大呀！
所以，我们该怎么解决呢？
这里我们借鉴状态压缩的思想，将每个用户一个月的签到状态压缩为一个31位的二进制串，0代表未签到，1代表已签到，而这恰好能使用Redis中的BitMap数据结构完成。
BitMap的常见命令：

  • SETBIT：向指定位置（offset）存入一个0或1
  • GETBIT：获取指定位置（offset）的bit值
  • BITCOUNT：统计BitMap中值为1的bit位的数量
  • BITFIELD：操作（查询、修改、自增）BitMap中bit数组中的指定位置（offset）的值
  • BITFIELD_RO：获取BitMap中bit数组，并以十进制形式返回
  • BITOP：将多个BitMap的结果做位运算（与、或、异或）
  • BITPOS：查找bit数组中指定范围内第一个0或1出现的位置

实现签到功能

需求：实现用户当天签到功能

由于BitMap底层是基于String数据结构，因此对其的操作在Java中也都封装在字符串相关操作中了

Controller：

1
2
3
4
@PostMapping("/sign")
public Result sign() {
    return userService.sign();
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Override
public Result sign() {
    // 1. 获取当前用户
    Long userId = UserHolder.getUser().getId();
    // 2. 获取日期
    LocalDateTime now = LocalDateTime.now();
    // 3. 拼接key
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = USER_SIGN_KEY + userId + keySuffix;
    // 4. 获取今天是当月第几天(1~31)
    int dayOfMonth = now.getDayOfMonth();
    // 5. 写入Redis  BITSET key offset 1
    stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true);
    return Result.ok();
}

连续签到统计

需求：统计出从今天开始往前的连续签到天数

其实就是一个位运算的基本操作

Controller：

1
2
3
4
@GetMapping("/sign/count")
public Result signCount() {
    return userService.signCount();
}

ServiceImpl：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Override
public Result signCount() {
    // 1. 获取当前用户
    Long userId = UserHolder.getUser().getId();
    // 2. 获取日期
    LocalDateTime now = LocalDateTime.now();
    // 3. 拼接key
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = USER_SIGN_KEY + userId + keySuffix;
    // 4. 获取今天是当月第几天(1~31)
    int dayOfMonth = now.getDayOfMonth();
    // 5. 获取截止至今日的签到记录  BITFIELD key GET uDay 0
    List<Long> result = stringRedisTemplate.opsForValue().bitField(key, BitFieldSubCommands.create()
            .get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0));
    if (result == null || result.isEmpty()) {
        return Result.ok(0);
    }
    // 6. 循环遍历
    int count = 0;
    Long num = result.get(0);
    while (true) {
        if ((num & 1) == 0) {
            break;
        } else
            count++;
        // 数字无符号右移，抛弃最后一位
        num >>>= 1;
    }
    return Result.ok(count);
}

利用BitMap实现简单布隆过滤器

之前的章节中我们介绍过缓存穿透的问题，其中有一个方案就是利用布隆过滤器快速判断查询的id在数据库中是否存在，那么，我们来看看其实现思想是什么。
我们对每一个id进行相同的哈希运算，得到一个值，然后将这个值在BitMap上对应的位修改为1来表示这个id存在，0则表示不存在。
既然是哈希，那冲突是肯定避免不了的，这也就是布隆过滤器会有误判的存在（当然，别人那是优化过的，利用多次哈希较少冲突）
64.png

UV统计

概念介绍与HyperLogLog简介

UV： 全称Unique Visitor，也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次。
PV： 全称Page View，也叫页面访问量或点击量，用户每访问网站的一个页面，记录1次PV，用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。

本博客首页底部就有本网站的访问人数及浏览总量，对应的就是UV与PV。

UV统计在服务端做会很麻烦，因为要判断该用户是否已经统计过了，需要将统计过的信息保存，但是如果每个访问的用户都保存到Redis中，那么数据库会非常恐怖，那么该如何处理呢？
HyperLogLog(HLL)是从Loglog算法派生的概率算法，用户确定非常大的集合基数，而不需要存储其所有值，算法相关原理可以参考下面这篇文章：HyperLogLog 算法的原理讲解
Redis中的HLL是基于string结构实现的，单个HLL的内存永远小于16kb，内存占用低的令人发指！作为代价，其测量结果是概率性的，有小于0.81％的误差。不过对于UV统计来说，这完全可以忽略。
HyperLogLog的三个命令：

1
2
3
4
5
6
7
8
PFADD key element [element...]
summary: Adds the specified elements to the specified HyperLogLog

PFCOUNT key [key ...]
Return the approximated cardinality of the set(s) observed by the HyperLogLog at key(s).

PFMERGE destkey sourcekey [sourcekey ...]
lnternal commands for debugging HyperLogLog values

测试百万数据的统计

使用单元测试，向HyperLogLog中添加100万条数据，看看内存占用是否真的那么低，以及统计误差如何

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Test
public void testHyperLogLog() {
    String[] users = new String[1000];
    int j = 0;
    for (int i = 0; i < 1000000; i++) {
        j = i % 1000;
        users[j] = "user_" + i;
        // 每1000条数据向HyperLogLog中插入一次
        if (j == 999) {
            stringRedisTemplate.opsForHyperLogLog().add("HLL", users);
        }
    }
    Long count = stringRedisTemplate.opsForHyperLogLog().size("HLL");
    System.out.println("count = " + count);
}

插入100W条数据，得到的count为997593，误差率为0.002407%
去Redis图形化界面中查看占用情况为：14K字节（确实实现了吹的牛）