Redis!I love Redis ? - 亥子曜工厂

初始Redis

什么是NoSql

原子性，一致性，隔离性，持久性

结构化与非结构化
关联与非关联
sql查询与非sql
是否支持事:ACID与BASE
存储方式
扩展性:垂直与水平(是否支持分布式)

认识Redis

k-v数据库：支持多种类型
单线程，具有原子性
低延迟，速度快(基于内存，IO多路复用，良好的编码)
支持数据持久化
支持主从集群，分布集群
安装Redis
安装路径/usr/local/bin
启动 redis-server

redis默认是前台启动，我们要修改为后台启动

/usr/local/src/redis-版本号/redis.conf

修改配置文件

  bind 127.0.0.1  -> 0.0.0.0
  daemonize no -> yes     #守护进程
  requirepass  123456 #设置密码
  port 6379
  dir .
  databases 1 #默认是16个库
  maxmemory 512mb
  logfile "redis.log"

redis-server redis.conf #这里要写好路径位置
ps -ef grep redis

开机启动脚本

写脚本

  [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

重启服务：sustemctl daemon-reload
systemctl start redis
systemctl enable redis

Redis命令行客户端

redis-cli [options] [commonds]
redis-cli -h 127.0.0.1 -p 6379 -a 密码

redis-cli
AUTH 密码

Redis常见命令

Redis数据结构

key一般是String value多样

基础类型
- String
- Hash
- List
- Set
- SortedSet
特殊类型
- GEO
- BitMap
- HyperLog

命令

help
keys
- keys [pattern]不建议使用，因为是模糊查询，速度慢，容易堵塞
DEL
- DEL k1 k2 k4
EXISTS
- exist key
EXPIRE
- expire key time
- expire age 20
TTL
- TTL key
- 值为-1：永久有效
- 值为-2：死掉了

字符串String

String虽然叫做字符串，但是对于存储数字，如果我用字符数组去存，占用空间肯定很大，所以底层是编码为数字再到二进制数据存储数字，所以String底层不一定是String，你甚至可以把图片存为”String”

最大空间不能超过512M

msg
num
score

命令:

set key value ：设置或更改
get key
mset k1 v1 k2 v2 ：批量set
mget k1 k2 :
incr key : 让value自增1
incrby key increament : 自定义自增
- incrby ket -1 :实现自减 (虽然说有自减这个操作，但是我们没有定义)
incrbyfloat: 浮点数自增，必须定义步长
setnx key val : 只增加，如果有就不管了
- set key val nx :一样的用法
setex key seconds value
- set key value ex seconds
key层级结构项目：业务：类型：id
Hash类型
HSET:
- hset key field value
HGET
- hget key field
HMSET:
- hmset key f1 v1 f2 v2
HMGET
- hmget key f1 f2
HGETALL (输出f1,v1,f2,v2)
- hgetall key
HKEYS
HVALS
HINCRBY:
HSETNX: 判断field是否存在
- hsetnx key field value

List类型

底层可以看做双向链表

有序
元素可重复
插入删除块
查询速度一般
LPUSH key element [element]
LPOP key
RPUSH key element
LPOP key
LRANGE key start end
- lrange key 1 2
BLPOP 和 BRPOP ：与LOPOP和RPOP相似，但是可以在没有元素的时候等待指定时间，而不是直接返回nil
- BLPOP key 100 等100s

Set

无序
不可重复
查找快
支持集合操作
SADD key member
SREM key member
SCARD key: 返回总数
SISMEMBER key member
SMEMBERS: 获取所有元素
SINTER k1 k2 ：交集
SDIFF k1 k2 : 差集 k1差k2
SUNION k1 k2 : 并集

SortedSet

可排序
元素不重复
查询速度快
底层是跳表和hash表
经常用来实现排行榜
ZADD key score member
ZREM key member
ZSCORE key member
ZRANK key member
ZCARD key
ZCOUNT key min max :根据score获得count,两端都是闭合[0,80]
ZINCRBY key increment member
ZRANGE key min max : 根据排名获取元素
ZRANGEBYSCORE key min max :
ZDIFF,ZINTER,ZUNION
使用ZREVRANGE可使用降序排序，都可以用ZREV

客户端

jedis
- 以命令作为方法名，简单实用
- 线程不安全，多线程环境需要连接池
lettuce
- netty实现的，线程安全
- 支持哨兵，集群，管道
Redisson
- 基于Redis实现的分布式，
Spring Data Redis : 底层可有jedis和lettuce实现

//注意事项，记得放开redis的端口，要么就直接关闭防火墙

Jedis入门

@SpringBootTest
class RedisApplicationTests {
    private Jedis jedis;
    @BeforeEach
    void connect(){
        jedis = new Jedis("192.168.200.130",6379);
        jedis.auth("123456");
        jedis.select(1);
    }
    @AfterEach
    void disconnect(){
        if(jedis != null){
            jedis.close();
        }
    }

    @Test
    void contextLoads() {
        jedis.set("hello", "world");
        String hello = jedis.get("hello");
        System.out.println(hello);
    }
}

jedis线程池

alt text

Spring Data Redis

提供对不同客户端的整合
提供 RedisTemplate来进行
支持Redis的发布订阅模型
支持Redis哨兵和集群
支持Lettuce的响应式编程
支持序列化和反序列化
支持基于Redis的JDKCollection实现

依赖

Redis
连接池 commons
配置redis
- spring.data.redis.host=192.168.200.130
- spring.data.redis.password=123456
- 其他应该都默认配置好了

@SpringBootTest
public class Test2 {
    @Autowired
    private RedisTemplate redisTemplate;
    @Test
    public void test() {
        redisTemplate.opsForValue().set("name", "hzy");
        System.out.println(redisTemplate.opsForValue().get("name"));
    }
}

//但是打开redis，存的竟然是这个
//\xAC\xED\x00\x05t\x00\x03name
//\xAC\xED\x00\x05t\x00\x03hzy
//底层默认序列化了
//这样肯定有很多弊端，比如占用内存高，比如难看
//解决方法就是，我们自己配
@Configuration
public class RedisConfig {
 
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory){
        // 创建RedisTemplate对象
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        // 设置连接工厂
        template.setConnectionFactory(connectionFactory);
        // 创建JSON序列化工具
        GenericJackson2JsonRedisSerializer jsonRedisSerializer = 
            							new GenericJackson2JsonRedisSerializer();
        // 设置Key的序列化
        template.setKeySerializer(RedisSerializer.string());
        template.setHashKeySerializer(RedisSerializer.string());
        // 设置Value的序列化
        template.setValueSerializer(jsonRedisSerializer);
        template.setHashValueSerializer(jsonRedisSerializer);
        // 返回
        return template;
    }
}

    public void test() {
        redisTemplate.opsForValue().set("name",new Person("hzy",88));
        System.out.println(redisTemplate.opsForValue().get("name"));
    }

/*
{
  "@class": "com.xjtuhelper.redis.bean.Person",
  "name": "hzy",
  "age": 88
}
*/

StringRedisTemplate

“@class”: “com.xjtuhelper.redis.bean.Person” 序列化和反序列化都得靠这个，但是对于redis而言，我们放在内存中，怎么能让这么一个东西占用空间呢，所以我们统一用String存，自己手动序列化

我们直接使用StringRedisTemplate

private ObjectMapper objectMapper=new ObjectMapper();
@Test
public void test() throws JsonProcessingException {
    Person hzy = new Person("hzy", 111);
    String json = objectMapper.writeValueAsString(hzy);
    stringRedisTemplate.opsForValue().set("hzy", json);
    String s = stringRedisTemplate.opsForValue().get("hzy");
    Person person = objectMapper.readValue(s, Person.class);
    System.out.println(person);
}

Redis实战–黑马点评

短信登录
查询缓存
达人探店
优惠券秒杀
好友关注
附近商户
用户签到
UV统计

导入项目

遇到了一个很恶心的ERROR

java: 警告: 源发行版 17 需要目标发行版 17

短信登录

基于Session的短信验证码登录

alt text

Redis代替Session解决session共享问题

对于用户信息，用String还是Hash??? 用String

商户查询缓存

什么是缓存

缓存的作用
- 降低后端负载
- 提高读写效率
- 降低响应时间
缓存的成本
- 数据一致性成本
- 代码维护成本
- 运维成本
  开发经验
用String 还是 hash ？
- 使用场景不同
我们这个缓存，缓存的是一个不会变的值，如果变了，其实是我们从数据库又读了一遍，所以针对这种情况我们直接使用string而不是hash

缓存更新策略

内存淘汰
- Redis自己的淘汰机制
超时剔除
- 一致性一般
- 维护成本低
主动更新
- 一致性好
- 维护成本高

面对业务进行选择

低一致性需求
高一致性需求

主动更新策略分类

Cache Aside Pattern
- coder自己写调用缓存的逻辑
Read/Write Through Pattern
- 我们不用管，服务会自己决定用redis还是sql
- 我们只需要调服务就行
Write Behind Caching Pattern
- 调用者只操作缓存
- 其他线程异步将缓存数据持久到数据库

最终选择Cache Aside Pattern

删除缓存还是更新缓存
- 更新缓存:
每次数据库更新，缓存就更新，无效写操作较多
- 删除缓存:
每次数据更新，我直接把缓存删了，你下次有人访问的时候再读。所以包选这个啊
如何保证缓存与数据库的操作同时成功或失败
- 单体系统：把缓存和数据库操作放在一个事务里面
- 分布式系统：利用TCC等分布式事务方案
先操作缓存还是先操作数据库？多线程并发视角
- 先删除缓存，再更新数据库：拿到旧数据，并存入缓存，更新缓存失效
- 先更新数据库，再删除缓存：也是拿到旧数据，但可能性耕地

缓存穿透

当有恶意的小人查一堆空id(或者其他)，redis查不到，db也查不到，如果他搞10000次请求，那就炸了，10000请求到数据库上

缓存空对象(使用这个)
- 实现简单，维护方便
- 缺点：
  - 数据不一样
  - 额外的内存消耗
布隆过滤(经典加一层)
- 存在误判的问题

实际开发中，我也不知道怎么选，这个老师选的存空值，只能说所有事情都是各有利弊,我觉得布隆运算器有点复杂，因为我们还需要考虑项目启动如何预热的问题

缓存雪崩

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

添加key时给TTL添加随机的存活值，别让一堆数据同时死，同时复活
Redis集群
给缓存业务添加降级限流策略
给业务添加多级缓存

缓存击穿

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

互斥锁：我们可以使用redis自身所具有的性质
- 优点
  - 没有额外的内存消耗
  - 保证一致性
  - 实现简单
- 缺点
  - 线程还是需要等待
  - 可能有死锁的风险

这里不多做解释：代码也不展示了，这里存在一些问题，比如：我们加锁和删锁还是有风险的，如果线程1的锁过期了，线程还没结束，那线程一可能就删除了线程二的锁了，我们需要给锁加一个校验的功能，这里就是使用lua进行改进，多加一个身份校验功能

逻辑过期

优点

线程无需等待，性能较好

缺点

不保证一致性

有额外内存消耗

实现复杂

实践代码

package com.hmdp.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.Shop;
import com.hmdp.mapper.ShopMapper;
import com.hmdp.service.IShopService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

import static com.hmdp.utils.RedisConstants.CACHE_SHOP_KEY;
import static com.hmdp.utils.RedisConstants.CACHE_SHOP_TTL;

 
@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Override
    public Result getShopById(Long id) {
        String key = CACHE_SHOP_KEY + id;
        //查redis有没有资源
        String s = stringRedisTemplate.opsForValue().get(key);
        //有就放行
//        if (s != null) {
//            Shop bean = JSONUtil.toBean(s, Shop.class);
//            System.out.println("从redis里面拿了 bean:"+bean.toString()+"  s:"+s);
//            return Result.ok(bean);
//        }
        //这里还是下面这种方法更好，因为这样避免了读空数据
        if (StrUtil.isNotBlank(s)){
            Shop bean = JSONUtil.toBean(s, Shop.class);
            System.out.println("从redis里面拿了 bean:"+bean.toString()+"  s:"+s);
            return Result.ok(bean);
        }

        //没有直接查数据库，
        Shop db_shop = getById(id);
        //数据库没有返回404
        if (db_shop == null) {
            // 防止缓存穿透，我们把null放进去
            stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL, TimeUnit.SECONDS);
            return Result.ok("店铺不存在");
        }
        //数据库有放置到redis
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(db_shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);
        System.out.println("从数据库里面拿了 bean:"+db_shop.toString());
        //返回数据
        return Result.ok(db_shop);
    }

    @Override
    public Result updateShopById(Shop shop) {
        Long id = shop.getId();
        //数据修改，先查数据库有没有这个id
        Shop db_shop = getById(id);
        //没有这个数据，直接返会
        if (db_shop == null) {
            return Result.fail("数据不存在idError");
        }
        //不为null，修改数据库
        updateById(shop);
        //存在这个数据，我直接把redis里面的数据删了，那我要不要设置上？不需要，留给下次来查询的人
        //让他负责放在redis里面
        stringRedisTemplate.delete(CACHE_SHOP_KEY+id);
        return Result.ok();
    }
}

//记得需要缓存预热
@Slf4j
@Service
public class CacheWarmUpService {

    @Autowired
     private IShopService shopService;

    @PostConstruct
    public void warmUp() {
        // 1. 查询所有需要预热的数据
        List<Shop> shops = shopService.list();

        // 2. 写入 Redis（逻辑过期结构）
        //这里，你想想，我们上面都把shops都拿出来了，
        // 我们下面执行的这个方法又去数据库查一遍，性能肯定浪费了
        //不过不担心，预热的时候不怕资源浪费？？我不信。。回来再说优化
        for (Shop shop : shops) {
            shopService.saveShop2redis(shop.getId());
        }

        log.info("Shop cache warm-up finished, count={}", shops.size());
    }
}

封装了Redis通用方法，虽然还有一点bug，但是已经很好用了

时看时新，

package com.hmdp.utils;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.jni.Time;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.CacheManager;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.hmdp.utils.RedisConstants.*;

/**
 * 作者：亥子曜
 * -后之览者，亦将有感于斯文
 */

@Component
@Slf4j
public class CacheClient {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private ObjectMapper objectMapper;
    //writeValueAsString   readValue

    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
    //java转json，string存储，支持过期时间
    private  <T,ID> void setT(String prefix,ID id,T t, Long time, TimeUnit timeUnit) throws JsonProcessingException {
        String s = objectMapper.writeValueAsString(t);
        String key = prefix + id;
        stringRedisTemplate.opsForValue().set(key, s, time, timeUnit);
    }
    private  <T> void setT(String key ,T t, Long time, TimeUnit timeUnit) throws JsonProcessingException {
        stringRedisTemplate.opsForValue().set(key, objectMapper.writeValueAsString(t), time, timeUnit);
    }
    //其实这么一写就知道了什么时候要加泛型，什么时候不加了
    //对于这里，你只要是个obj我就能写进去，这里是不是泛型没有任何作用
    //使用CACHE_+obj_id改进
    public void set(String obj_id ,Object obj, Long time, TimeUnit timeUnit) throws JsonProcessingException {
        stringRedisTemplate.opsForValue().set(CACHE_+ obj_id, objectMapper.writeValueAsString(obj), time, timeUnit);
    }




    //java转json, string存储，支持解决缓存击穿
    //缓存击穿：这里使用逻辑过期去存
    //这里也是，用不到泛型
    @Deprecated
    private <T> void setWithLogicExpireT(String obj_key,T t, Long time, TimeUnit timeUnit) throws JsonProcessingException {
        RedisData redisData = new RedisData();
        redisData.setData(t);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time)));
        stringRedisTemplate.opsForValue().set(CACHE_+obj_key, objectMapper.writeValueAsString(redisData));
    }

    //示例，我们搞定了"cache:",只需要填obj:20
    //使用CACHE_+obj_id改进
    public  void setWithLogicExpire(String obj_id,Object obj, Long time, TimeUnit timeUnit) throws JsonProcessingException {
        RedisData redisData = new RedisData();
        redisData.setData(obj);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time)));
        stringRedisTemplate.opsForValue().set(CACHE_+obj_id, objectMapper.writeValueAsString(redisData));
    }
    public  void setWithLogicExpire(String obj_id,Object obj ) throws JsonProcessingException {
        RedisData redisData = new RedisData();
        redisData.setData(obj);
        redisData.setExpireTime(LocalDateTime.now().plusMinutes(CACHE_TTL));
        stringRedisTemplate.opsForValue().set(CACHE_+obj_id, objectMapper.writeValueAsString(redisData));
    }

    //根据key查询缓存，并反序列化，利用缓存空值解决缓存穿透
    //我们到底用id查询还是用key查询，老师用的id，还得传进去一个prefix，这太繁琐了，我们就传进去key
    //但是尴尬的又来了，我们要用到id，所以这样不行
    //使用CACHE_+obj_key_id改进
    public <T,ID> T queryWithMutex(String obj_, ID id, Class<T> clazz,
                                         Function<ID,T> dbFallback, Long time, TimeUnit unit) throws JsonProcessingException, InterruptedException {
        String key = CACHE_+ obj_ + id ;

        int count = 20 ;
        while (count-- > 0) {
            //先去redis里面查
            String s = stringRedisTemplate.opsForValue().get(key);
            //查到了返回
            if (s != null && !s.isEmpty()) {
                if (NULL_VALUE.equals(s)) return null ;
                return objectMapper.readValue(s, clazz);
            }
            //没查到，去拿锁
            if (getlock(obj_+id)) break;
            //没拿到锁
            //睡一会
            Thread.sleep(50);
            //再去查值
        }
        if (count<=0) return null; //这里不要管返回null怎么怎么没有提示，我们留给调用者，
                                // 发现值为null，自己构造一个Result，因为我们这里是又泛型的，所以我们不能随便返回东西
        T t;

        try {
            //拿到锁了，去数据库里面找值
            t = dbFallback.apply(id);

            //没找到，放入redis里面一个""
            if (t == null){
                stringRedisTemplate.opsForValue().set(key, NULL_VALUE, CACHE_NULL_TTL, TimeUnit.SECONDS);

            }else{
                stringRedisTemplate.opsForValue().set(key, objectMapper.writeValueAsString(t), time, unit);
            }
        } finally {
            unlock(obj_+id);
        }
        return t;
    }
    //这个老师封装函数的能力还不如我们
    public <T,ID> T queryWithMutex(String obj_, ID id, Class<T> clazz, Function<ID,T> dbFallback) throws JsonProcessingException, InterruptedException {
        return queryWithMutex(obj_,id,clazz,dbFallback,CACHE_TTL,TimeUnit.MINUTES);
    }

    //根据key查询缓存，并反序列化，利用逻辑过期解决缓存击穿问题
    //待会再写

    public <T,ID> T queryWithLogicalExpire(String obj_, ID id, Class<T> clazz,
                                           Function<ID,T> dbFallback,Long time, TimeUnit unit) throws JsonProcessingException {
        String key = CACHE_+ obj_ + id ;
        // 去redis里面查询
        String s = stringRedisTemplate.opsForValue().get(key);
        //没查到，直接返回null
            //预热的时候没这个数据
            //我们这里旧按照没查到就是没数据处理，至于新加的数据如何预热，
            // 我们留给添加的方法，每添加一个就自动放入redis，对于热点数据就该这样
        if (s == null || s.isEmpty()) {
            return null ;
        }
        //查到了
        RedisData redisData = objectMapper.readValue(s, RedisData.class);
        T t = objectMapper.convertValue(redisData.getData(),clazz);
        LocalDateTime life = redisData.getExpireTime();
        //检查过期了没有
        //没过期就返回
        //if (life.isAfter(LocalDateTime.now()) ) return t ;
        //过期了
        //去拿锁，拿不到，返回旧数据
        //合并写了
        if (!life.isAfter(LocalDateTime.now()) && getlock(obj_+id) ) {
            //拿到锁了
            //new thread，返回旧数据
            //查询数据库
            //没查到，返回null
            //查到了，
            CACHE_REBUILD_EXECUTOR.submit(()->{
                try {
                    T newt = dbFallback.apply(id);
                    this.setWithLogicExpire(obj_+ id,newt,time,unit);
                } catch (JsonProcessingException e) {
                    log.error("缓存重建失败 key={} err={}", key, e.toString());
                }finally {
                    unlock(obj_+id);
                }
            });

        }
        return t;
    }
    //给了默认值的函数
    public <T,ID> T queryWithLogicalExpire(String obj_, ID id, Class<T> clazz,
                                           Function<ID,T> dbFallback) throws JsonProcessingException {
        return queryWithLogicalExpire(obj_,id,clazz,dbFallback,CACHE_TTL,TimeUnit.MINUTES);
    }

    boolean getlock(String obj_id) {
        try {
            return Boolean.TRUE.equals(
                    stringRedisTemplate.opsForValue()
                            .setIfAbsent(LOCK_+obj_id, "1", LOCK_TTL, TimeUnit.SECONDS)
            );
        } catch (Exception e) {
            log.error("Redis 获取锁异常 key={cache:"+obj_id+"}  error:"+e.getMessage());
            return false;
        }
    }

    private void unlock(String obj_id) {
        stringRedisTemplate.delete(LOCK_+obj_id);
    }
}

再次升级

package com.hmdp.utils;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.hmdp.utils.RedisConstants.*;

@Component
@Slf4j
public class CacheClient {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private ObjectMapper objectMapper;

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

    public void set(String objId, Object obj, Long time, TimeUnit unit) {
        String key = CACHE_ + objId;
        setJsonIfNotNull(key, obj, time, unit);
    }

    public void setWithLogicExpire(String objId, Object obj, Long time, TimeUnit unit) {
        String key = CACHE_ + objId;
        RedisData redisData = new RedisData();
        redisData.setData(obj);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
        setJsonIfNotNull(key, redisData);
    }

    public void setWithLogicExpire(String objId, Object obj) {
        String key = CACHE_ + objId;
        RedisData redisData = new RedisData();
        redisData.setData(obj);
        redisData.setExpireTime(LocalDateTime.now().plusMinutes(CACHE_TTL));
        setJsonIfNotNull(key, redisData);
    }

    public <T, ID> T queryWithMutex(
            String objPrefix, ID id, Class<T> clazz,
            Function<ID, T> dbFallback, Long time, TimeUnit unit) {

        String key = CACHE_ + objPrefix + id;

        int count = 20;
        while (count-- > 0) {
            String s = stringRedisTemplate.opsForValue().get(key);
            if (s != null && !s.isEmpty()) {
                if (NULL_VALUE.equals(s)) return null;
                return readJsonQuietly(s, clazz, key);
            }
            if (getlock(objPrefix + id)) break;
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
        }

        if (count <= 0) return null;

        T t;
        try {
            t = dbFallback.apply(id);
            if (t == null) {
                stringRedisTemplate.opsForValue().set(key, NULL_VALUE, CACHE_NULL_TTL, TimeUnit.SECONDS);
            } else {
                setJsonIfNotNull(key, t, time, unit);
            }
        } finally {
            unlock(objPrefix + id);
        }
        return t;
    }

    public <T, ID> T queryWithMutex(
            String objPrefix, ID id, Class<T> clazz,
            Function<ID, T> dbFallback) {
        return queryWithMutex(objPrefix, id, clazz, dbFallback, CACHE_TTL, TimeUnit.MINUTES);
    }

    public <T, ID> T queryWithLogicalExpire(
            String objPrefix, ID id, Class<T> clazz,
            Function<ID, T> dbFallback, Long time, TimeUnit unit) {

        String key = CACHE_ + objPrefix + id;
        String s = stringRedisTemplate.opsForValue().get(key);
        if (s == null || s.isEmpty()) return null;

        RedisData redisData = readJsonQuietly(s, RedisData.class, key);
        if (redisData == null) return null;

        T t = objectMapper.convertValue(redisData.getData(), clazz);
        LocalDateTime life = redisData.getExpireTime();

        if (!life.isAfter(LocalDateTime.now()) && getlock(objPrefix + id)) {
            CACHE_REBUILD_EXECUTOR.submit(() -> {
                try {
                    T newt = dbFallback.apply(id);
                    this.setWithLogicExpire(objPrefix + id, newt, time, unit);
                } catch (Exception e) {
                    log.error(e.getMessage());
                } finally {
                    unlock(objPrefix + id);
                }
            });
        }
        return t;
    }

    public <T, ID> T queryWithLogicalExpire(
            String objPrefix, ID id, Class<T> clazz,
            Function<ID, T> dbFallback) {
        return queryWithLogicalExpire(objPrefix, id, clazz, dbFallback, CACHE_TTL, TimeUnit.MINUTES);
    }

    private boolean getlock(String objId) {
        try {
            return Boolean.TRUE.equals(
                    stringRedisTemplate.opsForValue()
                            .setIfAbsent(LOCK_ + objId, "1", LOCK_TTL, TimeUnit.SECONDS)
            );
        } catch (Exception e) {
            log.error("Redis 获取锁异常 key={cache:" + objId + "}  error:" + e.getMessage());
            return false;
        }
    }

    private void unlock(String objId) {
        stringRedisTemplate.delete(LOCK_ + objId);
    }

    private <T> T readJsonQuietly(String json, Class<T> clazz, String key) {
        try {
            return objectMapper.readValue(json, clazz);
        } catch (Exception e) {
            log.error("反序列化失败 key={} json={} err={}", key, json, e.toString());
            return null;
        }
    }

    private String writeJsonQuietly(Object obj, String key) {
        try {
            return objectMapper.writeValueAsString(obj);
        } catch (Exception e) {
            log.error("序列化失败 key={} objType={} err={}", key, obj == null ? "null" : obj.getClass().getName(), e.toString());
            return null;
        }
    }

    private void setJsonIfNotNull(String key, Object obj, long time, TimeUnit unit) {
        String json = writeJsonQuietly(obj, key);
        if (json != null) {
            stringRedisTemplate.opsForValue().set(key, json, time, unit);
        }
    }

    private void setJsonIfNotNull(String key, Object obj) {
        String json = writeJsonQuietly(obj, key);
        if (json != null) {
            stringRedisTemplate.opsForValue().set(key, json);
        }
    }
}

全局唯一ID

需要全局ID生成器
绝对不能使用Mysql的自增长
ID（Long）

企业中会用到不同的ID获取办法

UUID
Redis自增
- 符号位
- 时间戳31位
- 序列号32位
snowflake算法
数据库专门搞一张表用来自增

优惠券秒杀下单

超卖问题

悲观锁
- 认为线程安全问题一定发生，因此先加锁，确保串行执行
- 性能低
乐观锁
- 认为线程安全不一定发生，只在修改数据时加锁
- 实现方法：
  - 版本号法：
  - CAS法：
    - 直接把剩余数量作为版本号，判断版本号是否变化。这样虽然实现很简单，但实质上是变成了”拒绝所有并发操作”,用户体验极差，成功率极低
    - 乐观锁，update的时候去判断剩余数量是否>0。由于update在mysql有行级锁，所以可行，订单的创建会因为事务而回滚取消

一人一单

最简单实现: 多一个查询，查询user_id 和 voucher_id ，保证count不能有多个，但是仍然有并发问题
解决：加悲观锁。直接用synchronized锁住
- 给谁加锁？通过user_id加锁，而且，userId.toString().intern()这个是为了区分对象和真正的常量池的
- 这里有一个事务失效的点，也就是说，一个方法内部调用另一个开启事务的方法，事务是失效的。（详细见AOP动态代理的原理）
用到了 aspectj的依赖 @EnableAspectJAutoProxy(exposeProxy=true)

分布式锁

多进程可见
互斥
高可用
高性能
安全性
其他功能性

问题一：分布式锁，我们需要注意如果在第一个请求没有结束的时候，锁过期了的问题。我们要搞一个锁删除时候的校验
问题二：问题一解决了一部分情况，但是，在释放锁的时候，需要先查再释放，由于这是两步而不是原子操作，所以在这两步的间隙中，很可能产生并发问题导致误删。
lua脚本

EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name Rose

分布式锁优化

不可重入:方法A拿到锁之后调用方法B，而方法B也需要锁，就导致了死锁
不可重试:获取锁没有重试机制
超时释放:
主从一致性:

Redisson

可重入锁

上面的复杂逻辑绝对不能用java代码写，因为需要保证原子性操作，底层也是用的lua脚本实现的

锁重试和WatchDog

multilock

主从一致性问题

做主节点集群，这样只有三个同时获取锁成功，才算成功

秒杀优化

阻塞式队列BlockingQueue实现

配合消费线程使用
但是用的是jvm的资源，不太建议
应该优化为消息队列
应为这个阻塞式队列拿出来就没了，出现异常情况就啥也没有了
Redis消息队列

Redis也能实现消息队列

基于list实现

LPUSH k1 v1
BRPOP k1 #阻塞式

PubSub

可以通过订阅channel 广播，定向，话题和RabbitMQ相类似

基于Stream

XADD s1 * k1 v1 
XREAD COUNT 1 STREAMS s1 0
XREAD COUNT 1 BLOCK 0 STREAMS s1 $

如果使用$,会出现漏读消息的情况

XGROUP CREATE s1 g1 0 
XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >  
XACK s1 g1 {message_id}
XPENDING s1 g1 - + 10
XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0

实战

XGROUP CREATE stream.orders g1 0 MKSTREAM

XADD stream.orders * k1 v1 k2 v2 k3 v3

XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.order >

这部分感觉老师就是降维打击我去，太复杂了

#### 达人探店 ##### 点赞功能就是按照锁那种形式存储在redis中，和加锁一样的思路，实现每个用户只能点赞一次 ##### 点赞排行榜

SortedSet是个好东西

ZADD z1 1 m1 2 m2 3 m3 
ZSCORE  z1 m1
ZRANGE  z1 0 4 #按照范围查询

SELECT * FROM table WHERE id IN (5,1);

当你看到这句话的时候，应该特别注意：MySQL的查询永远是默认正序查询，所以即使用了 IN (5,1) 仍然得到的结果不是我们想要的顺序，而是默认正向顺序

ORDER BY FIELD(id,5,1)

加上这一句话就好了

List<UserDTO> userDTOS = userService.query()
    .in("id",ids).last("ORDER BY FIELD(id,"+ idStr+ ")").list()
    .stream().map(user->BeanUtils.copyProperties(user,UserDTO.lcass))
    .collect(Collectors.toList());

关注和取关

就是一个基本的功能，这里就是建立了一个follow表

表中存储用户关注情况

共同好友

利用redis求交集

SADD s1 v1 v2 SADD s2 v1 v3 SINTER s1 s2

利用redis中set的特性可以求出交集

关注推送(Feed流)

Timeline: 不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈
- 优点：信息全面，不会有缺失，实现简单
- 缺点：信息噪声较多，用户不一定感兴趣，内容获取效率低
智能排序：利用智能算法屏蔽掉违规的，用户不感兴趣的内容。推送用户感兴趣信息来吸引用户
- 优点：投喂用户感兴趣信息，用户黏度很高，容易沉迷
- 缺点：算法不精准可能起到反作用

TimeLine

拉模式: 读扩散
- 缺点: 每次读需要先拉取，速度慢
推模式: 写扩散
- 缺点: 内存占用高，一个消息需要写多份
推拉结合: 读写混合
- 对于普通up: 推模式，内存占用低
- 对于大V:
  - 普通粉丝：拉模式
  - 活跃粉丝: 推模式

上千万用户才算用户量大

推模式实现

对于分页功能用list还是zset
- 传统分页的弊端：使用list我们查询了第一页之后，有新的推送，再次查第二页，就会有重复查询
- 滚动分页：每次都会记录last_index，保证了不会有重复查询
  - 仍然会有bug，比如说如果有10个数据得分相同，而我们的count小于10，那么分页查询直接失效。因为从这个得分往后数count个还是这个得分，就陷入这里出不来了。当然也有解决办法，我们就统计一下和最后一个数据相同的数量，用作offset，但仍然不能根治问题，当重复超过count的二倍
  - 严重不支持跳跃分页查询，适合一直往下滑这种(类似抖音) ``` bash ZREVRANGE z1 0 2 WITHSCORES

ZREVRANGEBYSCORE z1 100 0 WITHSCORES LIMIT 0 3 # offset | count ZREVRANGEBYSCORE z1 100 0 WITHSCORES LIMIT 1 3

* `ZREVRANGEBYSCORE`需要传入四个参数max min offset count

``` java
 listById(); //这个方法底层基于In，顺序不是我们想要的数据
 query().in("in",ids).last("ORDER BY 省略了"); //这样自己写

 //但是，我比较建议在java程序中修改，因为数据量不大这种情况
 List<Blog> sorted =
    list.stream()
        .sorted(Comparator.comparingInt(
            b -> ids.indexOf(b.getId())
        ))
        .toList();

附近商户

基于Redis实现地理坐标的数据结构GEO

需要知道GEO实现的只不过是直线距离，但是日常生活中我们需要用到的是路程距离
尽管如此，GEO的实用价值仍然很高，因为，我们可以实用GEO来做粗筛，然后再调用地理精确地图接口来再次判断
好用，便宜

GEOADD  #添加一个点
GEODIST
GEOHASH   #将坐标转为hash字符串并返回
GEOPOS  # 返回坐标
GEORADIUS  # 指定圆心半径，按照距离排序后返回 6.2废弃
GEOSEARCH # 在指定范围搜索，排序返回
GEOSEARCHSTORE  # 自动缓存到ZSet 


GEOADD key1 116 39 name_detail1 116 37 name_detail2
# 底层还是ZSET, value 是name_detail, score是经纬度的hash+算法

GEODIST g1 name1 name2 # 默认返回单位是m
GEODIST g1 name1 name2 km
GEOSEARCH g1 name1 BYRADIUS 10 km 
GEOSEARCH g1 FROMLONLAT 116 39  BYRADIUS 10 km WITHDIST #默认升序 
GEOHASH g1 name1

Map<Long ,List<Shop>> map = list.stream().collect(
    Collectors.groupingBy(Shop::getTypeID);
);

stringRedisTemplate.opsForGeo().add("shop:geo", map_list);

//老师这里写的代码逻辑很乱。我们直接装一波
Map<Long, List<NewType>> map =
    list.stream()
        .collect(Collectors.groupingBy(
            Shop::getTypeID,
            Collectors.mapping(
                shop -> {
                    // 在这里对 shop 做处理
                    NewType newType = new NewType();
                    newType.setId(shop.getId());
                    newType.setName(shop.getName());
                    // 你可以做更多逻辑
                    return newType;
                },
                Collectors.toList()
            )
        ));



stringRedisTemplate.opsForGeo().
    search(
        key,GeoReference.fromCoordinate(x,y),
        new Distance(5000),
        RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end) //这里很遗憾
        //limit不能自己指定数量
        //所以需要我们查询所有，然后自己去截取
    );

list.stream().skip(from_index).forEach(result -> {
    String shopIdStr = result.getContent().getName();
    Distance distance = result.getDistance();
});

//这个skip操作会有很大风险，导致后面出现空list
//加个判断 
if(list.size () <= from) return Result.ok()

用户签到

如果我们使用MySQL，那么数据量就大得不行，天文数字
所以我们推荐使用位图(BitMap)

SETBIT bm1 0 1
GETBIT bm1 2
BITCOUNT bm1
BITFIELD bm1 GET u2 0  #从0开始数2位无符号
BITFIELD_RO 
BITOP
BITPOS key bit [start] [end]

签到统计

连续签到次数

对于这样一个问题，我想到了一个非常奇妙的操作
我利用for一直查0再查1

# 这里是伪代码，但是这个不可行
#原因是 BITPOS  这个时间复杂度不低
local max = 0 
local bgein
for  then
    local start = BITPOS key 1 0
    local end = BITPOS key 0 start
    local gap = end - start
    IF gap > max then
        max = gap
    END
    begin = end
end


# 建议用这个

local max = 0
local run = 0
for i = 0, days-1 do
  if bit.band(bit.rshift(v, i), 1) == 1 then
    run = run + 1
    if run > max then max = run end
  else
    run = 0
  end
end

UV统计

HyperLogLog

PFADD hl1 e1 e2 e4 e3
PFCOUNT hl1