听书项目 Day 10｜排行榜与详情性能优化 | 𝐺ℎ𝑎𝑜 の 𝐵𝑙𝑜𝑔

梳理 Redis 在登录、统计、排行榜和幂等中的使用，并记录专辑排行榜与详情页并发优化思路。

本篇要点

盘点 Redis 在项目中的用途
实现专辑排行榜相关接口
分析详情页的并发与缓存问题

内容回顾

1、根据专辑id查询声音列表

1、根据专辑id获取专辑下面所有声音列表，但是声音包含免费和收费的

2、判断用户是否登录状态，如果用户没有登录：

– 判断如果专辑不是免费的，去掉试看的集数，其他是收费的

3、如果用户当前是登录状态：根据专辑类型判断

– 如果vip免费专辑：判断当前登录用户是否是vip用户，如果不是vip收费，如果是vip但是vip过期了也是收费

– 如果付费专辑：需要收费，但是查询当前用户是否购买专辑或者声音

— 如果用户购买过专辑，专辑里面声音可以免费看

— 如果用户购买过专辑里面某些声音，购买的声音可以免费看

2、添加（修改）声音播放进度

更新声音播放量– 同一个用户对于同一个声音在24小时计算一次播放量

使用redis里面bitmap类型实现
1
redis bitmap类型 key：userId 偏移量：声音id value：1
RabbitMQ幂等性保证– 相同消息发送多次只会消费一次

使用redis里面setnx实现（锁）

3、获取声音播放进度

今天内容

0、Redis项目中应用场景

第一个：登录存储用户信息

登录时候，生成token，把用户信息数据存储到redis里面，redis的key是token，value是用户信息。每次登录时候，请求头传递token，获取token查询redis，如果可以查询到登录状态

第二个：更新声音播放量

同一个用于对于同一个声音，在24小时计算一次播放量
使用redis里面bitmaps类型实现功能
bitmaps的key是用户id，offset是声音id，value是值

第三个：更新排行榜

使用redis的hash类型存储自己排行榜数据
hash类型key是一级分类id，field是排序字段，value是专辑数据

第四个：MQ保证消息幂等性

使用redis里面setnx（锁）机制解决
相同消息发送多次，只消费一次

第五个：分布式锁

使用Redis实现：setnx + 过期时间 + uuid + lua脚本

1、专辑排行榜功能

分析

查看每个一级分类里面按照不同方式排序出来专辑数据

查询mysql获取所有一级分类数据
查询es，根据一级分类id查询一级分类下面所有专辑，根据不同方式排序，获取前10条数据

GET /albuminfo/_search
{
  "query": {
    "term": {
      "category1Id": {
        "value": "1"
      }
    }
  }
  ,"sort": [
    {
      "playStatNum": {
        "order": "desc"
      }
    }
  ],
  "size": 10
}

定时任务后面实现
目前实现：编写获取排行榜数据接口

查询mysql获取一级分类数据

查询es得到专辑排行榜数据

把专辑排行榜数据存储到redis里面
使用redis里面哪个类型放数据？

– 使用redis里面hash类型存储排行榜数据

key：一级分类

field：排序字段

value：专辑排序数据

添加排行榜数据接口

在service-search模块的SearchApiController 中

/**
 * 更新排行榜
 * @return
 */
@SneakyThrows
@Operation(summary = "更新排行榜")
@GetMapping("updateLatelyAlbumRanking")
public Result updateLatelyAlbumRanking() {
    //  调用服务层方法
    searchService.updateLatelyAlbumRanking();
    //  默认返回
    return Result.ok();
}

@Autowired
private RedisTemplate redisTemplate;
//更新排行榜
@Override
public void updateLatelyAlbumRanking() throws Exception {
    //1 远程调用：获取所有一级分类
    Result<List<BaseCategory1>> result = categoryFeignClient.findAllCategory1();
    List<BaseCategory1> category1List = result.getData();
    Assert.notNull(category1List,"一级分类集合为空");

    //2 把所有一级分类list遍历，得到每个一级分类
    for (BaseCategory1 category1 : category1List) {
        //获取一级分类id
        Long category1Id = category1.getId();

        //创建数组，定义好排序字段
        String[] rankingDimensionArray =
                new String[]{"hotScore", "playStatNum",
                "subscribeStatNum", "buyStatNum", "commentStatNum"};
        //把数组遍历，得到每个排序字段
        //3 dsl语句，根据一级分类id + 排序字段查询排行榜数据
        for (String ranging : rankingDimensionArray) {
            //dsl语句
            SearchRequest.Builder builder = new SearchRequest.Builder();
            builder.index("albuminfo").query(
                    q->q.term(f->f.field("category1Id").value(category1Id)))
                    .sort(s->s.field(o->o.field(ranging).order(SortOrder.Desc)))
                    .size(10);

            //调用得到结果
            SearchResponse<AlbumInfoIndex> response =
                    elasticsearchClient.search(builder.build(),AlbumInfoIndex.class);

            //response获取结果
            List<AlbumInfoIndex> albumInfoIndexList = response.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
            //放到redis里面 hash类型
            // hash类型 key：一级分类id    field：排序字段   value：专辑数据
            //4 把查询专辑排行榜数据放到redis的hash类型里面
            String rangKey = "ranking:"+category1Id;
            //redisTemplate.boundHashOps(category1).put(ranging,albumInfoIndexList);
            redisTemplate.opsForHash().put(rangKey,ranging,albumInfoIndexList);
        }
    }
}

远程调用接口：查询所有一级分类

/**
 * 查询所有的一级分类数据
 * @return
 */
@Operation(summary = "查询所有的一级分类信息")
@GetMapping("findAllCategory1")
public Result<List<BaseCategory1>> findAllCategory1() {
    // 获取所有的一级分类数据
    List<BaseCategory1> baseCategory1List = baseCategoryService.findAllCategory1();
    return Result.ok(baseCategory1List);
}

查询排行榜数据接口

/**
 * 获取排行榜列表
 * @param category1Id
 * @param dimension
 * @return
 */
@Operation(summary = "获取排行榜列表")
@GetMapping("findRankingList/{category1Id}/{dimension}")
public Result<List<AlbumInfoIndexVo>>
             findRankingList(@PathVariable Long category1Id,
                             @PathVariable String dimension) {
    //  调用服务层方法
    List<AlbumInfoIndexVo> infoIndexVoList = searchService.findRankingList(category1Id, dimension);
    //  返回结果集
    return Result.ok(infoIndexVoList);
}

//查询排行榜数据
@Override
public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String dimension) {
    return (List<AlbumInfoIndexVo>)redisTemplate
            .boundHashOps("ranking:"+category1Id)
            .get(dimension);
}

2、详情优化（背）

2.1 缓存实现过程

数据库优化方式

如果专辑有大量用户进行访问，造成mysql压力过大（io瓶颈），对数据库进行优化
一般对系统进行优化，首先考虑的数据库优化
数据库优化有几种方式：

第一种创建索引（什么情况创建索引）

第二种 SQL语句优化（单表和多表优化）

第三种分库分表（单表500w行或者单表容量2GB）

第四种添加数据缓存（不经常变化数据）

缓存过程

2.2 缓存常见问题（★）

概述

添加缓存之后，产生以下几个问题
产生四个问题

第一个问题：缓存穿透

第二个问题：缓存雪崩

第三个问题：缓存击穿

第四个问题：数据一致性

1 缓存穿透问题

问题描述

是指查询一个不存在的数据，缓存中没有，mysql数据库也没有
在流量大时，可能mysql数据库就宕机了，缓存没有起到任何作用

解决方案

**方案一：**空结果也进行缓存（null值进行缓存），但是不能防止随机穿透

**方案二：**使用布隆过滤器来解决随机穿透问题（后面讲）

2 缓存雪崩问题

问题描述

是指把多个缓存设置了相同的过期时间，导致缓存在某一时刻同时失效，缓存刚失效恰巧有大量请求，压力都在mysql数据库，可能mysql数据库就宕机了

解决方案

让多个缓存设置随机过期时间（不同过期时间）

3 缓存击穿问题

问题描述

热点数据缓存过期了，还有大量请求过来，压力都会到mysql数据库，可能mysql数据库就宕机了

解决方案

使用分布式锁解决

4 数据一致性问题

问题描述

mysql里面数据发生变化，应该同步到redis缓存里面，但是mysql数据变化了，还没有来得及同步到redis，有请求查询redis，造成mysql和redis数据不一致

解决方案

延迟双删策略
数据同步工具canal

2.3 本地锁

概述

之前学习过，使用synchronized 及lock可以实现锁操作，这些锁都是本地锁
本地锁作用范围有局限性，只是作用于某一个jvm或者某一个微服务模块，如果跨jvm或者跨微服务模块不起作用了

演示本地锁局限性

@RestController
@RequestMapping("api/album/test")
public class TestController {

    //  注入服务层方法
    @Autowired
    private TestService testService;

    /**
     * 测试分布式锁
     * @return
     */
    @GetMapping("testLock")
    public Result testLock(){
        testService.testLock1();
        return Result.ok();
    }
}

@Service
public class TestServiceImpl implements  TestService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    //1 测试本地锁
    @Override
    public synchronized void testLock1() {
        //1 从redis获取数据
        //提前在redis手动添加初始值  set num  1
        String value = stringRedisTemplate.opsForValue().get("num");

        //2 判断非空
        if(StringUtils.isEmpty(value)) {
            return;
        }

        //3 把redis获取数据+1
        //4 +1之后放回到redis里面
        int num = Integer.parseInt(value);
        stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));
    }
}

测试

把模块部署集群效果，使用idea工具部署

– 到nacos配置文件注释端口号配置

– 在idea操作集群部署

使用三个服务集群部署

– 启动网关模块，三个专辑微服务模块

模拟100个请求，使用测试工具apache-jmeter

– 启动jmeter

2.4 分布式锁（Redis实现）

为了解决跨JVM加锁，使用分布式锁
分布式锁实现有很多种方案，使用Redis实现分布式锁，这种方式性能最高

Redis的setnx实现分布式锁

//setnx实现分布式锁
@Override
public void testLock2() throws Exception{
    //获取当前锁
    Boolean ifAbsent =
            stringRedisTemplate.opsForValue().setIfAbsent("lock", "lock");
    if(ifAbsent) { //获取锁成功
        //1 从redis获取数据
        //提前在redis手动添加初始值  set num  1
        String value = stringRedisTemplate.opsForValue().get("num");
        //2 判断非空
        if(StringUtils.isEmpty(value)) {
            return;
        }
        //3 把redis获取数据+1
        //4 +1之后放回到redis里面
        int num = Integer.parseInt(value);
        stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));

        //释放锁
        stringRedisTemplate.delete("lock");
    } else { //获取锁失败
        Thread.sleep(100);
        //自旋
        this.testLock2();
    }
}

锁无法释放问题

上面代码中，如果操作时候完成时候出现异常，造成锁无法释放
解决：

第一种把释放锁代码写到finally里面

第二种设置锁自动过期时间

//setnx实现分布式锁
@Override
public void testLock2() throws Exception{
    //获取当前锁  setIfAbsent == senx命令
    //设置过期时间
    Boolean ifAbsent =
            stringRedisTemplate.opsForValue()
                    .setIfAbsent("lock","lock",
                            10, TimeUnit.SECONDS);
    if(ifAbsent) { //获取锁成功
        try {
            //1 从redis获取数据
            //提前在redis手动添加初始值  set num  1
            String value = stringRedisTemplate.opsForValue().get("num");
            //2 判断非空
            if(StringUtils.isEmpty(value)) {
                return;
            }
            //3 把redis获取数据+1
            //4 +1之后放回到redis里面
            int num = Integer.parseInt(value);
            stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));

            //异常
        }finally {
            //释放锁
            stringRedisTemplate.delete("lock");
        }
    } else { //获取锁失败
        Thread.sleep(100);
        this.testLock2();
    }
}

锁误删问题

前提条件：业务执行时间大于锁自动释放时间
谁加的锁有谁去释放，解铃还须系铃人
**解决：**使用UUID 防止误删锁

//setnx实现分布式锁
@Override
public void testLock2() throws Exception{
    //获取当前锁  setIfAbsent == senx命令
    String uuid = UUID.randomUUID().toString();
    Boolean ifAbsent =
            stringRedisTemplate.opsForValue()
                    .setIfAbsent("lock",uuid,
                            3, TimeUnit.SECONDS);
    if(ifAbsent) { //获取锁成功
        try {
            //1 从redis获取数据
            //提前在redis手动添加初始值  set num  1
            String value = stringRedisTemplate.opsForValue().get("num");
            //2 判断非空
            if(StringUtils.isEmpty(value)) {
                return;
            }
            //3 把redis获取数据+1
            //4 +1之后放回到redis里面
            int num = Integer.parseInt(value);
            stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));

            //异常
        }finally {

            //释放锁
            //当前uuid 和 redis存储uuid是否相同
            String redis_uuid = stringRedisTemplate.opsForValue().get("lock");
            if(uuid.equals(redis_uuid)) {
                stringRedisTemplate.delete("lock");
            }

        }
    } else { //获取锁失败
        Thread.sleep(100);
        this.testLock2();
    }
}

原子性问题

保证释放锁过程具备原子性
解决方案：

– 使用Redis调用lua脚本保证释放锁原子性

– 基于Redis单线程特性

//setnx + 过期时间 + uuid + lua脚本
public void testLock3() throws Exception{
    //获取当前锁  setIfAbsent == senx命令
    String uuid = UUID.randomUUID().toString();
    Boolean ifAbsent =
            stringRedisTemplate.opsForValue()
                    .setIfAbsent("lock",uuid,
                            3, TimeUnit.SECONDS);
    if(ifAbsent) { //获取锁成功
        try {
            //1 从redis获取数据
            //提前在redis手动添加初始值  set num  1
            String value = stringRedisTemplate.opsForValue().get("num");
            //2 判断非空
            if(StringUtils.isEmpty(value)) {
                return;
            }
            //3 把redis获取数据+1
            //4 +1之后放回到redis里面
            int num = Integer.parseInt(value);
            stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));

            //异常
        }finally {

            //lua脚本释放锁
            //创建对象
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            //编写lua脚本
            String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                    "then\n" +
                    "    return redis.call(\"del\",KEYS[1])\n" +
                    "else\n" +
                    "    return 0\n" +
                    "end";
            //设置对象里面
            redisScript.setScriptText(script);
            //设置返回类型
            redisScript.setResultType(Long.class);
            //执行lua脚本
            stringRedisTemplate.execute(redisScript, Arrays.asList("lock"),uuid);
        }
    } else { //获取锁失败
        Thread.sleep(100);
        this.testLock2();
    }
}

总结：

1 分布式锁如何实现的？

使用Redis实现
setnx + 过期时间 + uuid + lua脚本

2 分布式锁实现同时满足以下四个条件

**互斥性。**在任意时刻，只有一个客户端能持有锁。
**不会发生死锁。**即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
**解铃还须系铃人。**加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
加锁和解锁必须具有原子性。