黑马Redis学习笔记 (基础篇+实战篇)

一.初始Redis

1.1SQL 和 NoSql的区别

1.1.1结构化和非结构化

(1) SQL关系性数据库

​ 传统关系型数据库是结构化数据，每一张表都有严格的约束信息：字段名、字段数据类型、字段约束等等信息，插入的数据必须遵守这些约束

(2) NoSql数据库

NoSql对数据库格式没有严格约束，往往形式松散，自由。

可以是key-value,可以是文档，或者图格式

1.1.2关联和非关联

(1) 关系型数据库

(2) 非关系型数据库

{
  id: 1,
  name: "张三",
  orders: [
    {
       id: 1,
       item: {
	 id: 10, title: "荣耀6", price: 4999
       }
    },
    {
       id: 2,
       item: {
	 id: 20, title: "小米11", price: 3999
      
       }
    }
  ]
}

1.1.3查询方式

1.1.4 事务

传统关系型数据库能满足事务ACID的原则 ，而非关系型数据库往往不支持事务，或者不能严格保证ACID的特性，只能实现基本的一致性。

1.1.5总结

1.2 认识Redis

特征：

• 键值（key-value）型，value支持多种不同数据结构，功能丰富

• 单线程，每个命令具备原子性

• 低延迟，速度快（基于内存、IO多路复用、良好的编码）。

• 支持数据持久化(定期将内存搬运到磁盘)

• 支持主从集群、分片集群（数据拆分）

• 支持多语言客户端

Redis的官方网站地址：RedisRedis Redis

1.3 Redis安装启动

Redis是基于C编写，所以需要先安装Redis所需的gcc依赖

yum install -y gcc 

如果有了就跳过

安装包上传到usr/local/src

tar -zxvf redis-6.2.6.tar.gz 

cd到redis的目录：

cd redis-6.2.6/

编译和运行

make && make install

默认启动：

需要一直挂着页面

redis-server

后台启动：

前提是必须修改配置文件(/usr/local/src/redis-6.2.6/redis.conf)

(1) 先备份一份

cp redis.conf redis.conf.bak

(2) 修改

vi redis.conf

eg:修改密码为1234

(3) 运行

cd /usr/local/src/redis-6.2.6
redis-server redis.conf

(4) 查看是否启动

ps -ef | grep redis

(5) 停止redis -9:强制但是不安全

kill -9 进程号

开机自启：

(1) 新建系统服务文件

vi /etc/systemd/system/redis.service

内容：

[Unit]
Description=redis-server
After=network.target

[Service]
Type=forking
ExecStart=/usr/local/bin/redis-server /usr/local/src/redis-6.2.6/redis.conf
PrivateTmp=true

[Install]
WantedBy=multi-user.target

(2) 重新加载系统服务

systemctl daemon-reload

(3)启动

systemctl start redis

(4)查看状态

systemctl status redis

(5) 设置开机自启

systemctl enable redis

1.4 Redis客户端

安装完成Redis，我们就可以操作Redis，实现数据的CRUD了。这需要用到Redis客户端，包括：

• 命令行客户端
• 图形化桌面客户端
• 编程客户端

1.4.1.Redis命令行客户端

Redis安装完成后就自带了命令行客户端：redis-cli，使用方式如下：

redis-cli [options] [commonds]

redis-cli -h 192.168.200.131 -p 6379 -a 1234

其中常见的options有：

• -h 127.0.0.1：指定要连接的redis节点的IP地址，默认是127.0.0.1
• -p 6379：指定要连接的redis节点的端口，默认是6379
• -a 123321：指定redis的访问密码

其中的commonds就是Redis的操作命令，例如：

• ping：与redis服务端做心跳测试，服务端正常会返回pong

不指定commond时，会进入redis-cli的交互控制台：

也可以先不写密码，后面来补充！

1.4.2.图形化桌面客户端

GitHub上的大神编写了Redis的图形化桌面客户端，地址：https://github.com/uglide/RedisDesktopManager

不过该仓库提供的是RedisDesktopManager的源码，并未提供windows安装包。

在下面这个仓库可以找到安装包：https://github.com/lework/RedisDesktopManager-Windows/releases

resp.exe

二.Redis命令

expire设置存活周期，ttl查看剩余时间，不设置expire的话ttl为-1

1.String

setex key expireTime value

由于Redis为NoSql,我们不知道value对应的属性的数据类型是什么

如果Value是一个Java对象，例如一个User对象，则可以将对象序列化为JSON字符串后存储：

一旦我们向redis采用这样的方式存储，那么在可视化界面中，redis会以层级结构来进行存储，形成类似于这样的结构，更加方便Redis获取数据**

2.Hash

本身Redis就是一个key-value的结构，而hash的value还是一个key-value的结构

支持对单个值进行修改

3.List

从左侧推

lpush users 1 2 3 

从右侧推

从左侧右侧弹出

lpop users 1
rpop users 1

阻塞弹出 blpop/brpop key second

blpop user1 100

4.Set

sadd s1 a b c

删除元素

 srem s1 a 

查看元素数量

scard s1 

多个集合之间的操作

sinsert s1 s2 //s1和s2的交集
sdiff s1 s2//s1和s2的差集
sunion s1 s2 //s1和s2的并集

sadd zs lisi wnagwu zhaoliu
sadd ls wangwu mazi ergou

scard zs
sinter zs ls
adiff zs ls
sunion zs ls
ismember zs lisi
ismember ls zhangsan
srem zs lisi

smembers zs
smembers ls

5.SortedSet

每个元素都带上分数，所以才能实现排序

zadd stus 85 jack 89 Lucy 82 Rose 95 Tom 78 Jerry 92 Amy 76 Miles 
zrem stus Tom 
zscore stus Amy
zrank stus Rose //升序 zrevrank stus Rose//降序
zcount stus 0 80
zincrby stus 2 Amy
zrange stus 0 2 //升序 zrevrange stus 0 2 //降序
zrangebyscore stus 0 80

三.Redis的java客户端

1.Jedis

在Redis官网中提供了各种语言的客户端，地址：https://redis.io/docs/clients/

基本用法

(1) 导入依赖

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>3.8.0</version>
</dependency>

(2) 建立连接

public class JedisTest {
    private Jedis jedis;

    @BeforeEach
    void setUp(){
        //1.建立连接
        jedis = new Jedis("192.168.200.130",6379);
        //2.设置密码
        jedis.auth("1234");
        //3.选择库
        jedis.select(0);
    }

    @Test
    void testString(){
        String result = jedis.set("name", "小明");
        System.out.println("result= " + result);

        String name = jedis.get("name");
        System.out.println("name= "+name);
    }

    @AfterEach
    void tearDown(){
        if(jedis!=null){
            jedis.close();
        }
    }


}

连接不上/报错 得使用设置密码

config set requirepass 12349

@Test
void testHash(){
    jedis.hset("user:1","name","jack");
    jedis.hset("user:1","age","21");

    Map<String, String> map = jedis.hgetAll("user:1");
    System.out.println(map);
}

Jedis连接池

建立Factory

public class JedisConnectFactory {
    private static final JedisPool jedisPool;

    static{
        //配置连接池
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(8);
        poolConfig.setMaxIdle(8);
        poolConfig.setMinIdle(0);
        poolConfig.setMaxWait(Duration.ofMillis(1000));
        jedisPool = new JedisPool(poolConfig,"192.168.200.130",6379,1000,"1234");
    }
    
    public static Jedis getJedis(){
        return jedisPool.getResource();
    }

}

代码说明：

• 1） JedisConnectionFacotry：工厂设计模式是实际开发中非常常用的一种设计模式，我们可以使用工厂，去降低代的耦合，比如Spring中的Bean的创建，就用到了工厂设计模式

• 2）静态代码块：随着类的加载而加载，确保只能执行一次，我们在加载当前工厂类的时候，就可以执行static的操作完成对 连接池的初始化

• 3）最后提供返回连接池中连接的方法.

改造原始代码

代码说明：

1.我们在完成了使用工厂设计模式来完成代码的编写后，我们可以通过工厂来获得连接，不用去new对象，减低耦合，且使用的还是连接池对象

2.当我们使用连接池后，当我们关闭连接其实并不是关闭，而是将Jedis归还给连接池

@BeforeEach
void setUp(){
    //1.建立连接
    //jedis = new Jedis("192.168.200.130",6379);
    //2.设置密码
    //jedis.auth("1234");
    jedis = JedisConnectFactory.getJedis();
    //3.选择库
    jedis.select(0);
}

····

 @AfterEach
    void tearDown() {
        if (jedis != null) {
            jedis.close();
        }
    }

2.SpringDataRedis

SpringData是Spring中数据操作的模块，包含对各种数据库的集成，其中对Redis的集成模块就叫做SpringDataRedis，官网地址：https://spring.io/projects/spring-data-redis

• 提供了对不同Redis客户端的整合（Lettuce和Jedis）
• 提供了RedisTemplate统一API来操作Redis
• 支持Redis的发布订阅模型
• 支持Redis哨兵和Redis集群
• 支持基于Lettuce的响应式编程(Lettuce之前实在es那里有)
• 支持基于JDK.JSON.字符串.Spring对象的数据序列化及反序列化
• 支持基于Redis的JDKCollection实现

SpringDataRedis中提供了RedisTemplate工具类，其中封装了各种对Redis的操作。并且将不同数据类型的操作API封装到了不同的类型中：

(1) 导入依赖

        <!--Redis依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <!--连接池依赖-->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>

(2) 配置文件

spring:
  redis:
    host: 192.168.200.130
    port: 6379
    password: 1234
    database: 0
    lettuce:
      pool:
        max-active: 8 #最大连接数
        max-idle: 8 #最大空闲连接
        min-idle: 0 #最小空闲连接
        max-wait: 100 #连接等待时间

(3) 自定义的RedisTemplate，注入到IOC中

先去(4) 写测试类测试一下就会发现问题

由于我们使用的是String-Object的话，你value的值如果传入一个字符串或者其他类型的话，并没有序列化，就会造成以下情况：get key

这是由于key和value会被当成对象，被redis底层的默认序列化方法：jdk序列化工具jdkSerializationRedisSerialliszer

而它采用的是objectOutputStream(把java对象转成字节)

先把这个key删除掉，用del

写一个RedisConfig，写bean: 拥有我们的序列化后的redisTemplate

@Configuration

public class RedisConfig {
    @Bean
    public RedisTemplate<String,Object> redisTemplate( RedisConnectionFactory factory){
        RedisTemplate<String,Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);

        // 设置key的序列化方式
        template.setKeySerializer(RedisSerializer.string());
        // 设置value的序列化方式
        template.setValueSerializer(RedisSerializer.json());
        // 设置hash的key的序列化方式
        template.setHashKeySerializer(RedisSerializer.string());
        // 设置hash的value的序列化方式
        template.setHashValueSerializer(RedisSerializer.json());

        template.afterPropertiesSet();
        return template;
    }
}

(4) 测试类

@SpringBootTest
public class RedisTest {
    @Autowired
    private RedisTemplate redisTemplate;


    @Test
    public void RedistestString(){
        // 写入一条String数据
        redisTemplate.opsForValue().set("name", "wxp");
        // 获取string数据
        Object name = redisTemplate.opsForValue().get("name");
        System.out.println("name = " + name);
    }
}

接下里试试实体类的序列化

@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
   private String name;
   private Integer age;
}

@Test
public void testSaveUser(){
   User user = new User();
   user.setName("阿廖莎");
   user.setAge(21);
   redisTemplate.opsForValue().set("user:100",user);
   System.out.println(redisTemplate.opsForValue().get("user:100"));
}

控制台打印：

.

.

发现这个json对象会将类的class写入，这个是为这个class进行反序列化的，但是会存在内存开销

现在只需要用成StringRedisTemplate即可

@Autowired
private StringRedisTemplate stringRedisTemplate;

//JSON工具类ObjectMapper,或者可以用fastjson:JSON.toJSONString(), JSON.parseObject()
private static final ObjectMapper mapper = new ObjectMapper();

@Test
    public void testSaveUser() throws JsonProcessingException {
        User user = new User();
        user.setName("阿廖莎");
        user.setAge(21);
        //手动序列化
        String json = mapper.writeValueAsString(user);

        stringRedisTemplate.opsForValue().set("user:100",json);
        //反序列化
        User user1 = mapper.readValue(stringRedisTemplate.opsForValue().get("user:100"), User.class);
        System.out.println("user1 = " + user1);
    }

.

.

接下来测试下哈希结构

string:

@Test
void testHash(){
    stringRedisTemplate.opsForHash().put("user:200","name","张三");
    stringRedisTemplate.opsForHash().put("user:200","age","21");

    Object name = stringRedisTemplate.opsForHash().get("user:200", "name");
    System.out.println("属性name: " + name);
    Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries("user:200");//获取全部的hashkey-hashvalue
    System.out.println("所有属性: " + entries);
}

.

.

实战篇====

四.Redis实战

一.短信登录

1.1导入hmdp项目

(1) 数据库

少数据的话，用navicat创建数据库hmdp，用idea连接mysql执行sql，用的还是本地的数据库

(2) 导入半成品hm-dianping

(3) 导入前端的话就导入nginx的文件夹，内部有hmdp的前端资源，我们导入后启动它即可

nginx目录下cmd输入start nginx.exe

访问：http://localhost:8080/

1.2session实现短信登录

(1) userService创建sendCode接口，实现它

@Override
public Result sendCode(String phone, HttpSession session) {
    //1.校验手机号:利用util下RegexUtils进行正则验证
    if(RegexUtils.isPhoneInvalid(phone)){
        return Result.fail("手机号格式不正确!");
    }
    //2.生成验证码:导入hutool依赖，内有RandomUtil
    String code = RandomUtil.randomNumbers(6);
    //3.保存验证码到session
    session.setAttribute("code",code);
    //4.发送验证码
    log.info("验证码为: " + code);
    log.debug("发送短信验证码成功!");

    return Result.ok();

}

(2) login功能

这个老师的写法也是存在问题：假如我先用自己的获取验证码，再换别人的手机号用我的验证码登录，也可以登录

@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
    //1.校验手机号
    String phone = loginForm.getPhone();
    if(RegexUtils.isPhoneInvalid(phone)){
     return Result.fail("手机号格式错误!");
    }
    //2.校验验证码
    Object cacheCode = session.getAttribute("code");
    String code = loginForm.getCode();
    if(code==null||!cacheCode.toString().equals(code)){
        //3.不一致，报错
        return Result.fail("验证码错误!");
    }
     //4.一致，根据手机号查询用户(需要写对应的单表查询方法:select * from tb_user where phone = #{phone})
    User user = query().eq("phone", phone).one();
    if(user==null){
        //5.注册用户
        user.setPhone(phone);
        user.setNickName("user_"+RandomUtil.randomString(10));
        //保存用户
        save(user);
    }
    //6.存入session
    session.setAttribute("user",user);
    return Result.ok();
}

(3) 登录验证功能

其实就是携带登录配置，这里用的是cookie,然后用拦截器进行拦截然后验证，但是一般我们用jwt令牌放入localstroagecookie

上面我们不能直接把user存入，而是要把保留一些不隐私的信息(UserDto)然后传入Session

a.

//6.存入session
session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));

b.拦截器

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.判断用户是否存在. 不存在:拦截;存在:放入ThreadLocal,放行(写了ThreadLocal的封装工具类UserHolder)
        if(user==null){
            response.setStatus(401);
            response.getWriter().write("用户未登录!");
            return false;
        }
        UserHolder.saveUser((UserDTO) user);
        return true;
    }

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

c.在MvcConfig内添加上我们的拦截器

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/upload/**",
                        "/voucher/**"
                );
    }
}

1.3集群的session共享问题

多台Tomcat不共享session存储空间，当请求切换到不同的tomcat服务时导致数据丢失的问题

所以我们把数据存入Redis，集群的Redis可以替代session

1.4基于Redis实现共享session登录

我们应该选择String类型存验证码即可，value:验证码，但是key要区分开来

​ 选择Hash存储用户信息，因为每个字段独立，比较好去DRUD,内存占用少，key用token即可(随机字符串)

之前的session的话，tomcat会自动把session的Id存入Cookie,每次请求都会携带Cookie,所以我们需要手动把token返回给客户端，每次请求客户端都会携带着token

基于上面的来进行修改

RedisConstants工具类存储key常量

public class RedisConstants {
    public static final String LOGIN_CODE_KEY = "login:code:";
    public static final Long LOGIN_CODE_TTL = 2L;
    public static final String LOGIN_USER_KEY = "login:token:";
    public static final Long LOGIN_USER_TTL = 36000L;

    public static final Long CACHE_NULL_TTL = 2L;

    public static final Long CACHE_SHOP_TTL = 30L;
    public static final String CACHE_SHOP_KEY = "cache:shop:";

    public static final String LOCK_SHOP_KEY = "lock:shop:";
    public static final Long LOCK_SHOP_TTL = 10L;

    public static final String SECKILL_STOCK_KEY = "seckill:stock:";
    public static final String BLOG_LIKED_KEY = "blog:liked:";
    public static final String FEED_KEY = "feed:";
    public static final String SHOP_GEO_KEY = "shop:geo:";
    public static final String USER_SIGN_KEY = "sign:";
}

sendCode做以下修改

//3.保存验证码到Redis
stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY +phone,code,LOGIN_CODE_TTL, TimeUnit.MINUTES);//有效期2mins

login

UserServiceImpl的sendCode和Login

@Service
@Slf4j
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements IUserService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Override
    public Result sendCode(String phone, HttpSession session) {
        //1.校验手机号:利用util下RegexUtils进行正则验证
        if(RegexUtils.isPhoneInvalid(phone)){
            return Result.fail("手机号格式不正确!");
        }
        //2.生成验证码:导入hutool依赖，内有RandomUtil
        String code = RandomUtil.randomNumbers(6);
        //3.保存验证码到Redis
        stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY +phone,code,LOGIN_CODE_TTL, TimeUnit.MINUTES);//有效期2mins
        //4.发送验证码
        log.info("验证码为: " + code);
        log.debug("发送短信验证码成功!");

        return Result.ok();

    }

    @Override
    public Result login(LoginFormDTO loginForm, HttpSession session) {
        //1.校验手机号
        String phone = loginForm.getPhone();
        if(RegexUtils.isPhoneInvalid(phone)){
         return Result.fail("手机号格式错误!");
        }
        //2.从Redis中获取验证码
        String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY+phone);
        String code = loginForm.getCode();
        if(cacheCode==null||!cacheCode.equals(code)){
            //3.不一致，报错
            return Result.fail("验证码错误!");
        }
         //4.一致，根据手机号查询用户(需要写对应的单表查询方法:select * from tb_user where phone = #{phone})
        User user = query().eq("phone", phone).one();
        if(user==null){
            //5.注册用户
            User newUser = new User();
            newUser.setPhone(phone);
            newUser.setNickName("user_"+RandomUtil.randomString(10));
            save(newUser);
            user = newUser;
        }
        //6.保存用户到Redis
            //(1)生成token
            String token = UUID.randomUUID().toString(true);//hutools
            //(2)User转为HashMap存储
        UserDTO userDTO = BeanUtil.copyProperties(user,UserDTO.class);
        HashMap<Object, Object> userMap = new HashMap<>();
        userMap.put("id", userDTO.getId().toString());
        userMap.put("nickName", userDTO.getNickName());
        userMap.put("icon", userDTO.getIcon());
            //(3)存储到Redis
            String tokenKey = LOGIN_USER_KEY + token;
            stringRedisTemplate.opsForHash().putAll(tokenKey,userMap);

            //(4) 设置有效期
            stringRedisTemplate.expire(tokenKey,LOGIN_USER_TTL,TimeUnit.MINUTES);

        return Result.ok(token);
    }
}

MvcConfig注入stringRedisTemplate,然后传给LoginInterceptor,因为LoginInterceptor不是bean不能用spring注入其他bean

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/upload/**",
                        "/voucher/**"
                );
    }
}

LoginInterceptor

public class LoginInterceptor implements HandlerInterceptor{

    private final StringRedisTemplate stringRedisTemplate;

    public LoginInterceptor(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)){
            //不存在，拦截 设置响应状态吗为401（未授权）
            response.setStatus(401);
            return false;
        }
        //2.基于token获取redis中用户
        String key=RedisConstants.LOGIN_USER_KEY + token;
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(key);
        //3.判断用户是否存在
        if (userMap.isEmpty()){
            //4.不存在则拦截，设置响应状态吗为401（未授权）
            response.setStatus(401);
            return false;
        }
        //5.将查询到的Hash数据转化为UserDTO对象
        UserDTO userDTO=new UserDTO();
        BeanUtil.fillBeanWithMap(userMap,userDTO, false);
        //6.保存用户信息到ThreadLocal
        UserHolder.saveUser(userDTO);
        //7.更新token的有效时间，只要用户还在访问我们就需要更新token的存活时间
        stringRedisTemplate.expire(key, RedisConstants.LOGIN_USER_TTL, TimeUnit.SECONDS);
        //8.放行
        return true;
    }


    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        //销毁，以免内存泄漏
        UserHolder.removeUser();
    }
}

用户请求进去拦截器，我们试着去获取请求头内的token，根据token去查询用户信息，判断是否拦截，保存在ThreadLocal,刷新token的有效期

但是，这个拦截器是拦截需要登录之后才需要进行请求的路径，那我如果一直在访问的是不需要拦截的页面的话，我还是会过期？这就不合理。所以我们需要在这个拦截器前面再加个拦截器，然后在新增拦截器上进行保存ThreadLocal和刷新有效期，不理解其他

其实就是对之前的拦截器进行功能拆分

MvcConfig

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/upload/**",
                        "/voucher/**"
                ).order(1);//RefreshTokenInterceptor 先于 LoginInterceptor 执行
        registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).order(0);//默认拦截所有请求
    }

}

RefreshTokenInterceptor

public class RefreshTokenInterceptor implements HandlerInterceptor{

    private final 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=RedisConstants.LOGIN_USER_KEY + token;
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(key);
        //3.判断用户是否存在
        if (userMap.isEmpty()){
            return true;
        }
        //5.将查询到的Hash数据转化为UserDTO对象
        UserDTO userDTO=new UserDTO();
        BeanUtil.fillBeanWithMap(userMap,userDTO, false);
        //6.保存用户信息到ThreadLocal
        UserHolder.saveUser(userDTO);
        //7.更新token的有效时间，只要用户还在访问我们就需要更新token的存活时间
        stringRedisTemplate.expire(key, RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);
        //8.放行
        return true;
    }


    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        //销毁，以免内存泄漏
        UserHolder.removeUser();
    }
}

LoginInterceptor

public class LoginInterceptor implements HandlerInterceptor{

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //判断是否需要拦截
        if(UserHolder.getUser()==null){
            response.setStatus(401);
            response.getWriter().write("用户未登录!");
            return false;
        }

        return true;
    }


    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        //销毁，以免内存泄漏
        UserHolder.removeUser();
    }
}

刷新以下首页

二.商户查询缓存

2.1什么是缓存

数据库发生改变，Redis还没及时更新，那么从缓存内取到的数据就会出错，就是数据一致性问题

2.2添加商户缓存

我们通过这个接口查询到的数据有很多，我们希望在此做个Redis缓存数据，提供查询速度

（1）service定义方法，实现它

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result querygetById(Long id) {
        //1.从Redis内查询商品缓存
        String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
        if(StrUtil.isNotBlank(shopJson)){
            //手动反序列化
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return Result.ok(shop);
        }
        //2.不存在就根据id查询数据库
        Shop shop = getById(id);
        if(shop==null){
            return Result.fail("商户不存在！");
        }
        //3.数据库数据写入Redis
        //手动序列化
        String shopStr = JSONUtil.toJsonStr(shop);
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,shopStr,CACHE_SHOP_TTL, TimeUnit.MINUTES);
        return Result.ok(shop);
    }
}

controller

@GetMapping("/{id}")
public Result queryShopById(@PathVariable("id") Long id) {
    return shopService.querygetById(id);
}

可以通过调试查看是否是从Redis内拿出来

时间变快了很多

2.3 添加商户类型缓存

作业：

@Service
public class ShopTypeServiceImpl extends ServiceImpl<ShopTypeMapper, ShopType> implements IShopTypeService {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result queryList() {
        //1.从Redis中查询
        String key = CACHE_SHOPTYPE_KEY;
        List<String> list = stringRedisTemplate.opsForList().range(key, 0, -1);
        if(!list.isEmpty()){
            //手动反序列化
            List<ShopType> typeList = new ArrayList<>();
            for (String s : list) {
                ShopType shopType = JSONUtil.toBean(s, ShopType.class);
                typeList.add(shopType);
            }
            return Result.ok(typeList);
        }

        //2.从数据库内查询
        List<ShopType> typeList = query().orderByAsc("sort").list();
        if(typeList.isEmpty()){
            return Result.fail("不存在该分类!");
        }
          //序列化
        for (ShopType shopType : typeList) {
            String s = JSONUtil.toJsonStr(shopType);
            list.add(s);
        }

        //3.存入缓存
        stringRedisTemplate.opsForList().rightPushAll(key,list);
        stringRedisTemplate.expire(key,CACHE_SHOPTYPE_TTL,TimeUnit.MINUTES);
        return Result.ok(list);
    }
}

2.4缓存更新策略

由于数据库的操作速度比操作缓存的速度慢，所以操作缓存的时候极低概率会被操作数据库的线程抢去cpu，反过来就会出现线程安全问题，所以采用先更新数据库再删除缓存

2.5实现商铺缓存和数据库的双写一致

ShopServiceImpl

@Override
public Result update(Shop shop) {
    if(shop.getId()==null){
        return Result.fail("店铺id不能为空!");
    }
    //1.更新数据库
    updateById(shop);
    //2.删除缓存
    String key = CACHE_SHOP_KEY + shop.getId();
    stringRedisTemplate.delete(key);
    return Result.ok();
}

ShopController

@PutMapping
public Result updateShop(@RequestBody Shop shop) {
    // 写入数据库
    return shopService.update(shop);
}

后面再访问的时候才会重新添加上缓存，这个之前就写过了

重新刷新

2.6缓存穿透的解决思路

避免数据库也查不到，还把null存入缓存，那么以后缓存就永远不生效

2.7解决商铺查询的缓存问题

2.8缓存雪崩

解决方案：

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

2.9缓存击穿

缓存击穿问题，也叫 热点 Key 问题；就是一个被 高并发访问 并且 缓存中业务较复杂的 Key 突然失效，大量的请求在极短的时间内一起请求这个 Key 并且都未命中，无数的请求访问在瞬间打到数据库上，给数据库带来巨大的冲击。

解决方案：

• 互斥锁：查询缓存未命中，获取互斥锁，获取到互斥锁的才能查询数据库重建缓存，将数据写入缓存中后，释放锁。
• 逻辑过期：查询缓存，发现逻辑时间已经过期，获取互斥锁，开启新线程；在新线程中查询数据库重建缓存，将数据写入缓存中后，释放锁；在释放锁之前，查询该数据时，都会将过期的数据返回。

基于互斥锁解决缓存击穿问题

核心：利用 Redis 的 setnx 方法来表示获取锁。该方法的含义是：如果 Redis 中没有这个 Key，则插入成功；如果有这个 Key，则插入失败。通过插入成功或失败来表示是否有线程插入 Key，插入成功的 Key 则认为是获取到锁的线程；释放锁就是将这个 Key 删除，因为删除 Key 以后其他线程才能再执行 setnx 方法。

stenx lock 1

/**
 * 获取互斥锁
 */
private boolean tryLock(String key) {
    Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, "1", TTL_TEN, TimeUnit.SECONDS);
    return BooleanUtil.isTrue(flag);
}

/**
 * 释放互斥锁
 */
private void unLock(String key) {
    redisTemplate.delete(key);
}

一次请求的过程

1. 请求打进来，先去 Redis 中查，未命中；

2. 获取互斥锁：将一个 Key 为 LOCK_SHOP_KEY + id 的数据写入 Redis 中，此时其他线程就无法拿到这个 Key，也就无法继续后续操作；

3. 获取失败就进行休眠，休眠结束后通过递归再次请求；

4. 获取成功，查询数据库、将需要查询的那个数据写入 Redis；

5. 最后，删除通过 setnx 创建的那个 Key。

需求：修改根据id查询店铺的业务（互斥锁方式解决）

/**互斥锁实现解决缓存击穿**/
public Shop queryWithMutex(Long id){
    //1.从Redis内查询商品缓存
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    if(StrUtil.isNotBlank(shopJson)){
        //手动反序列化
        return JSONUtil.toBean(shopJson, Shop.class);
    }
    //如果上面的判断不对，那么就是我们设置的""(有缓存"",证明数据库内肯定是没有的)或者null(没有缓存)
    //判断命中的是否时空值
    if(shopJson!=null){//
        return null;
    }

    //a.实现缓存重建
    //a.1 获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    Shop shop = null;
    try {
        boolean hasLock = tryLock(lockKey);
        //a.2 判断是否获取到，获取到:根据id查数据库 获取不到:休眠
        if(!hasLock){
            Thread.sleep(50);
            return queryWithMutex(id);
        }

        //2.不存在就根据id查询数据库
        shop = getById(id);
        //模拟重建的延时
        Thread.sleep(200);
        if(shop==null){
            stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
            return null;
        }
        //3.数据库数据写入Redis
        //手动序列化
        String shopStr = JSONUtil.toJsonStr(shop);
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,shopStr,CACHE_SHOP_TTL, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } finally {
        //释放互斥锁
        unlock(lockKey);
    }

    return shop;
}

@Override
public Result querygetById(Long id) {
    //缓存穿透
    //Shop shop = queryWithPassThrough(id);

    //互斥锁解决缓存击穿
    Shop shop = queryWithMutex(id);
    if(shop==null) return Result.fail("店铺不存在!");

    return Result.ok(shop);
}

利用postman测试多线程：先把这个key的缓存删除

访问1000次，数据库只查询一次，都可以200，说明互斥锁设置后效果成功

基于逻辑过期解决缓存击穿问题

private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);//开启10个线程

/**逻辑过期实现解决缓存击穿**/
public Shop queryWithLogical(Long id){
    //1.从Redis内查询商品缓存
    String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    //2.判断是否存在
    if(StrUtil.isBlank(shopJson)){
        return null;
    }
    //3.命中，需要先把json反序列化为对象
    RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
    JSONObject data = (JSONObject) redisData.getData();
    Shop shop = JSONUtil.toBean(data, Shop.class);

    //4.判断是否过期
    LocalDateTime expireTime = redisData.getExpireTime();
    if(expireTime.isAfter(LocalDateTime.now())){
        //未过期直接返回
        return shop;
    }
        //5.过期的话需要缓存重建
    //5.1 获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean hasLock = tryLock(lockKey);
    //5.2判断是否获取到，获取到:根据id查数据库 获取不到:休眠
   if(hasLock){
       //成功就开启独立线程，实现缓存重建, 这里的话用线程池
       CACHE_REBUILD_EXECUTOR.submit(()->{
           try {
               //重建缓存
               this.saveShop2Redis(id,20L);
           } catch (Exception e) {
               throw new RuntimeException(e);
           }finally {
               //释放锁
               unlock(lockKey);
           }

       });

   }

    return shop;
}

/**缓存重建方法**/
public void saveShop2Redis(Long id,Long expireSeconds) throws InterruptedException {
    //1.查询店铺信息
    Shop shop = getById(id);
    Thread.sleep(200);
    //2.封装逻辑过期时间
    RedisData redisData = new RedisData();
    redisData.setData(shop);
    redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));
    //3.写入Redis
    stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(redisData));
}

先用saveShop2Redis把热点key提前放入缓存，提升速度(设置了逻辑过期时间)

@Test
void testSaveShop() throws InterruptedException {
    shopService.saveShop2Redis(1L,10L);
}ja'v

把数据库修改一下

大约200ms后就会进行缓存重建

2.10 封装Redis工具类

方法3：

id2 -> getById(id2) 在java8是可以用 this::getById 代替的

/**解决缓存穿透**/
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(CACHE_SHOP_KEY + id);
    if(StrUtil.isNotBlank(json)){
        //手动反序列化
        return JSONUtil.toBean(json, type);
    }
    //如果上面的判断不对，那么就是我们设置的""(有缓存"",证明数据库内肯定是没有的)或者null(没有缓存)
    //判断命中的是否时空值
    if(json!=null){//
        return null;
    }
    //2.不存在就根据id查询数据库
    R r = dbFallBack.apply(id);//由于不知道这段逻辑，所以我们需要用户传进来函数逻辑
    if(r==null){
        stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
        return null;
    }
    //写入Redis
    this.set(key,r,time,unit);
    return r;
}

@Override
public Result querygetById(Long id) {
    //解决缓存穿透
    Shop shop = cacheClient.queryWithPassThrough(CACHE_SHOP_KEY,id,Shop.class,this::getById,CACHE_SHOP_TTL,TimeUnit.MINUTES);
    if(shop==null) return Result.fail("店铺不存在!");

    return Result.ok(shop);
}

测试：数据库访问一次，传入Redis为“”，多刷后不会查询到数据库

方法4：

private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);//10个线程的线程池

/**逻辑过期实现解决缓存击穿**/
public <R,ID> R queryWithLogical(String keyPrefix, ID id, Class<R> type, Function<ID,R> dbFallBack,Long time,TimeUnit unit) {
    String key = CACHE_SHOP_KEY + id;
    //1.从Redis内查询商品缓存
    String shopJson = stringRedisTemplate.opsForValue().get(key);
    //2.判断是否存在
    if (StrUtil.isBlank(shopJson)) {
        return null;
    }
    //3.命中，需要先把json反序列化为对象
    RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
    JSONObject data = (JSONObject) redisData.getData();
    R r = JSONUtil.toBean(data, type);

    //4.判断是否过期
    LocalDateTime expireTime = redisData.getExpireTime();
    if (expireTime.isAfter(LocalDateTime.now())) {
        //未过期直接返回
        return r;
    }
    //5.过期的话需要缓存重建
    //5.1 获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean hasLock = tryLock(lockKey);
    //5.2判断是否获取到，获取到:根据id查数据库 获取不到:休眠
    if (hasLock) {
        //成功就开启独立线程，实现缓存重建, 这里的话用线程池
        CACHE_REBUILD_EXECUTOR.submit(() -> {
            try {
                //重建缓存(查数据库+传入Redis)
                R r1 = dbFallBack.apply(id);
                this.setWithLogicalExpire(key,r1,time,unit);
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                //释放锁
                unlock(lockKey);
            }
        });
    }
    return r;
}

//设置锁
private boolean tryLock(String key){
    Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);//如果存在
    return BooleanUtil.isTrue(flag);

}
//修改锁
private void unlock(String key){
    stringRedisTemplate.delete(key);
}

@Override
public Result querygetById(Long id) {
    //解决缓存穿透
    //Shop shop = cacheClient.queryWithPassThrough(CACHE_SHOP_KEY,id,Shop.class,this::getById,CACHE_SHOP_TTL,TimeUnit.MINUTES);

    //互斥锁解决缓存击穿
    //Shop shop = queryWithMutex(id);

    //逻辑过期解决缓存击穿
    Shop shop = cacheClient.queryWithLogical(CACHE_SHOP_KEY,id,Shop.class,this::getById,20L,TimeUnit.SECONDS);//方便测试
    if(shop==null) return Result.fail("店铺不存在!");

    return Result.ok(shop);
}

测试：

我们先在Redis内插入逻辑过期时间的key

public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit){
    //逻辑过期时间 redisData有属性expireTime和Data,把value封装到里面就有了逻辑过期时间
    RedisData redisData = new RedisData();
    redisData.setData(value);
    redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
    //写入Redis
    stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(redisData));
}

@Test
void testSaveShop() throws InterruptedException {
    Shop shop = shopService.getById(1L);
    cacheClient.setWithLogicalExpire(CACHE_SHOP_KEY,shop,1L, TimeUnit.SECONDS);
}

postman发送多次请求

三.优惠卷秒杀

3.1全局唯一ID

tb_voucher_order表

在分布式系统下生成全局唯一ID的工具，满足 唯一性，高可用，高性能，递增性，安全性

这里利用的是Redis自增id策略

为了增加ID的安全性，不要直接使用Redis自增的数值，而是拼接一些其它信息：

一秒接收2^32，完全够用

ID的组成部分：符号位：1bit，永远为0

时间戳：31bit，以秒为单位，可以使用69年

序列号：32bit，秒内的计数器，支持每秒产生2^32个不同ID

@Component
public class RedisIdWorker {
    //到今年第一天的秒数
    private static final long BEGIN_TIMESTAMP = 1640995200L;

    //序列号的位数
    private static final long COUNT_TIMESTAMP = 32L;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    public long nextId(String keyPrefix){//不同业务

        //生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timeStamp = nowSecond-BEGIN_TIMESTAMP;

        //生成序列号
        String today = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        long count = stringRedisTemplate.opsForValue().increment("irc:" + keyPrefix + ":" + today);

        //拼接
        return timeStamp << COUNT_TIMESTAMP  |  count ;

    }

    public static void main(String[] args) {
        LocalDateTime time = LocalDateTime.of(2022, 1, 1, 0, 0, 0);
        long second = time.toEpochSecond(ZoneOffset.UTC);
        System.out.println("second = " + second);
    }
}

测试：

@Resource
private RedisIdWorker redisIdWorker;

private ExecutorService es = Executors.newFixedThreadPool(500);//线程池

@Test
void testRedisId() throws InterruptedException {
    //CountDownLatch大致的原理是将任务切分为N个，让N个子线程执行，并且有一个计数器也设置为N，哪个子线程完成了就N-1
    CountDownLatch latch = new CountDownLatch(300);

    Runnable task =()->{
        for(int i=0;i<100;i++){
            Long id = redisIdWorker.nextId("order");
            System.out.println("id = " + id);
        }
        latch.countDown();
    };
    Long begin = System.currentTimeMillis();
    for(int i=0;i<300;i++){
        es.submit(task);
    }
    latch.await();
    Long end = System.currentTimeMillis();
    System.out.println("time = " + (end - begin));
}

3.2添加优惠卷

类似拓展

{
    "shopId":1,
    "title":"100元代金券",
    "subTitle":"周一至周五均可使用",
    "rules":"全场通用\n无需预约\n可无限叠加\n不兑现、不找零\n仅限堂食",
    "payValue":8000,
    "actualValue":10000,
    "type":1,
    "stock":100,
    "beginTime":"2022-11-13T10:09:17",
    "endTime":"2022-11-13T22:10:17"
}

点击抢购：

3.3实现秒杀下单

@Override
@Transactional
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠卷
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    //2.判断秒杀是否开始，是否结束
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        return Result.fail("秒杀尚未开始!");
    }
    if(voucher.getEndTime().isBefore(LocalDateTime.now())){
        return Result.fail("秒杀已结束!");
    }
    //3.判断库存是否充足
    if(voucher.getStock()<=0){
        return Result.fail("优惠券库存不足!");
    }
    //4.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock -1")
            .eq("voucher_id", voucherId).update();
    //5.创建订单
    if(!success){
        return Result.fail("优惠券库存不足!");
    }
    //6.返回订单id
    VoucherOrder voucherOrder = new VoucherOrder();
    //6.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    //6.2用户id
    Long userId = UserHolder.getUser().getId();
    voucherOrder.setUserId(userId);
    //6.3代金券id
    voucherOrder.setVoucherId(voucherId);

    //7.订单写入数据库
    save(voucherOrder);
    
    //8.返回订单Id
    return Result.ok(orderId);
}

测试：

但是存在很多问题，多线程问题，单用户抢多张文图

3.4超卖现象

测试：

类似postman携带header

3.5超卖问题分析

悲观锁比较简单，直接加锁即可，乐观锁难在判断

第一种：携带另一个变量进行判断

第二种：用数据本身有没有变化进行判断

3.6乐观锁解决超卖问题

//4.扣减库存
boolean success = seckillVoucherService.update()
        .setSql("stock = stock -1")
        .eq("voucher_id", voucherId).eq("stock",voucher.getStock())//where id = ? and stock =? 添加了乐观锁
        .update();

结果是不会出现线程安全问题，但是优惠券会出现过剩的情况，这就是乐观所的弊端：例如多个线程一开始标识stock为100，然后有个线程把stock减一了，其他那些线程就会返回错误

改进：

不去判断库存是否改变，判断库存>0即可

//4.扣减库存
boolean success = seckillVoucherService.update()
        .setSql("stock = stock -1")
        .eq("voucher_id", voucherId).gt("stock",0)//where id = ? and stock >0 添加了乐观锁
        .update();

3.7实现一人一单功能

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

但是你要加锁，不让遇到多线程还是会下多个单，所以需要改进：没法判断这个数据是否修改过，因为一开始不存在，不能用乐观锁，所以只能用乐观锁

问题：能否用乐观锁执行？

不能，原因是乐观锁只能操作单个变量，而创建订单需要操作数据库

(1) 对实现类进行修改

@Override
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠卷
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    //2.判断秒杀是否开始，是否结束
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        return Result.fail("秒杀尚未开始!");
    }
    if(voucher.getEndTime().isBefore(LocalDateTime.now())){
        return Result.fail("秒杀已结束!");
    }
    //3.判断库存是否充足
    if(voucher.getStock()<=0){
        return Result.fail("优惠券库存不足!");
    }

    Long userId = UserHolder.getUser().getId();
    synchronized (userId.toString().intern()) {//userId一样的持有同一把锁，最好不要放在整个方法上,intern:去字符串常量池找相同字符串
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();//获得代理对象
        return proxy.createVoucherOrder(voucherId);//默认是this,我们要实现事务需要proxy
    }//先获取锁，然后再进入方法，确保我的前一个订单会添加上,能先提交事务再释放锁
}

@Transactional
public Result createVoucherOrder(Long voucherId){
    //查询订单看看是否存在
    Long userId = UserHolder.getUser().getId();

    if (query().eq("user_id",userId).eq("voucher_id",voucherId).count()>0) {
        return Result.fail("用户已经购买过一次!");
    }

    //4.扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock -1")
            .eq("voucher_id", voucherId).gt("stock",0)//where id = ? and stock >0 添加了乐观锁
            .update();
    //5.创建订单
    if(!success){
        return Result.fail("优惠券库存不足!");
    }

    //6.返回订单id
    VoucherOrder voucherOrder = new VoucherOrder();
    //6.1订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    //6.2用户id
    //Long userId = UserHolder.getUser().getId();
    voucherOrder.setUserId(userId);
    //6.3代金券id
    voucherOrder.setVoucherId(voucherId);

    //7.订单写入数据库
    save(voucherOrder);
    //8.返回订单Id
    return Result.ok(orderId);
}

(2) 导入依赖

<dependency>
    <groupId>org.aspectj</groupId>
    <artifactId>aspectjweaver</artifactId>
</dependency>

(3) 启动类上加 @EnableAspectJAutoProxy(exposeProxy = true) 暴露出代理对象

测试：

3.8集群下的线程并发安全问题

Postman发送两个请求

两个id相同的进入到锁里面，证明没有被锁住，那就会生成2个订单

因为相当于我们开了两个jvm,所以有两个锁监视器，这样就出现并行的2个线程执行，这样就出现了线程安全问题

3.9分布式锁-原理

不去使用jvm内部的锁监视器，我们要在外部开一个锁监视器，让它监视所有的线程

多进程可见，互斥，高可用，高性能，安全性，. . . . .

常见的分布式锁

• MySQL：MySQL 本身带有锁机制，但是由于 MySQL 性能一般，所以采用分布式锁的情况下，使用 MySQL 作为分布式锁比较少见。
• Redis：Redis 作为分布式锁比较常见，利用 setnx 方法，如果 Key 插入成功，则表示获取到锁，插入失败则表示无法获取到锁。
• Zookeeper：Zookeeper 也是企业级开发中比较好的一个实现分布式锁的方案。

3.10分布式锁-实现思路

问题: 如何做到添加锁操作和释放锁操作必须具备同成功同失败？

set操作和expire写在同个语句即可：set lock thread1 ex 10 nx (nx表示存在的时候才可以set)

3.11实现Redis分布式锁版本1

ILock接口

public interface ILock {
    /**
     * 尝试获取锁
     * @Param timeoutSec 锁的持有时间
     * @return true:获取成功 false:获取失败
     */
    boolean tryLock(long timeoutSec);

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

SimpleRedisLock

public class SimpleRedisLock implements ILock{
    private String name;
    private StringRedisTemplate stringRedisTemplate;
    private static final String KEY_PREFIX = "lock:";

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


    @Override
    public boolean tryLock(long timeoutSec) {
        String key = KEY_PREFIX + name;
        //value的话一般设置为哪个线程持有该锁即可
        //获取线程标识
        long threadId = Thread.currentThread().getId();
        //获取锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, threadId+"", timeoutSec, TimeUnit.SECONDS);
        //最好不要直接return success，因为我们返回的是boolean类型，现在得到的是Boolean的结果，就会进行自动装箱，如果success为null,就会出现空指针异常
        return Boolean.TRUE.equals(success);//null的话也是返回false
    }

    @Override
    public void unlock() {
        String key = KEY_PREFIX + name;
        stringRedisTemplate.delete(key);
    }
}

在VoucherOrderServiceImpl内使用：

Long userId = UserHolder.getUser().getId();
//创建锁对象
SimpleRedisLock lock = new SimpleRedisLock("order" + userId,stringRedisTemplate);
//获取锁
boolean hasLock = lock.tryLock(1200);
if(!hasLock){
    //获取锁失败: return fail 或者 retry 这里业务要求是返回失败
    return Result.fail("请勿重复下单!");
}

try {
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();//获得代理对象
    return proxy.createVoucherOrder(voucherId);//默认是this,我们要实现事务需要proxy
} catch (IllegalStateException e) {
    throw new RuntimeException(e);
} finally {
    lock.unlock();
}

分布式锁的测试：

证明现在只有一个线程获取锁成功了，8081线程持有了分布式锁，而8082没有，查看结果：

3.12 Redis分布式锁误删问题

解决：在释放锁的时候判断锁的标识是否一致，Redis锁的标识一般是指value的区分，这里一般我们标识的是线程id,比如Thread1,Thread2…

3.13解决Redis分布式锁误删问题

这里的线程标识，我们之前用的是线程id进行标识，但是如果放到集群线程下，多个jvm可能会出现同个线程id的线程，这样会引发线程安全问题，所以这里要用ThreadID + UUID

@Override
public boolean tryLock(long timeoutSec) {
    String key = KEY_PREFIX + name;
    //value的话一般设置为哪个线程持有该锁即可
    //获取线程标识
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    //获取锁
    Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, threadId, timeoutSec, TimeUnit.SECONDS);
    //最好不要直接return success，因为我们返回的是boolean类型，现在得到的是Boolean的结果，就会进行自动装箱，如果success为null,就会出现空指针异常
    return Boolean.TRUE.equals(success);//null的话也是返回false
}

@Override
public void unlock() {
    String key = KEY_PREFIX + name;
    //获取线程标识
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    //获取锁中的标识
    if (stringRedisTemplate.opsForValue().get(key).equals(threadId)) {
        stringRedisTemplate.delete(key);
    }
}

测试:线程1获取锁，然后我把锁的标识删除，让线程2获取锁

然后线程1就不会释放锁，线程2后面会释放锁

3.14 分布式锁的原子性问题

判断锁标识和释放锁是两个操作，这里有原子性问题

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

Lua语言调用Redis:

eg:

-- 锁的key
-- local key = KEYS[1]
-- 当前线程标识
-- local threadId = ARGV[1]
-- 获取锁中的线程标识
local id = redis.call('get',KEYS[1])
-- 比较
if(id == ARGV[1]) then return redis.call('del',KEYS[1])
end return 0    

简化：

-- 比较
if(redis.call('get',KEYS[1]) == ARGV[1]) then return redis.call('del',KEYS[1])
end return 0

接下来就是java去调用lua,lua执行Redis

3.16java调用lua脚本改造分布式锁

unlock操作：

private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

static{//写成静态代码块，类加载就可以完成初始定义，就不用每次释放锁都去加载这个，性能提高咯
    UNLOCK_SCRIPT = new DefaultRedisScript<>();
    UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));//设置脚本位置
    UNLOCK_SCRIPT.setResultType(Long.class);
}
    public void unlock(){
        //调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId()
        );
    }

Lua脚本：

-- 比较
if(redis.call('get',KEYS[1]) == ARGV[1]) then return redis.call('del',KEYS[1])
end return 0

模拟超时释放锁： 8082拿到锁，把锁删除，8081也拿到锁，8082进行unlock的时候不会把8081的锁删除，8081unlock删除自己的锁

执行8082unlock后Redis中lock还在，8081执行unlock后就删除了

3.17分布式锁-Redission简介

3.18 Redisson快速入门

1.导入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

2.RedissonConfig,注入一个配置好的RedissonClient

@Configuration
public class RedissionConfig {
    @Bean
    public RedissonClient redissionClient(){
        //配置
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.200.131:6379").setPassword("1234");
        //创建RedissonClient对象
        return Redisson.create(config);
    }
}

3.在我们的业务实现类中使用RedissionClient

@Autowired
private RedissonClient redissonClient;

业务方法{
//···前面流程
 Long userId = UserHolder.getUser().getId();
        //创建锁对象
        //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId,stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //获取锁
        boolean hasLock = lock.tryLock( );
        if(!hasLock){
            //获取锁失败: return fail 或者 retry 这里业务要求是返回失败
            return Result.fail("请勿重复下单!");
        }

        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();//获得代理对象
            return proxy.createVoucherOrder(voucherId);//默认是this,我们要实现事务需要proxy
        } catch (IllegalStateException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
}

测试单线程：

测试多线程：记得登录验证token是否失效，失效了重新加

3.19 Redisson的可重入锁原理

为什么要引入可重入锁这种机制?

我们知道“对象一把锁，多个对象多把锁”，可重入锁的概念就是：自己可以获取自己的内部锁。

假如有一个线程 T 获得了对象 A 的锁，那么该线程 T 如果在未释放前再次请求该对象的锁时，如果没有可重入锁的机制，是不会获取到锁的，这样的话就会出现死锁的情况

这里除了记录key和ThreaId外，还需有记录重入次数，所以我们需要用hash, 每次的thead一样时，次数+1；

直到次数为0的时候才可以删除锁

nx ：判断锁是否存在

ex： 设置过期时间

但是Redis的hash结构没有nx这个命令，所以我们只能先判是否存在(exist)，再设置过期时间

@Slf4j
@SpringBootTest
public class RedissonTest {

    @Resource
    private RedissonClient redissonClient;

    private RLock lock;

    @BeforeEach
        // 创建 Lock 实例（可重入）
    void setUp() {
        lock = redissonClient.getLock("order");
    }

    @Test
    void methodOne() throws InterruptedException {
        boolean isLocked = lock.tryLock();
        log.info(lock.getName());
        if (!isLocked) {
            log.error("Fail To Get Lock~1");
            return;
        }
        try {
            log.info("Get Lock Successfully~1");
            methodTwo();
        } finally {
            log.info("Release Lock~1");
            lock.unlock();
        }
    }

    @Test
    void methodTwo() throws InterruptedException {
        boolean isLocked = lock.tryLock();
        if (!isLocked) {
            log.error("Fail To Get Lock!~2");
            return;
        }
        try {
            log.info("Get Lock Successfully!~2");
        } finally {
            log.info("Release Lock!~2");
            lock.unlock();
        }
    }
}

method1取到锁，redis内也存储lock:order:

3.20 Redisson的锁重试和WatchDog机制

锁重试：获取锁失败后重新获取

tryLock(waitTime,leaseTime,TimeUnit)

waitTime:获取锁的等待时长，获取锁失败后等待waitTime再去获取锁

leaseTime: 锁自动失效时间，这里测试锁重试不需要用到

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-89rNAz0V-1669222806273)(http://itsawaysu.oss-cn-shanghai.aliyuncs.com/note/Redisson%23tryLock%20%E9%94%81%E9%87%8D%E8%AF%95.png)]

上面不给leaseTime的话30s其实就是Lock WatchdogTimeOut，给到他，最后给到intrtnslLockleaseTime(用去watchdog续约)

tryLockInnerAsync其实就是执行lua脚本 state就是hashValue，就是次数

后面的ttl的结果只有nil或者30s,nil就是获取锁成功

WatchDog-----超时释放

对抢锁过程进行监听，抢锁完毕后，scheduleExpirationRenewal(threadId) 方法会被调用来对锁的过期时间进行续约，在后台开启一个线程，进行续约逻辑，也就是看门狗线程。

// 续约逻辑
commandExecutor.getConnectionManager().newTimeout(new TimerTask() {... }, 锁失效时间 / 3, TimeUnit.MILLISECONDS);

Method(new TimerTask(){}, 参数2, 参数3)

通过参数2、参数3 去描述，什么时候做参数1 的事情。

• 锁的失效时间为 30s，10s 后这个 TimerTask 就会被触发，于是进行续约，将其续约为 30s；
• 若操作成功，则递归调用自己，重新设置一个 TimerTask 并且在 10s 后触发；循环往复，不停的续约。

剩下的就是主从一致性问题

3.21 Redisson的multiiLock原理

三个结点：创建对应的三个client

//TODO

3.22 Redis优化秒杀

一人一单

key value1 value2 value3 … (value不重复) ----> set

3.23 基于Redis完成秒杀资格判断

完成需求1，2

1.新增优惠券的同时加入到Redis

测试：

2.编写lua,基于lua完成一人一单

-- 1.参数列表
-- 1.1 优惠券id
local voucherId = ARGV[1]
-- 1.2用户id
local userId = ARGV[2]

-- 2.数据key
-- 2.1 库存key
local stockKey = 'seckill:stock:'..voucherId
-- 2.2 订单key
local orderKey = 'seckill:order:'..voucherId

-- 3.脚本业务
-- 3.1判断库存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0)
    then return 1
end
-- 3.2判断用户是否下单 sismember orderKey userId
if(redis.call('sismember',orderKey,userId)==1)
    then return 2
end
-- 3.3扣库存 incrby stockKey -1
redis.call('incrby',stockKey,-1)
-- 3.4下单 sadd orderKey userId
redis.call('sadd',orderKey,userId)
return 0

测试第一次领券

第二次领券：

3.24 基于阻塞队列实现秒杀异步下单

之前的代码逻辑：

完善后的完整代码：

 	private static final DefaultRedisScript<Long> SECKI_SCRIPT;

    static{//写成静态代码块，类加载就可以完成初始定义，就不用每次释放锁都去加载这个，性能提高咯
        SECKI_SCRIPT = new DefaultRedisScript<>();
        SECKI_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//设置脚本位置
        SECKI_SCRIPT.setResultType(Long.class);
    }

    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);//创建阻塞队列

    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//创建线程池

// 判断库存和进行一人一单判断后将信息放入阻塞队列
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠卷
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2.判断秒杀是否开始，是否结束
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀尚未开始!");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已结束!");
        }
        //3.判断库存是否充足
        if(voucher.getStock()<=0){
            return Result.fail("优惠券库存不足!");
        }
    //获取当前用户
    Long userId = UserHolder.getUser().getId();
    //1.执行Lua脚本
    Long result = stringRedisTemplate.execute(
            SECKI_SCRIPT,
            Collections.emptyList(),//空List
            voucherId.toString(), userId.toString()
    );
    //2.判断结果是否0 是0就是成功，可下单，下单信息保存到阻塞队列
    if(result!=0){
        return Result.fail(result==1?"库存不足!":"不能重复下单!");
    }
    //生成订单id
    long orderId = redisIdWorker.nextId("order");
    //创建订单数据
    VoucherOrder voucherOrder = new VoucherOrder();
    voucherOrder.setUserId(userId);
    voucherOrder.setId(orderId);
    voucherOrder.setVoucherId(voucherId);
    //放入阻塞队列
    orderTasks.add(voucherOrder);
    //获取代理对象
    proxy = (IVoucherOrderService) AopContext.currentProxy();//获得代理对象
    //3.返回订单id
    return Result.ok(orderId);

}

// 类加载后就持续从阻塞队列出取出订单信息

@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();
                    handleVoucherOrder(voucherOrder);
                    //2.创建订单
                } catch (Exception e) {
                    log.error("处理订单异常:",e);
                }
            }
        }
    }

//异步下单
private void handleVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();//由于多线程，所以不能直接去ThreadLocal取
        //创建锁对象
        //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId,stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //获取锁
        boolean hasLock = lock.tryLock( );
        if(!hasLock){
            //获取锁失败
            log.error("不允许重复下单!");
            return;
        }

        try {
            //代理对象改成全局变量
            proxy.createVoucherOrder(voucherOrder);//默认是this,我们要实现事务需要proxy
        } catch (IllegalStateException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }

    }

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder){
        //查询订单看看是否存在
        Long userId = UserHolder.getUser().getId();

        if (query().eq("user_id",userId).eq("voucher_id", voucherOrder.getUserId()).count()>0) {
            log.error("用户已经购买过一次!");
            return;
        }

        //4.扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock -1")
                .eq("voucher_id", voucherOrder.getVoucherId()).gt("stock",0)//where id = ? and stock >0 添加了乐观锁
                .update();

        if(!success){
            log.error("优惠券库存不足!");
            return;
        }

        //7.订单写入数据库
        save(voucherOrder);
    }

3.25 认识消息队列

上一节我们基于jvm的阻塞队列进行秒杀存在2个问题：

1. jvm的内存限制问题

2. 数据安全问题：jvm的内存数据没有持久化，每当服务器重启或者宕机或者从阻塞队列取的时候遇到异常，数据都会丢失

   解决方法： 消息队列

Redis提供了三种不同方式来实现消息队列

1.list结构：模拟消息队列

2.Pubsub：基本的点对点模型

3.Stream ：比较完善的消息队列模型

3.26 基于list实现的消息队列

xshell开两个一样的会话

3.27 基于PubSub实现的消息队列

xshell开三个一样的会话

两个消费者：

生产者发布消息：

没人接收你的消息，消息就没了

3.28 基于Stream实现的消息队列

//获取最新消息
xread count 1 streams s1 $
//阻塞获取最新消息
xread count 1 block 0 streams s1 $ //block后面的数是阻塞毫秒数，0的话是永久阻塞

读最新数据会出现漏读现象：一下子发了5条最新消息，只读一条，其他4条漏读

3.29 Stream的消费者组模式

解决数据漏读的问题

不用自己去创建消费者，监听消息的时候然后发现无该消费者，则会自动创建

创建的时候ID注意如果想要之前的数据就从0开始，不想要就从$开始

3.30 基于stream消息队列实现异步秒杀下单

1.控制台stream类型创建消息队列

xgroup create stream.orders g1 0 mkstream

2.1 修改Lua脚本

-- 1.参数列表
-- 1.1 优惠券id
local voucherId = ARGV[1]
-- 1.2用户id
local userId = ARGV[2]
-- 1.3订单Id
local orderId = ARGV[3] -- 新增

-- 2.数据key
-- 2.1 库存key
local stockKey = 'seckill:stock:'..voucherId
-- 2.2 订单key
local orderKey = 'seckill:order:'..voucherId

-- 3.脚本业务
-- 3.1判断库存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0)
    then return 1
end
-- 3.2判断用户是否下单 sismember orderKey userId
if(redis.call('sismember',orderKey,userId)==1)
    then return 2
end
-- 3.3扣库存 incrby stockKey -1
redis.call('incrby',stockKey,-1)
-- 3.4下单 sadd orderKey userId
redis.call('sadd',orderKey,userId)

--4.发送消息到消息队列， xadd stream.order * k1 v1 k2 v2
redis.call('xadd','stream.order','*','userId',userId,'voucherId',voucherId,'id',orderId) --新增
return 0

2.2 修改执行lua脚本

private IVoucherOrderService proxy;
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠卷
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    //2.判断秒杀是否开始，是否结束
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        return Result.fail("秒杀尚未开始!");
    }
    if(voucher.getEndTime().isBefore(LocalDateTime.now())){
        return Result.fail("秒杀已结束!");
    }
    //3.判断库存是否充足
    if(voucher.getStock()<=0){
        return Result.fail("优惠券库存不足!");
    }
    //获取当前用户
    Long userId = UserHolder.getUser().getId();
    //生成订单id
    long orderId = redisIdWorker.nextId("order");
    //1.执行Lua脚本
    Long result = stringRedisTemplate.execute(
            SECKI_SCRIPT,
            Collections.emptyList(),//空List
            voucherId.toString(), userId.toString(),String.valueOf(orderId)
    );
    //2.判断结果是否0 是0就是成功，可下单，下单信息保存到阻塞队列
    if(result!=0){
        return Result.fail(result==1?"库存不足!":"不能重复下单!");
    }
    //获取代理对象
    proxy = (IVoucherOrderService) AopContext.currentProxy();//获得代理对象
    //3.返回订单id
    return Result.ok(orderId);

}

3.获取消息队列中的消息

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>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                );
                //2.判断获取是否成功
                //3.失败就再循环
                if(list==null||list.isEmpty()){
                    continue;
                }
                //4.成功就创建订单且ACK确认
                //解析消息
                MapRecord<String, Object, Object> record = list.get(0);//消息id,key,value
                //取出每个消息
                Map<Object, Object> values = record.getValue();
                //转为VoucherOrder实体类
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                //创建订单
                handleVoucherOrder(voucherOrder);
                //ACK确认
                stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());
            } catch (Exception e) {
                log.error("处理订单异常:",e);
                //5.出现异常，从pendingList中取出数据后重新操作
                handlePendingList();
            }
        }
    }
}

private void handlePendingList() {
    while (true) {
        try {
            // 1. 获取 pending-list 中的订单信息
            // XREAD GROUP orderGroup consumerOne COUNT 1 STREAM stream.orders 0
            List<MapRecord<String, Object, Object>> readingList = stringRedisTemplate.opsForStream().read(
                    Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1),
                    StreamOffset.create("stream.orders", ReadOffset.from("0"))
            );

            // 2. 判断消息是否获取成功
            if (readingList.isEmpty() || readingList == null) {
                // 获取失败 pending-list 中没有异常消息，结束循环
                break;
            }

            // 3. 解析消息中的订单信息并下单
            MapRecord<String, Object, Object> record = readingList.get(0);
            Map<Object, Object> recordValue = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(recordValue, new VoucherOrder(), true);
            handleVoucherOrder(voucherOrder);

            // 4. XACK
            stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
        } catch (Exception e) {
            log.error("订单处理异常（pending-list）", e);
            try {
                // 稍微休眠一下再进行循环
                Thread.sleep(10);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
}