Redis从入门到实践

Redis初识

Redis是什么

• 是一个高性能Key-Value服务器
• 多种数据结构
• 丰富的功能
• 高可用分布式支持

谁在使用Redis

• Github、twitter、微博、百度、美团、阿里巴巴等

Redis特性

• 速度快（数据存储在内存中，由C语言实现，单线程）
• 持久化（数据的更新异步保存在磁盘上，确保断电不丢数据）
• 多种数据结构
• 支持多种编程语言
• 主从复制
• 高可用分布式

Redis典型使用场景

• 缓存系统
• 计数器
• 消息队列系统
• 排行榜
• 社交网络（点赞数、评论数）
• 实时系统

Redis启动方式

• 最简单的启动方法

  redis-server
  
  ps -ef|grep redis           # 查看redis进程
  netstat -antpl|grep redis   # 查看redis端口

• 动态参数启动

  redis-server --port 6380    # 以6380端口启动redis

• 配置文件启动

  redis-server configPath

Redis常用配置

• daemonize 是否是守护进程(no|yes) 默认为no
• port 启动端口 默认6379
• logfile 系统日志
• dir 工作目录

Redis的常用API

通用命令

• keys（时间复杂度o(n)，一般不在dev环境使用）

  keys *          # 遍历输出所有key的值
  keys [pattern]  # 指定匹配样式

• dbsize（计算key的总数）

• exists key（若key存在返回1，否则返回0）

• del key [key …]（成功删除返回1，否则返回0）

• expire key seconds（key在seconds秒后过期）

• ttl key（查看key剩余的过期时间，若key过期返回-2，若key存在且没有过期时间返回-1）

• persist key（去掉key的过期时间）

• type key（返回key的类型）

  注意：以上命令除了keys，其他命令时间复杂度都是O(1)

数据库与内部编码

五大数据结构

字符串

• 使用场景：缓存、计数器、分布式锁等

• 常用API

  命令	说明	时间复杂度
  incr key	key自增1，若key不存在，自增后get(key)=1	o(1)
  decr key	key自减1，若key不存在，自减后get(key)=-1	o(1)
  incrby key k	key自增k，若key不存在，自增后get(key)=k	o(1)
  decrby key k	key自减k，若key不存在，自减后get(key)=-k	o(1)
  set key value	不管key是否存在，都设置	o(1)
  setex key 30 value	设置key-value，30秒后过期
  setnx key value	key不存在才设置	o(1)
  set key value xx	key存在才设置	o(1)
  mget key1 key2 …	批量获取key，原子操作	o(n)
  mset key1 value1 key2 value2 …	批量设置key-value	o(n)
  getset key newvalue	set key newvalue并返回旧value	o(1)
  append key value	将value追加到旧的value	o(1)
  strlen key	返回value的字符串长度	o(1)

• 应用：记录网站每个用户个人主页的访问量

  incr userid:pageview（单线程：无竞争）

• 应用：缓存视频的基本信息（数据源在mysql中）伪代码

  // 注意这是伪代码，了解大概逻辑即可
  public VideoInfo get(long id){
      String redisKey = redisPrefix + id;
      VideoInfo video = redis.get(redisKey);
      if(video == null){
          video = mysql.get(id);
          if(video!=null){
              //序列化
              redis.get(redisKey, serialize(videoInfo));
          }
      }
  }

• 应用：分布式id生成器

哈希 map

• 键值结构（类似于mapmap）

  key(例如user)->filed(例如name,age,Date)->value(…)

  field不能相同，value可以相同

• 常用api

  命令	说明	时间复杂度
  hget key field	获取hash key对应的field的value	o(1)
  hset key field value	设置hash key对应field的value	o(1)
  hdel key field	删除hash key对应的field的value	o(1)
  hexists key field	判断hash key是否含有field	o(1)
  hlen key	获取hash key field的数量	o(1)
  hmget key field1 field2 …	批量获取hash key中一批field的值	o(n)
  hmset key field1 value1 field 2 value 2…	批量设置hash key的一批field value	o(n)
  hgetall key	返回hash key对应的所有field和value	o(n)
  hvals key	返回hash key对应的所有field的value	o(n)
  hkeys key	返回hash key对应的所有field	o(n)
  hincrby key field num	对应key的field的value自增num	o(1)
  hincrbyfloat key field num	与上相同，浮点数版本	o(1)

• 应用：记录网站每个用户个人主页访问量

  hincrby user:1:info pageview count

• 应用：缓存视频的基本信息（数据源存mysql）伪代码

  public VideoInfo get(long id){
      String redisKey = redisPrefix+id;
      Map<String, String> hashMap = redis.hgetAll(redisKey);
      VideoInfo video = transferMapToVideo(hashMap);
      if(video == null){
          videoInfo = mysql.get(id);
      }else{
          redis.hmset(redisKey, transferVideoToMap(videoInfo));
      }
  }

列表 list

• 键值结构

  key-elements(列表)

• 有序，允许重复，可左右两边插入弹出

• API-增加

  命令	说明	时间复杂度
  rpush key value1,value2… valueN	从列表右端插入值(1~N个)	o(1~N)
  lpush key value1,value2… valueN	从列表右端插入值(1~N个)	o(1~N)
  linsert key before|after value newValue	在指定key的value前或后插新值	o(n)

• API-删除

  命令	说明	时间复杂度
  lpop key	从列表左端弹出一个项	o(1)
  lrem key count value	根据count值从列表删除与value相等的项	o(n)
  ltrim key start end	按照索引范围修剪列表	o(n)

  注意lrem命令：

  1.count>0时，从左到右，删除最多count个与value相等的项。

  2.count<0时，从右到左，删除最多|count|个与value相等的项。

  3.count=0时，删除所有value相等的项。

• API-修改

  命令	说明	时间复杂度
  lset key index newValue	设置指定索引值为newValue	o(n)

• API-查询

  命令	说明	时间复杂度
  lrange key start end	获取列表指定索引范围的项，0 -1代表查询全部	o(n)
  lindex key index	根据索引值获取项	o(n)
  lren key	获取列表长度	o(1)

• 应用：TimeLine 微博按时间顺序排序

集合 set

• 键值结构：key-values

• 特点：无序、不重复

• 常用API

  命令	说明	时间复杂度
  sadd key element	向集合key添加element，若存在则添加失败	o(1)
  srem key element	将集合key中的element删除	o(1)
  sismember key element	判断集合key中element是否存在	o(n)
  srandmember key	从集合随机挑选元素，后面可指定个数	o(1)
  spop key	从集合中随机弹出元素	o(1)
  smembers key	返回集合中所有元素 (谨慎使用)	o(n)
  scard key	计算集合元素个数	o(1)
  sdiff key1 key2	返回key1-key2差集
  sinter key1 key2	返回key1与key2交集
  sunion key1 key2	返回key1余key2并集

  sdiff|sinter|sunion + store destkey … 将差集、交集、并集结果保存在destkey中

• 应用：抽奖系统，抽到后用spop弹出

• 应用：标签，为用户添加标签等 sadd user1 tag1 tag2… ，也可为标签添加用户

• 应用：共同关注

有序集合 zset

• 无重复元素、有序 key - element + score

• 常用API

  命令	说明	时间复杂度
  zadd key score element	添加score与element	o(logN)
  zrem key element	删除元素	o(1)
  zscore key element	返回元素分数	o(1)
  zincrby key increScore element	增加或减少元素分数	o(1)
  zcard key	返回元素个数	o(1)
  zrank key element	获取指定元素排名(从小到大，从0开始)	o(1)
  zrange key start end [WITHSCORES]	获取指定索引范围升序元素	o(log(n)+m)
  zrangebyscore key minScore maxScore [WITHSCORES]	获取指定分数范围升序元素	o(log(n)+m)
  zcount key minScore maxScore	获取指定分数范围内元素个数	o(log(n)+m)
  zremrangebyrank key start end	删除指定排名内升序元素	o(log(n)+m)
  zremrangebyscore key minScore maxScore	删除指定分数内的升序元素	o(log(n)+m)

• 应用：排行榜

Redis客户端的使用

Java客户端：Jedis

• Maven依赖

  <dependency>
      <groupId>redis.clients</groupId>
      <artifactId>jedis</artifactId>
      <version>2.9.0</version>
      <type>jar</type>
      <scope>compile</scope>
  </dependency>

• 连接Redis示例代码

  Jedis jedis = new Jedis("127.0.0.1",6379); //TCP连接
  jedis.set("hello","world");
  String value = jedis.get("hello");
  // Jedis(String host, int port, int connectionTimeout, int soTimeout)

• Jedis连接池的使用

  连接池是将jedis对象进行统一管理，在使用完Redis后jedis对象并不会被销毁，而是归还至连接池中，这种方式和直接连接Redis有什么区别呢？

  	优点	缺点
  直接连接	简单方便 适用于少量长期连接的场景	每次连接要新建/关闭TCP 资源无法控制，存在连接泄露可能 Jedis对象线程不安全
  连接池	Jedis预先生成，降低开销 连接池的形式保护和控制资源的使用	使用较麻烦，在资源管理上需要很多参数，规划不合理时容易出现问题

  代码示例

  // 创建连接池配置类与连接池
  GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
  JedisPool jedisPool = new JedisPool(poolConfig,"127.0.0.1",6379);
  try{
      // 1.从连接池获取jedis对象
      jedis = jedisPool.getResource();
      // 2.执行操作
      jedis.set("hello","world");
  }catch(Exception e){
      e.printStackTrace();
  }finally{
      // 3.如果jedis从连接池获得，使用close不是关闭连接而是归还至连接池
      if(jedis!=null){
          jedis.close();	
      }   
  }

Redis的其他功能

慢查询

• 生命周期

  

• 两个配置

  slowlog-max-len

  slowlog-log-slower-than

• 配置方法

  config get slowlog-max-len = 128 （长度默认128）

  config get slowlog-log-slower-than = 10000（默认阈值10000微秒）

  动态配置 config set …

• 慢查询命令

  slowlog get[n]：获取慢查询队列

  slowlog reset：慢查询清空

pipeline

• 将多条命令打包统一发送

• 在jedis中使用举例

  /*将1w条指令拆分成100次发送，运行效率大幅度提高*/
  for(int i=0;i<100;i++){
      Pipeline pipeline = jedis.pipelined();
      for(int j = i*100; j<(i+1)*100; j++){
          pipeline.hset("hashkey:"+j, "field:"+j, "value"+j);
      }
      pipeline.syncAndReturnAll();
  }

• 注意：pipeline每次只能作用在一个Redis节点上

发布订阅

• 角色

  发布者（publisher）

  订阅者（subscriber）

  频道（channel）

• 常用API

  publish channel message 发布消息至某个频道

  subscribe [channel] 订阅一个或多个频道

  unsubscribe [channel] 取消订阅一个或多个频道

  psubscribe [pattern] 订阅模式

  punsubscribe [pattern] 取消订阅模式

  pubsub channels 列出最少有一个订阅者的频道

  pubsub numsub [channel] 列出给定频道的订阅者数量

三种特殊数据结构

Bitmaps 位图

• 位存储，使用场景有统计用户信息是否活跃、是否登录、打卡等（注意1字节=8位）

• setbit设置值（以打卡功能为例）

  setbit sign 0 0    # 星期一未打卡
  setbit sign 1 1    # 星期二已打卡
  setbit sign 7 1    # 此时索引2~6的值均为0

• 查看某一天是否打卡

  getbit sign 0    # 查看星期一是否打卡

• 统计操作

  bitcount sign    # 统计打卡天数，即统计值为1的个数

Hyperloglog

什么是基数？

基数为不重复的元素

Redis2.8.9中更新了Hyperloglog数据结构

用于做基数统计的算法，例如网页的UV（点击页面多次，但算做一个人访问网站），在传统方式使用set保存用户id，然后统计set集合中的元素数量。但统计目的是为了计数，而不是保存用户id。

Hyperloglog优点：占用内存固定，2^64不同元素的基数，只需12KB内存

缺点：有一点错误率，但可以忽略不计

常用方法

• 添加元素

  PFadd myCount a b c c    # 向myCount集合中添加元素

• 统计操作

  PFCOUNT myCount    # 结果为3

• 将两个集合合并成新集合

  PFMERGE newKey oldKey1 oldKey2    # 将oldKey1和oldKey2合并到newKey集合

Geospatial地理位置

可以查询测试数据：http://www.jsons.cn/lngcode/

比如推断两地之间距离

六个命令：GEOADD GEODIST GEOHASH GEOPOS GEORADIUS GEORADIUSBYMEMBER

添加地理位置（经纬度）

有效经度为-180到180，有效纬度为-85.05112878到85.05112878

geoadd china:city 114.13 22.54 shenzhen

获取经纬度

geopos china:city shenzhen

获取距离 m米 km千米 mi英里 ft英尺

geodist china:city shenzhen zhuhai km

以经纬度为中心查找

以某一点为中心，给出半径寻找

georadius china:city 110 30 1000 km
georadius china:city 110 30 1000 km withdist withcoord count 1
# withdist显示直线距离，withcoord显示被查找点的经纬度

以成员为中心查找

georadiusbymember china:city shenzhen 1000 km

返回元素的hash表示

geohash china:city shenzhen zhuhai
# 如果两个字符串越接近，则距离越近

底层是Zset，删除一个地理位置如下

zrem china:city shenzhen

Redis事务与锁

Redis单条命令保证原子性，但是Redis事务不保证原子性，没有隔离级别的概念。

开启事务

127.0.0.1:6379> multi    # 开启事务
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> exec    # 执行事务
1) OK
2) OK
3) "v2"

关闭事务

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> get k1
QUEUED
127.0.0.1:6379> discard    # 放弃事务

注意：

编译型异常（命令错误），事务中所有命令都不执行

运行时异常，其他命令正常执行

乐观锁与悲观锁

悲观锁无论什么时候都会加锁

乐观锁无论什么时候都不上锁（但在更新数据时进行判断是否有人修改过数据）

# 使用乐观锁
set money 100
set out 0

# 监视money，并使用事务
127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY out 20
QUEUED
127.0.0.1:6379> exec
1) (integer) 80
2) (integer) 20
# 以上是正常执行的情况
# 下面是修改失败的情况（中途有线程修改了money）
127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 10
QUEUED
127.0.0.1:6379> INCRBY out 10
QUEUED
127.0.0.1:6379> exec
(nil)
# 总结：使用watch可以当做乐观锁操作,unwatch可以解锁

Redis整合SpringBoot

整合方式

在SpringBoot2.x中，原来的jedis被替换成lettuce

jedis：采用直接连接，多个线程操作不安全，因此要使用连接池

lettuce：采用netty，实例可以在多个线程中共享，不能存在线程不安全情况

整合

1.导入依赖

<!--操作Redis-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.配置连接

# 更多配置项可以点进源码查看
spring.redis.host=127.0.0.1
spring.redis.port=6379

3.代码

@SpringBootTest
class RedisSpringbootApplicationTests {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    void contextLoads() {

        // opsForValue 操作字符串 String
        // opsForList 操作List 列表
        // opsFor ... 以此类推
        
        // 还有一些常用操作也可以直接调用redisTemplate的方法
        
        redisTemplate.opsForValue().set("banana","123");
        System.out.println(redisTemplate.opsForValue().get("banana"));

    }
}

注意：

一般开发中，要将实体类序列化

@Component
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User implements Serializable {
    private String name;
    private int age;
}

// 或者
String jsonUser = new ObjectMapper().writeValueAsString(user);
redisTemplate.opsForValue().set("user",jsonUser);

或者使用Redis模板

RedisTemplate

@Configuration
public class RedisConfig {
    // 编写自己的配置类
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory)
            throws UnknownHostException {

        // 方便开发，一般直接使用<String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setConnectionFactory(factory);

        // Json序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        //String的序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        //key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        //hash的key也采用String序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        //value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();

        return template;
    }
}

RedisUtils

注意：这个类是基于上方模板类实现

// 在我们真实的分发中，或者你们在公司，一般都可以看到一个公司自己封装RedisUtil
@Component
public final class RedisUtil {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    // =============================common============================
    /**
     * 指定缓存失效时间
     * @param key  键
     * @param time 时间(秒)
     */
    public boolean expire(String key, long time) {
        try {
            if (time > 0) {
                redisTemplate.expire(key, time, TimeUnit.SECONDS);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 根据key 获取过期时间
     * @param key 键 不能为null
     * @return 时间(秒) 返回0代表为永久有效
     */
    public long getExpire(String key) {
        return redisTemplate.getExpire(key, TimeUnit.SECONDS);
    }


    /**
     * 判断key是否存在
     * @param key 键
     * @return true 存在 false不存在
     */
    public boolean hasKey(String key) {
        try {
            return redisTemplate.hasKey(key);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除缓存
     * @param key 可以传一个值 或多个
     */
    @SuppressWarnings("unchecked")
    public void del(String... key) {
        if (key != null && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            } else {
                redisTemplate.delete(CollectionUtils.arrayToList(key));
            }
        }
    }


    // ============================String=============================

    /**
     * 普通缓存获取
     * @param key 键
     * @return 值
     */
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }
    
    /**
     * 普通缓存放入
     * @param key   键
     * @param value 值
     * @return true成功 false失败
     */

    public boolean set(String key, Object value) {
        try {
            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 普通缓存放入并设置时间
     * @param key   键
     * @param value 值
     * @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期
     * @return true成功 false 失败
     */

    public boolean set(String key, Object value, long time) {
        try {
            if (time > 0) {
                redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
            } else {
                set(key, value);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 递增
     * @param key   键
     * @param delta 要增加几(大于0)
     */
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }


    /**
     * 递减
     * @param key   键
     * @param delta 要减少几(小于0)
     */
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }


    // ================================Map=================================

    /**
     * HashGet
     * @param key  键 不能为null
     * @param item 项 不能为null
     */
    public Object hget(String key, String item) {
        return redisTemplate.opsForHash().get(key, item);
    }
    
    /**
     * 获取hashKey对应的所有键值
     * @param key 键
     * @return 对应的多个键值
     */
    public Map<Object, Object> hmget(String key) {
        return redisTemplate.opsForHash().entries(key);
    }
    
    /**
     * HashSet
     * @param key 键
     * @param map 对应多个键值
     */
    public boolean hmset(String key, Map<String, Object> map) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * HashSet 并设置时间
     * @param key  键
     * @param map  对应多个键值
     * @param time 时间(秒)
     * @return true成功 false失败
     */
    public boolean hmset(String key, Map<String, Object> map, long time) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @param time  时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value, long time) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除hash表中的值
     *
     * @param key  键 不能为null
     * @param item 项 可以使多个 不能为null
     */
    public void hdel(String key, Object... item) {
        redisTemplate.opsForHash().delete(key, item);
    }


    /**
     * 判断hash表中是否有该项的值
     *
     * @param key  键 不能为null
     * @param item 项 不能为null
     * @return true 存在 false不存在
     */
    public boolean hHasKey(String key, String item) {
        return redisTemplate.opsForHash().hasKey(key, item);
    }


    /**
     * hash递增 如果不存在,就会创建一个 并把新增后的值返回
     *
     * @param key  键
     * @param item 项
     * @param by   要增加几(大于0)
     */
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }


    /**
     * hash递减
     *
     * @param key  键
     * @param item 项
     * @param by   要减少记(小于0)
     */
    public double hdecr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, -by);
    }


    // ============================set=============================

    /**
     * 根据key获取Set中的所有值
     * @param key 键
     */
    public Set<Object> sGet(String key) {
        try {
            return redisTemplate.opsForSet().members(key);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 根据value从一个set中查询,是否存在
     *
     * @param key   键
     * @param value 值
     * @return true 存在 false不存在
     */
    public boolean sHasKey(String key, Object value) {
        try {
            return redisTemplate.opsForSet().isMember(key, value);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将数据放入set缓存
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSet(String key, Object... values) {
        try {
            return redisTemplate.opsForSet().add(key, values);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 将set数据放入缓存
     *
     * @param key    键
     * @param time   时间(秒)
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSetAndTime(String key, long time, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().add(key, values);
            if (time > 0)
                expire(key, time);
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 获取set缓存的长度
     *
     * @param key 键
     */
    public long sGetSetSize(String key) {
        try {
            return redisTemplate.opsForSet().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 移除值为value的
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 移除的个数
     */

    public long setRemove(String key, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().remove(key, values);
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }

    // ===============================list=================================
    
    /**
     * 获取list缓存的内容
     *
     * @param key   键
     * @param start 开始
     * @param end   结束 0 到 -1代表所有值
     */
    public List<Object> lGet(String key, long start, long end) {
        try {
            return redisTemplate.opsForList().range(key, start, end);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 获取list缓存的长度
     *
     * @param key 键
     */
    public long lGetListSize(String key) {
        try {
            return redisTemplate.opsForList().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 通过索引 获取list中的值
     *
     * @param key   键
     * @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
     */
    public Object lGetIndex(String key, long index) {
        try {
            return redisTemplate.opsForList().index(key, index);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     */
    public boolean lSet(String key, Object value) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将list放入缓存
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     */
    public boolean lSet(String key, Object value, long time) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            if (time > 0)
                expire(key, time);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @return
     */
    public boolean lSet(String key, List<Object> value) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     * @return
     */
    public boolean lSet(String key, List<Object> value, long time) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            if (time > 0)
                expire(key, time);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 根据索引修改list中的某条数据
     *
     * @param key   键
     * @param index 索引
     * @param value 值
     * @return
     */

    public boolean lUpdateIndex(String key, long index, Object value) {
        try {
            redisTemplate.opsForList().set(key, index, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 移除N个值为value
     *
     * @param key   键
     * @param count 移除多少个
     * @param value 值
     * @return 移除的个数
     */

    public long lRemove(String key, long count, Object value) {
        try {
            Long remove = redisTemplate.opsForList().remove(key, count, value);
            return remove;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }

    }

}

Redis.conf配置文件

单位

unit单位对大小写不敏感，例如1GB 1Gb 1gB 1gb是一样的单位

配置

可以使用include导入多个conf配置文件

网络

bind用以绑定ip

protected-mode为保护模式

port为端口设置

通用

daemonize为以守护进程的方式进行，默认为no，需自己配置为yes

pidfile用以指定pid文件，如果要以后台方式运行则需要指定

loglevel用以指定日志级别

logfile用以指定日志文件位置

databases为数据库的数量，默认16

always-show-logo用以配置控制台是否打印redis的logo，默认为yes

快照

持久化，在规定的时间执行了多少次操作，会持久化到文件

save 900 1    # 900秒内至少有1个key进行修改，则持久化
save 30 100
save 60 10000

stop-writes-on-bgsave-error yes    # 若持久化出错是否还继续工作
rdbcompression yes    # 是否压缩rdb文件，消耗cpu资源
rdbchecksum yes    # 保存rdb时，进行错误校验
dir ./    # rdb文件保存目录

安全

requirepass 123456    # 配置密码为123456
config set requirepass 密码    # 或者在命令执行这条语句
auth 密码    # 使用密码进行登录

限制

maxclients 10000 # 设置连接最大客户端数量
maxmemory <bytes> # 最大内存容量
maxmemory-policy noeviction # 内存到达上限后处理策略

AOF配置

appendonly no    # 默认不开启，默认使用rdb方式持久化
appendfilename "appendonly.aof"    # 持久化的文件名字
appendfsync everysec    # 配置保存频率, always | everysec | no

Redis持久化

RDB（Redis Database）

默认保存的文件是dump.rdb

优点：

1. 适合大规模数据恢复，高效

2. 对数据完整性要求不高

缺点：

1. 需要一定时间间隔进行操作，最后一次持久化的数据可能会丢失

2. fork进程会占用一定内存空间

触发机制

1. save的规则满足条件，会触发rdb规则

2. 执行flushall命令，也会触发rdb规则

3. 退出redis，也会产生rdb文件

恢复rdb文件

只需要将rdb文件放在redis启动目录下即可

config get dir    # 获取目录位置，若存在rdb文件会自动恢复数据

AOF（Append Only File）

将所有命令（除了读命令）都记录下来，恢复的时候重新执行一遍命令

默认不开启，命令保存在appendonly.aof文件

若aof文件损坏，redis将无法启动，需要修复aof文件

redis-check-aof --fix appendonly.aof    #修复

优点：

相对rdb来说备份数据完整性会更好

缺点：

数据文件比rdb大很多，修复速度、运行效率也更慢

Redis主从复制

概念与基础

主从复制指的是将一台Redis服务器数据复制到其他的Redis服务器，前者称为 主节点(master/leader)，后者称为 从节点(slave/follower)，数据的复制是单向的。

作用：

1. 数据冗余：实现了数据热备份，是除持久化外的一种数据冗余方式
2. 故障恢复：当主节点出现问题，可以快速由从节点恢复服务
3. 负载均衡：在主从复制上配合读写分离，主节点提供写服务，从节点提供读服务，分担服务器负载
4. 高可用（集群）基石：主从复制是哨兵和集群的基础

mac:0>info replication    # 查看当前库的信息
"# Replication
role:master    # 角色
connected_slaves:0    # 从机数量
master_replid:f62f6712fde7fddf711ae05a1c89996fec9fb57b
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0
"

实现集群，至少要3个Redis服务器，即1主2从，要为每个Redis配置单独的conf文件

假设要配置主机(6379)，从机(6380 6381)，则需要在从机上执行下列命令：

SLAVEOF 127.0.0.1 6379

或者使用配置文件，在REPLICATION模块进行配置

复制原理

slave启动成功连接到master后会发送一个sync同步命令

master接到命令后，启动后台存盘进程，同时收集所有节后到用于修改数据集的命令，进程执行完毕后传送整个数据文件给slave，完成一次完全同步

全量复制：slave服务接收到数据库文件数据后，存盘并加载到内存

增量复制：master继续将收集到新的修改命令依次传给slave，完成同步

重新连接master时会执行一次全量复制

主机宕机后手动配置从机为主机

SLAVEOF no one

哨兵模式

Redis从2.8开始提供Sentinel（哨兵模式）

原理：哨兵作为独立进程，向Redis发送命令并等待响应，从而监控运行的多个Redis实例

配置

1. sentinel.conf

# sentinel monitor 被监控名称 地址 端口 数字代表有几个哨兵判断主节点挂了就进行故障转移
sentinel  monitor name host port 1

2. 启动哨兵

redis-sentinel

注意：假如主机6379宕机，选取6380为新的主机，当6379恢复后会被哨兵转为6380的从机

优点：主从可以进行自动切换，系统可用性更高

缺点：在线扩容、配置复杂（详细可以百度Redis哨兵的配置）

Redis缓存穿透与雪崩

缓存穿透

用户想查询一个数据，在Redis数据库中没有（即缓存没有命中），则向持久层数据库查询（数据库也没有命中），当访问量过大会造成巨大压力，例如发起id为-1的请求

布隆过滤器：是一种数据结构，对所有可能查询的参数以hash形式存储

缓存空对象：当存储层不命中，即使返回空对象也缓存起来，并设置过期时间

缓存击穿

注意与穿透的区别，击穿是指并发集中对某个点进行访问，在key失效瞬间直接请求持久层数据库导致的

即穿透是由于查不到导致的，击穿是指访问量太大导致的

解决方案：设置热点数据永不过期、加互斥锁

缓存雪崩

指在某个时间段，缓存集中过期失效

解决方案：异地多活、限流降级、数据预热