Redis

5.0.3

1. 介绍

REmote DIctionary Server(Redis) 是一个由 Salvatore Sanfilippo 写的 key-value 存储系统，是跨平台的非关系型数据库。

Redis 是一个开源的使用 ANSI C 语言编写、遵守 BSD 协议、支持网络、可基于内存、分布式、可选持久性的键值对(Key-Value)存储数据库，并提供多种语言的 API。

Redis 通常被称为数据结构服务器，因为值（value）可以是字符串(String)、哈希(Hash)、列表(list)、集合(sets)和有序集合(sorted sets)等类型。

并且可以对这些类型使用原子操作，比如追加字串，增加值到hash中等。

优点：

性能极高。单台redis情况下，官方提供的数据为： 读的速度是110000次/s，写的速度是81000次/s 。
支持类型多。支持字符串(String)、哈希(Hash)、列表(List)、集合(Sets)和有序集合(Sorted Sets)等类型。
原子性。要么都成功要么都失败。多个操作支持事务。
功能丰富。支持订阅者，通知，DLL等。
读写快。使用自己实现的分离器，代码很短，没有锁，效率高。

缺点：

耗内存。
持久化。Redis直接将数据存储到内存中，要将数据保存到磁盘上，Redis可以使用两种方式实现持久化过程。定时快照（snapshot）：每隔一段时间将整个数据库写到磁盘上，每次均是写全部数据，代价非常高。第二种方式基于语句追加（aof）：只追踪变化的数据，但是追加的log可能过大，同时所有的操作均重新执行一遍，恢复速度慢。

应用：

String：缓存、限流、计数器、分布式锁、分布式Session
Hash：存储用户信息、用户主页访问量、组合查询
List：微博关注人时间轴列表、简单队列
Set：赞、踩、标签、好友关系
Zset：排行榜

2. 安装

CentOS安装

yum install redis -y
systemctl start redis
redis-cli -h 127.0.0.1 -p 6379

127.0.0.1:6379> ping
PONG

# 启动
redis-server /etc/redis.conf

Docker安装

docker pull redis
docker run -d -p 6379:6379 --name redis redis

3. 实现

3.1. SDS

// sds.h/sdshdr
/*
 * 保存字符串对象的结构
 */
struct sdshdr {
    # 记录buf数组中已经使用字节的数量
    # buf 中已占用空间的长度
    int len;
    # buf 中剩余可用空间的长度
    int free;
    # 字节数组
    char buf[];
};

在C字符串的长度获取中，需要遍历字符串获取。在SDS中，用len记录其长度，获取长度的复杂度为O(1).

在C的字串拼接strcat(str1, str2)中，如未对str1分配足够的空间，会造成str1的内容溢出，造成str1地址后面的值被修改。SDS空间分配策略会先检查空间是否满足，如不满足会先进行扩容。无需关心SDS空间大小，也不用担心内存溢出的问题。

/*
 * 将给定字符串 t 追加到 sds 的末尾
 * 
 * 返回值
 *  sds ：追加成功返回新 sds ，失败返回 NULL
 *
 * 复杂度
 *  T = O(N)
 */
/* Append the specified null termianted C string to the sds string 's'.
 *
 * After the call, the passed sds string is no longer valid and all the
 * references must be substituted with the new pointer returned by the call. */
sds sdscat(sds s, const char *t) {
    return sdscatlen(s, t, strlen(t));
}

/*
 * 将长度为 len 的字符串 t 追加到 sds 的字符串末尾
 *
 * 返回值
 *  sds ：追加成功返回新 sds ，失败返回 NULL
 *
 * 复杂度
 *  T = O(N)
 */
/* Append the specified binary-safe string pointed by 't' of 'len' bytes to the
 * end of the specified sds string 's'.
 *
 * After the call, the passed sds string is no longer valid and all the
 * references must be substituted with the new pointer returned by the call. */
sds sdscatlen(sds s, const void *t, size_t len) {

    struct sdshdr *sh;

    # 原有字符串长度
    size_t curlen = sdslen(s);

    # 扩展 sds 空间
    # T = O(N)
    s = sdsMakeRoomFor(s,len);

    # 内存不足？直接返回
    if (s == NULL) return NULL;

    # 复制 t 中的内容到字符串后部
    # T = O(N)
    sh = (void*) (s-(sizeof(struct sdshdr)));
    memcpy(s+curlen, t, len);

    # 更新属性
    sh->len = curlen+len;
    sh->free = sh->free-len;

    # 添加新结尾符号
    s[curlen+len] = '\0';

    # 返回新 sds
    return s;
}

在C中，每次增长或者缩短一个C字符串，程序都总要对保存这个C字符串的数组进行一次内存重分配操作：

如果程序执行的是增长字符串的操作，比如拼接操作（append），那么在执行这个操作之前，程序需要先通过内存重分配来扩展底层数组的空间大小——如果忘了这一步就会产生缓冲区溢出。
如果程序执行的是缩短字符串的操作，比如截断操作（trim），那么在执行这个操作之后，程序需要通过内存重分配来释放字符串不再使用的那部分空间——如果忘了这一步就会产生内存泄漏。

SDS通过未使用空间解除了字符串长度和底层数组长度之间的关联：在SDS中，buf数组的长度不一定就是字符数量加一，数组里面可以包含未使用的字节，而这些字节的数量就由SDS的free属性记录。通过未使用空间，SDS实现了空间预分配（如果剩余未使用空间不足以存放新数据，在进行空间扩展时也会为SDS分配额外的未使用空间。如果修改后SDS长度小于1MB，则分配和len属性同样大小的未使用空间，这时len=free；如果大于等于1MB，则会分配1MB的未使用空间）和惰性空间释放（程序不立即使用内存重分配来回收缩短后多出来的字节，而是使用free属性将字节的数量记录并等待将来使用，或者在有需要的时候真正地释放SDS的未使用空间）两种优化策略。

SDS的API都是二进制安全的（binary-safe），所有SDS API都会以处理二进制的方式来处理SDS存放在buf数组里的数据，程序不会对其中的数据做任何限制、过滤、或者假设，数据在写入时是什么样的，它被读取时就是什么样。Redis不是用buf数组来保存字符，而是用它来保存一系列二进制数据。所以不仅可以保存文本，也能保存任何格式的二进制数据。

3.2. 链表

Redis的链表实现采用双端链表结构。

// adlist.h
/*
 * 双端链表结构
 */
typedef struct list {

    // 表头节点
    listNode *head;

    // 表尾节点
    listNode *tail;

    // 节点值复制函数
    void *(*dup)(void *ptr);

    // 节点值释放函数
    void (*free)(void *ptr);

    // 节点值对比函数
    int (*match)(void *ptr, void *key);

    // 链表所包含的节点数量
    unsigned long len;

} list;

无环，即表头结点prev指针和表尾结点next指针都指向NULL。

获取表头和表尾指针和元素、获取链表长度的时间复杂度都为O(1).

可以用于保存不同类型的值。

使用广泛，List、Set、Hash等也使用链表实现。

// dict.h
/*
 * 哈希表节点
 */
typedef struct dictEntry {

    // 键
    void *key;

    // 值
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
    } v;

    // 指向下个哈希表节点，形成链表
    struct dictEntry *next;

} dictEntry;

/*
 * 哈希表
 *
 * 每个字典都使用两个哈希表，从而实现渐进式 rehash 。
 */
typedef struct dictht {

    // 哈希表数组
    dictEntry **table;

    // 哈希表大小
    unsigned long size;

    // 哈希表大小掩码，用于计算索引值
    // 总是等于 size - 1
    unsigned long sizemask;

    // 该哈希表已有节点的数量
    unsigned long used;

} dictht;

4. 使用

4.1. 基本命令

Redis命令大全

一般，操作的返回时为数字的情况下，0表示失败，非0或者“OK”为成功。

key
    keys * 获取所有的key
    select 0 选择第一个库
    move myString 1 将当前的数据库key移动到某个数据库,目标库有，则不能移动
    flush db      清除指定库
    randomkey     随机key
    type key      类型
    set key1 value1 设置key
    get key1    获取key
    mset key1 value1 key2 value2 key3 value3
    mget key1 key2 key3
    del key1   删除key
    exists key      判断是否存在key
    expire key 10   设置过期时间，10s后过期
    pexpire key 1000 设置过期时间，单位毫秒
    persist key     删除过期时间

string
    set name myname
    get name
    getrange name 0 -1        字符串分段（根据开始和结束的下标获取子串）
    getset name new_myname       设置值，返回旧值
    mset key1 value [key2 value2 ...]            批量设置
    mget key1 key2            批量获取
    setnx key value           不存在就插入（not exists）
    setex key time value      过期时间（expire）单位s
    setrange key index value  从index开始替换value
    # 虽然是递增递减，也支持负数表达，返回值为操作后的结果
    incr age        递增
    incrby age 10   递增，步数10
    decr age        递减
    decrby age 10   递减，步数10
    incrbyfloat     增减浮点数，负数则减
    append          字串追加，返回strlen
    strlen          长度
    getbit/setbit/bitcount/bitop    位操作

hash
    # hset key field value
    hset myhash name myname
    # hget key field
    hget myhash name
    # hset key field value [key1 value1 ...]
    hmset myhash name myname age 25 note "i am notes"
    hmget myhash name age note   

    hgetall myhash               获取key下所有的值
    # 以key，value为一组显示
    127.0.0.1:6379> hgetall myhash
    1) "name"
    2) "myname"
    3) "age"
    4) "25"
    5) "note"
    6) "i am notes"

    hexists myhash name          是否存在
    hsetnx myhash score 100      设置不存在的
    hincrby myhash id 1          增加或减少，若是新的key，则初始值为0
    hdel myhash name             删除
    hkeys myhash                 只取key
    hvals myhash                 只取value
    hlen myhash                  长度

    > hmset user:1 name myname age 12
    OK
    > hget user:1 name
    "myname"

list
    lpush mylist a b c  左插入
    rpush mylist x y z  右插入

    # 插入的时候为逐个元素进行插入
    127.0.0.1:6379> lpush mylist a b c
    (integer) 3
    127.0.0.1:6379> rpush mylist x y z
    (integer) 6
    127.0.0.1:6379> lpush mylist 1 2 3
    (integer) 9
    127.0.0.1:6379> lrange mylist 0 -1
    1) "3"
    2) "2"
    3) "1"
    4) "c"
    5) "b"
    6) "a"
    7) "x"
    8) "y"
    9) "z"

    # list允许重复
    127.0.0.1:6379> lpush mylist 3 3 3
    (integer) 11
    127.0.0.1:6379> lrange mylist 0 -1
     1) "3"
     2) "3"
     3) "3"

    lrange mylist 0 -1  数据集合
    lpop mylist  弹出列表左侧第一个元素
    rpop mylist  弹出列表右侧第一个元素
    llen mylist  列表长度
    lrem mylist count value  删除，根据参数 count 的值，移除列表中与参数 value 相等的元素。
    # count 的值可以是以下几种：
    # count > 0 : 从表头开始向表尾搜索，移除与 value 相等的元素，数量为 count 。
    # count < 0 : 从表尾开始向表头搜索，移除与 value 相等的元素，数量为 count 的绝对值。
    # count = 0 : 移除表中所有与 value 相等的值。

    127.0.0.1:6379> lrem mylist 2 3
    (integer) 2
    127.0.0.1:6379> lrange mylist 0 -1
    1) "3"

    lindex mylist 2          指定索引的值，索引从0开始
    lset mylist 2 n          根据索引更新值，当索引超出范围或者key不存在时返回错误
    ltrim mylist 0 4         删除key。让列表只保留指定区间内的元素，不在指定区间之内的元素都将被删除。
    # 两个参数分别是start stop。
    # 情况 1：start和stop都在正常范围内。此时会正常删除不在范围内的数。
    # 情况 2：stop 比列表的最大下标还要大。此时会正常删除不在范围内的数。
    # 情况 3： start 和 stop 都比列表的最大下标要大，并且 start < stop。此时会清空列表。
    # 情况 4： start 和 stop 都比列表的最大下标要大，并且 start > stop。此时会清空列表。

    linsert mylist before value new  将new插入到值为value之前
    linsert mylist after value new   将new插入到值为value之后
    rpoplpush list list2     在list弹出最后一个元素返回，同时插入到list2的头部

set（无序、唯一）
    sadd myset redis 
    smembers myset       数据集合
    srem myset set1         删除
    sismember myset set1 判断元素是否在集合中（集合不存在也为0）
    scard key_name       返回集合个数（集合不存在也为0）
    sdiff | sinter | sunion 操作：集合间运算：差集 | 交集 | 并集
    srandmember          随机获取集合中的元素
    spop                 从集合中弹出一个随机元素
    smove src dist member    # 将src集合下的成员member移动到dist集合下

zset（sorted set 有序、唯一）

    # 不同的是每个元素都会关联一个 double 类型的分数。redis 正是通过分数来为集合中的成员进行从小到大的排序。
    # 有序集合的成员是唯一的,但分数(score)却可以重复。
    # ZADD key score member [[score member] [score member] ...]
    zadd zset 1 one
    zadd zset 2 two 3 three
    zincrby zset 1 one              增长分数，增量为1
    zscore zset two                 获取分数
    zrange zset 0 -1 withscores     指定下标范围的值
    zrangebyscore zset 10 25 withscores 指定score范围的值

    # ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]
    # min 和 max 表示为 -inf 和 +inf 
    # 默认情况下，区间的取值使用闭区间 (小于等于或大于等于)，你也可以通过给参数前增加 ( 符号来使用可选的开区间 (小于或大于)。
    # ZRANGEBYSCORE zset (1 5
    # 可选的 WITHSCORES 参数决定结果集是单单返回有序集的成员，还是将有序集成员及其 score 值一起返回。
    # limit 偏移量 获取数量
    zrangebyscore zset 10 25 withscores limit 1 2 分页
    zrevrangebyscore zset 10 25 withscores  指定范围的值
    zcard zset  元素数量
    zcount zset 1000 2000 获得指定分数范围内的元素个数
    zrem zset one two        删除一个或多个元素
    zremrangebyrank zset 0 1  按照排名范围删除元素，负数表示删除倒数的元素
    zremrangebyscore zset 0 1 按照分数范围删除元素

    zrank key member   返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递增(从小到大)顺序排列。
    zrevrank key member  返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递减(从大到小)排序。
    redis> ZRANGE salary 0 -1 WITHSCORES        # 显示所有成员及其 score 值
    1) "peter"
    2) "3500"
    3) "tom"
    4) "4000"
    5) "jack"
    6) "5000"

    redis> ZRANK salary tom                     # 显示 tom 的薪水排名，第二
    (integer) 1

    zinterstore
    # ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE SUM|MIN|MAX]
    zunionstore rank:last_week 7 rank:20150323 rank:20150324 rank:20150325  weights 1 1 1 1 1 1 1

排序：
    # 假设正确地设置了环境变量 LC_COLLATE ，Redis可以感知UTF-8编码。
    sort mylist  排序 （默认按照数值从小到大）
    sort mylist alpha desc limit 0 2 字母倒序排序
    sort list by it:* desc           通过外部key排序 by命令
    sort list by it:* desc get it:*  get参数获取外部key
    # sort命令之store参数：表示把sort查询的结果集保存起来
    sort list by it:* desc get it:* store sorc:result

订阅与发布：
    订阅频道：subscribe chat1
    发布消息：publish chat1 "hell0 ni hao"
    查看频道：pubsub channels
    查看某个频道的订阅者数量: pubsub numsub chat1
    退订指定频道： unsubscrible chat1   , punsubscribe java.*
    订阅一组频道： psubscribe java.*  
# 使用场景：实习消息系统、实时聊天、订阅和关注系统，复杂场景下用消息中间件MQ

redis事务：
     隔离性，原子性， 
     步骤：  开始事务muti，执行命令，提交事务exec
             multi  //开启事务
             sadd myset a b c
             sadd myset e f g
             lpush mylist aa bb cc
             lpush mylist dd ff gg
             exec

服务器管理
    dump.rdb
    appendonly.aof
    # BgRewriteAof 异步执行一个aop(appendOnly file)文件重写
    会创建当前一个AOF文件体积的优化版本

    # BgSave 后台异步保存数据到磁盘，会在当前目录下创建文件dump.rdb
    # save同步保存数据到磁盘，会阻塞主进程，别的客户端无法连接

    # client kill 关闭客户端连接
    # client list 列出所有的客户端

    # 给客户端设置一个名称
      client setname myclient1
      client getname

     config get port
     # configRewrite 对redis的配置文件进行改写

rdb
save 900 1
save 300 10
save 60 10000

aop备份处理
appendonly yes 开启持久化
appendfsync everysec 每秒备份一次

命令：
bgsave异步保存数据到磁盘（快照保存）
lastsave返回上次成功保存到磁盘的unix的时间戳
shutdown同步保存到服务器并关闭redis服务器
bgrewriteaof文件压缩处理（命令）

4.2. Bit位操作

Bit位图不是实际的数据类型，而是在String类类型上定义的一组面向位的操作。而String类型的key是二进制格式的，所以任意二进制格式的数据都可以作为key。

offset 参数必须大于或等于 0 ，小于 2^32 (bit 映射被限制在 512 MB 之内)。

用途：聊天工具计算在线人数。

和布隆过滤器使用的二进制向量类似，用1和0表示两种不同的状态，其下标则可以定义成需要的特殊的含义。

值得注意的是，offset从0开始，且从左往右计数。如01100001，左边第一个数offset是0，右边第一个数的offset是7.

使用：

# 使用set命令设置一个字符串，以二进制方式存储
set sbit AB
# A 0    1    0    0    0    0    0    1    
# B 0    1    0    0    0    0    1    0
# 最后 0100000101000010

# 观察可知，offset从左往右计数
127.0.0.1:6379> getbit sbit 0
(integer) 0
127.0.0.1:6379> getbit sbit 1
(integer) 1
127.0.0.1:6379> getbit sbit 7
(integer) 1
127.0.0.1:6379> getbit sbit 8
(integer) 0
127.0.0.1:6379> getbit sbit 9
(integer) 1
127.0.0.1:6379> getbit sbit 15
(integer) 0
127.0.0.1:6379> getbit sbit 14
(integer) 1
# offset超过时，统一为0
127.0.0.1:6379> getbit sbit 16
(integer) 0
# offset需大于0
127.0.0.1:6379> getbit sbit -1
(error) ERR bit offset is not an integer or out of range
127.0.0.1:6379>

# 通过setbit对原来的位图下的某一个位进行修改
# setbit key offset value
127.0.0.1:6379> setbit sbit 0 1
# 返回的值原来的值，同时修改这个值
(integer) 0
# getbit获取值
127.0.0.1:6379> getbit sbit 0
(integer) 1

# 可以对超过offset的位进行修改，会进行扩容。
127.0.0.1:6379> setbit sbit 16 1
(integer) 0
127.0.0.1:6379> getbit sbit 16
(integer) 1

# 计算位图中1的数量
127.0.0.1:6379> bitcount sbit
(integer) 4

# 获取二进制位串中第一个0或1的位置
127.0.0.1:6379> bitpos sbit 0
(integer) 0
127.0.0.1:6379> bitpos sbit 1
(integer) 1

# 位操作
# bitop operation destkey key [key ...]
# 在不同的字符串之间执行按位运算。提供的运算为AND，OR，XOR和NOT
# BITOP AND destkey key [key ...] ，对一个或多个 key 求逻辑并，并将结果保存到 destkey 。
# BITOP OR destkey key [key ...] ，对一个或多个 key 求逻辑或，并将结果保存到 destkey 。
# BITOP XOR destkey key [key ...] ，对一个或多个 key 求逻辑异或，并将结果保存到 destkey 。
# BITOP NOT destkey key ，对给定 key 求逻辑非，并将结果保存到 destkey 。
# 当 BITOP 处理不同长度的字符串时，较短的那个字符串所缺少的部分会被看作 0 。
# 空的 key 也被看作是包含 0 的字符串序列。
# 保存到 destkey 的字符串的长度，和输入 key 中最长的字符串长度相等。
127.0.0.1:6379> mset s1 ABC s2 CBA
OK
127.0.0.1:6379> bitop and res s1 s2
(integer) 3
127.0.0.1:6379> get res
"ABA"

# 当and, or, xor只有一个key的时候，相当于赋值操作。
127.0.0.1:6379> bitop and and s1
(integer) 3
127.0.0.1:6379> bitop or or s1
(integer) 3
127.0.0.1:6379> bitop xor xor s1
(integer) 3
127.0.0.1:6379> mget and or xor
1) "ABC"
2) "ABC"
3) "ABC"

127.0.0.1:6379> set s A
OK
127.0.0.1:6379> bitop not ss s
(integer) 1
127.0.0.1:6379> get ss
"\xbe"

# A 01000001    41
#   10111110    be

4.3. 特殊数据类型

Geospatial

定位、附近的人、打车距离、两地距离的计算。

有效的经度从-180度到 180 度。

有效的纬度从-85.05112878度到85.05112878度。

一般会下载城市数据一次性导入。

Redis 地理位置(geo) 命令

命令	描述
Redis GEOHASH 命令	返回一个或多个位置元素的 Geohash 表示
Redis GEOPOS 命令	从key里返回所有给定位置元素的位置（经度和纬度）
Redis GEODIST 命令	返回两个给定位置之间的距离
Redis GEORADIUS 命令	以给定的经纬度为中心，找出某一半径内的元素
Redis GEOADD 命令	将指定的地理空间位置（经度、纬度、名称）添加到指定的key中
Redis GEORADIUSBYMEMBER 命令	找出位于指定范围内的元素，中心点是由给定的位置元素决定

127.0.0.1:6379> geoadd china:city 116.40 39.90 beijing
(integer) 1

127.0.0.1:6379> geopos china:city beijing  # 获取指定的城市的经度和纬度！
1 ) 1 ) "116.39999896287918091"
    2 ) "39.90000009167092543"

单位：

m 表示单位为米 (默认单位)
km 表示单位为千米
mi 表示单位为英里
ft 表示单位为英尺
127.0.0.1:6379> geodist china:city beijing shanghai km  # 查看上海到北京的直线距离
"1067.3788"

GEORADIUS key 精度 纬度 距离 [可选项]
withdist 显示距离
withcoord 显示坐标
count x 显示最近的x条记录
127.0.0.1:6379> GEORADIUS china:city 110 30 500 km withdist withcoord count 2
1) 1) "chongqing"
   2) "341.9374"
   3) 1) "106.49999767541885376"
      2) "29.52999957900659211"
2) 1) "xian"
   2) "483.8340"
   3) 1) "108.96000176668167114"
      2) "34.25999964418929977"

georadiusbymember key k1 dist unit
127.0.0.1:6379> georadiusbymember china:city beijing 1000 km
1 ) "beijing"
2 ) "xian"

将二维的经纬度转换为一维的字符串
127.0.0.1:6379> geohash china:city beijing chongqing
1 ) "wx4fbxxfke0"
2 ) "wm5xzrybty0"

Geospatial基于ZSet封装，可以使用ZSet命令来操作。

127.0.0.1:6379> zrange china:city 0 -1 # 查看地图中全部的元素
127.0.0.1:6379> zrem china:city beijing  # 移除指定元素！

Hyperloglog

用途：基数统计，如网站用户统计（UV）。（允许误差的情况下）

优势：占用内存固定且小。

错误率0.81%

127.0.0.1:6379> pfadd mykey a b c d e f g h i j # 创建第一组元素 mykey
(integer) 1
127.0.0.1:6379> pfcount mykey  # 统计 mykey 元素的基数数量
(integer) 10
127.0.0.1:6379> pfadd mykey2 i j z x c v b n m # 创建第二组元素 mykey2
(integer) 1
127.0.0.1:6379> pfcount mykey2
(integer) 9
127.0.0.1:6379> pfmerge mykey3 mykey mykey2  # 合并两组 mykey mykey2 => mykey3 并集
OK
127.0.0.1:6379> pfcount mykey3  # 看并集的数量
(integer) 15

BItmap

用途：统计用户活跃信息（签到、打卡等只有两个状态的情况）。

# 使用bitmap 来记录 周一到周日的打卡！
# 周一： 1 周二： 0 周三： 1 周四：0 ......
127.0.0.1:6379> setbit sign 0 1
(integer) 0
127.0.0.1:6379> setbit sign 1 0
(integer) 0
127.0.0.1:6379> setbit sign 2 1
(integer) 0
127.0.0.1:6379> setbit sign 3 0
(integer) 0
127.0.0.1:6379> setbit sign 4 0
(integer) 0
127.0.0.1:6379> setbit sign 5 1
(integer) 0
127.0.0.1:6379> setbit sign 6 0
(integer) 0
# 查看某一天是否有打卡！
127.0.0.1:6379> getbit sign 5
(integer) 1
127.0.0.1:6379> getbit sign 6
(integer) 0
# 统计操作，统计 打卡的天数！
127.0.0.1:6379> bitcount sign  
(integer) 3

4.4. 事务

Redis的事务本质上是一组命令的集合，所有命令会被序列化。在事务执行过程中会按照顺序执行。

只有一次性、顺序性、排他性。

单条语句保证原子性，但事务不保证原子性。

Redis事务的方法：

开启事务（multi）
命令入队（...）
执行（exec）或放弃事务（discard）

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> set k3 v3
QUEUED
127.0.0.1:6379> exec  # 执行事务
1 ) OK
2 ) OK
3 ) "v2"
4 ) OK

事务出错分为两种：

在插入命令的时候能识别命令存在错误，则事务中所有命令都不会被执行。
如果命令正确但运行时错误，那么其他的命令也会正常执行。错误的命令将报错。

乐观锁Watch

乐观锁认为什么时候都不会出问题，数据修改时会判断version，以判断期间是否被更改过数据。

使用：

127.0.0.1:6379> watch money # 监视 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
# 执行之前，如果另外一个线程修改了money的值，这个时候，就会导致事务执行失败，返回nil。
127.0.0.1:6379> exec
(nil)

如果事务执行失败，需要用unwatch后重新执行watch再执行事务。

4.5. Jedis

Jedis是Redis官方推荐的Java操作Redis的中间件。

依赖

<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.2.3</version>
</dependency>

编码

过程：

连接数据库
命令操作
断开连接

public class RedisPing {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1",6379);
        System.out.println(jedis.ping());
        jedis.close();  //关闭连接
       // jedis.shutdown();  //关闭服务
    }
}

命令和命令行操作Redis相同jedis.(command)。

import com.alibaba.fastjson.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

public class RedisPing {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1",6379);
        jedis.flushDB(); // 清除数据
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("name", "rootwhois");
        jsonObject.put("age", 22);
        String jsonString = jsonObject.toJSONString();
        Transaction multi = jedis.multi();// 开启事务
        try {
            multi.set("key1", jsonString);
            multi.set("key2", jsonString);
            // int i = 1 / 0;    // 代码抛出异常事务，执行失败！
            multi.exec();  // 执行事务
        } catch (Exception e) {
            multi.discard(); // 出现异常，停止事务
            e.printStackTrace();
        } finally {
            System.out.println(jedis.get("key1"));
            System.out.println(jedis.get("key2"));
            jedis.close(); // 关闭连接
        }

    }
}

4.6. Spring Boot整合

在Spring Boot2.x之后，将Jedis替换成了Lettuce。

Jedis：采用直连，多个线程操作时不安全。解决方法：使用Jedis Pool连接池。更像BIO模式。

Lettuce：采用Netty。实例可以在多个线程中进行共享，不存在线程不安全的情况。更像NIO模式。

@Bean
@ConditionalOnMissingBean(name = "redisTemplate") 
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory                                                         isConnectionFactory)
throws UnknownHostException {
    // 默认的 RedisTemplate 没有过多的设置，redis 对象都是需要序列化！
    // 两个泛型都是 Object, Object 的类型，后使用需要强制转换 <String, Object>
    RedisTemplate<Object, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(redisConnectionFactory);
    return template;
}
@Bean
@ConditionalOnMissingBean 
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory
                                               redisConnectionFactory)
throws UnknownHostException {
    StringRedisTemplate template = new StringRedisTemplate();
    template.setConnectionFactory(redisConnectionFactory);
    return template;
}

导依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

连接配置

spring.redis.host=127.0.0.1
spring.redis.port= 6379

使用

@SpringBootTest
class Test {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    void contextLoads() {
        // redisTemplate 操作不同的数据类型，api和命令行操作是一样的
        // opsForValue 操作字符串 类似String ，位图在这个里面
        // opsForList 操作List 类似List
        // opsForSet
        // opsForHash
        // opsForZSet
        // opsForGeo
        // opsForHyperLogLog
        // 除了基本的操作，常用的方法都可以直接通过redisTemplate操作，比如事务，和基本的CRUD
        // 如果对同一个key做多个操作，可以通过bound...Ops绑定key，简化操作，如：
        BoundHashOperations<String, String, User> template = redisTemplate.boundHashOps(KEY);

        // 获取redis的连接对象
        RedisConnection connection =  redisTemplate.getConnectionFactory().getConnection();
        connection.flushDb();
        connection.flushAll();

        redisTemplate.opsForValue().set("mykey","myValue");
        System.out.println(redisTemplate.opsForValue().get("mykey"));
    }
}

自定义RedisTemplate

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory
                                                       factory) {
        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;
    }
}

一般的Redis操作都会封装成一个工具类，注入调用。

4.7. Redis.conf配置文件

启动通过配置文件启动。

配置文件对大小写不敏感。

# 带配置启动redis
./redis-server /path/to/redis.conf

redis.conf部分配置

# 在配置文件中包含其他的配置文件
include /path/to/local.conf
include /path/to/other.conf

# 网络
bind 127 .0.0.1  # 绑定的ip
protected-mode yes # 保护模式
port 6379 # 端口设置

# 通用配置
daemonize yes # 以守护进程的方式后台运行，默认是 no

pidfile /var/run/redis_6379.pid  # 如果以后台的方式运行，需要指定一个 pid 文件

# 日志
# Specify the server verbosity level.
# This can be one of:
# debug (a lot of information, useful for development/testing)
# verbose (many rarely useful info, but not a mess like the debug level)
# notice (moderately verbose, what you want in production probably)
# warning (only very important / critical messages are logged)
loglevel notice

logfile "" # 日志的文件位置名

databases 16 # 数据库的数量，默认是 16 个数据库

always-show-logo yes # 是否总是显示LOGO

# 快照
持久化问题，在规定的时间内，执行了多少次操作就会持久化到文件中。.rdb和.aof
redis是内存数据库，如果没有持久化，数据断电即失。

# 如果900s内，如果至少有一个1 key进行了修改，进行持久化操作
save 900 1
save 300 10
save 60 10000

stop-writes-on-bgsave-error yes # 持久化如果出错，是否还需要继续工作
rdbcompression yes # 是否压缩 rdb 文件，需要消耗一些cpu资源
rdbchecksum yes # 保存rdb文件的时候，进行错误的检查校验
dir ./  # rdb 文件保存的目录
dbfilename dump.rdb # rdb文件默认文件名为dump

# REPLICATION主从复制
replicaof <masterip> <masterport>
masterauth <master-password>

# 安全
127 .0.0.1:6379> config get requirepass # 获取redis的密码
1 ) "requirepass"
2 ) ""
127 .0.0.1:6379> config set requirepass "123456" # 设置redis的密码
OK
127 .0.0.1:6379> config get requirepass # 连接失败
(error) NOAUTH Authentication required.
127 .0.0.1:6379> auth 123456 # 使用密码进行登录
OK
127 .0.0.1:6379> config get requirepass
1 ) "requirepass"
2 ) "123456"

# 客户端设置
maxclients 10000 # 设置能连接上redis的最大客户端的数量
maxmemory <bytes>  # redis 配置最大的内存容量
maxmemory-policy noeviction  # 内存到达上限之后的处理策略
 1.volatile-lru：只对设置了过期时间的key进行LRU（默认值）
 2.allkeys-lru ： 删除lru算法的key
 3.volatile-random：随机删除即将过期key
 4.allkeys-random：随机删除
 5.volatile-ttl ： 删除即将过期的
 6.noeviction ： 永不过期，返回错误

# APPEND ONLY 模式  aof配置
appendonly no  # 默认是不开启aof模式的，默认是使用rdb方式持久化的
appendfilename "appendonly.aof" # 持久化的文件的名字

# appendfsync always  # 每次修改都会 sync。消耗性能
appendfsync everysec  # 每秒执行一次 sync，可能会丢失这1s的数据
# appendfsync no      # 不执行 sync

no-appendfsync-on-rewrite no

auto-aof-rewrite-percentage 100
# 如果aof文件大于64m，太大的话，会fork一个新的进程将文件进行重写。
auto-aof-rewrite-min-size 64mb

4.8. Redis持久化

Redis是内存数据库，如果不将内存中的数据库状态保存在磁盘，一旦服务器进程退出，服务器中的数据库状态也会消失。所以Redis提供了持久化的功能。

RDB（Redis DataBase）

在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是Snapshot快照，它恢复时是将快照文件直接读到内存里。

Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。整个过程中，主进程是不进行任何IO操作的。这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那 RDB 方式要比 AOF 方式更加的高效。RDB 的缺点是最后一次持久化后的数据可能丢失。默认的就是 RDB，一般情况下不需要修改这个配置！

rdb保存的文件是dump.rdb。都是在配置文件中快照中进行配置的！

rdb备份的触发机制：

save的规则满足的情况下，会自动触发rdb规则
执行 flushall 命令，也会触发rdb规则！
退出redis，也会产生 rdb 文件！

如何恢复rdb文件？

127 .0.0.1:6379> config get dir
1 ) "dir"
2 ) "/usr/local/bin" # 如果在这个目录下存在 dump.rdb 文件，启动就会自动恢复其中的数据

只需要将dump.rdb文件拷贝到上述文件夹下即可。

优点：

适合大规模的数据回复。
对数据完整性要求不高的情况下使用。效率较aof高。

缺点：

需要有一定的时间间隔操作，如果redis意外宕机则最后一次修改的数据可能丢失。
fork进程的时候会占用一定的内存空间。

AOF（Append Only File）

将所有的写操作命令都记录下来（history日志），恢复的时候会将这个文件所有命令重新执行一次。

以日志的形式来记录每个写操作，将Redis执行过的所有指令记录下来（读操作不记录），只许追加文件，但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

AOF保存的是 appendonly.aof 文件。

默认是不开启的，需要手动在配置文件中将appendonly改为yes后重启生效。

如果aof文件有错误，则redis无法启动，需要手动修复这个aof文件。

redis提供了一个工具redis-check-aof，位于bin文件夹中。

redis-check-aof --fix appendonly.aof

如果文件修复成功，提示Successfully truncated AOF。重启Redis即可。

aof会无限追加，文件会越来越大。

优点：

每一次修改都同步，文件完整性好。
每同步一次，可能丢失一秒数据。
从不同步的效率最高。

缺点：

文件远大于rdb且恢复速度慢。
运行效率慢。

对比

RDB 持久化方式能够在指定的时间间隔内对数据进行快照存储
AOF 持久化方式记录每次对服务器写的操作，当服务器重启的时候会重新执行这些命令来恢复原始的数据，AOF命令以 Redis 协议追加保存每次写的操作到文件末尾，Redis还能对AOF文件进行后台重写，使得AOF文件的体积不至于过大。
只做缓存，如果你只希望数据在服务器运行的时候存在，也可以不使用任何持久化
同时开启两种持久化方式
在这种情况下，当redis重启的时候会优先载入AOF文件来恢复原始的数据，因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整。
RDB 的数据不实时，同时使用两者时服务器重启也只会找AOF文件，那要不要只使用AOF呢？作者建议不要，因为RDB更适合用于备份数据库（AOF在不断变化不好备份），快速重启，而且不会有AOF可能潜在的Bug，留着作为一个万一的手段。

5 、性能建议

因为RDB文件只用作后备用途，建议只在Slave上持久化RDB文件，而且只要 15 分钟备份一次就够了，只保留 save 900 1 这条规则。
如果Enable AOF ，好处是在最恶劣情况下也只会丢失不超过两秒数据，启动脚本较简单只load自己的AOF文件就可以了，代价一是带来了持续的IO，二是AOF rewrite 的最后将 rewrite 过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要硬盘许可，应该尽量减少AOF rewrite 的频率，AOF重写的基础大小默认值64M太小了，可以设到5G以上，默认超过原大小100%大小重写可以改到适当的数值。
如果不Enable AOF ，仅靠 Master-Slave Repllcation 实现高可用性也可以，能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Master/Slave 同时宕掉，会丢失十几分钟的数据，启动脚本也要比较两个 Master/Slave 中的 RDB文件，载入较新的那个，微博就是这种架构。

5. 主从复制

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

数据的复制是单向的，只能由主节点到从节点。 Master以写为主，Slave 以读为主。

默认情况下每台Redis服务器都是主节点，且一个主节点可以有多个从节点，但是一个从节点只能有一个主节点。

主从复制的作用主要包括：

数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余方式。
故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复；实际上是一种服务的冗余。
负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（即写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），分担服务器负载；尤其是在写少读多的场景下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。
高可用（集群）基石：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是Redis高可用的基础。

一般来说，要将Redis运用于工程项目中，只使用一台Redis是万万不能的（宕机），原因如下：

从结构上，单个Redis服务器会发生单点故障，并且一台服务器需要处理所有的请求负载，压力较大；
从容量上，单个Redis服务器内存容量有限，就算一台Redis服务器内存容量为256G，也不能将所有内存用作Redis存储内存，一般来说，单台Redis最大使用内存不应该超过20G。

电商网站上的商品，一般都是一次上传，无数次浏览的，也就是"多读少写"。可以使用如下这种架构：

主从复制，读写分离。80% 的情况下都是在进行读操作。

减缓服务器的压力。架构中经常使用一主二从的结构。

集群配置有两种方法：命令行配置和配置文件配置。

5.1. 配置

命令行配置

查看当前库的信息

127 .0.0.1:6379> info replication # 查看当前库的信息
# Replication
role:master  # 角色 master
connected_slaves:0 # 没有从机
master_replid:b63c90e6c501143759cb0e7f450bd1eb0c70882a
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

slaveof命令指定宿主机。命令格式：slaveof <server> <port>

127.0.0.1:6380> slaveof 127.0.0.1 6379  # SLAVEOF host 6379 找谁当自己的老大！
OK
127.0.0.1:6380> info replication
# Replication
role:slave   # 当前角色是从机
master_host:127.0.0.1   # 可以的看到主机的信息
master_port:6379
master_link_status:up
master_last_io_seconds_ago:5
master_sync_in_progress:0
slave_repl_offset:14
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:c63801d05f947eb101e2d46cfd8209a41653d49b
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:14
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:14

此时宿主机上使用info replication命令：

#  主机信息
127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:1  
slave0:ip=127.0.0.1,port=6380,state=online,offset=112,lag=1  # 可以看到从机信息
master_replid:c63801d05f947eb101e2d46cfd8209a41653d49b
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:112
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:112

如需取消宿主机的集群配置，使用slaveof no one命令即可。

配置文件配置

在从机的conf配置文件中，在replication中添加两行连接配置：

replicaof <masterip> <masterport>
masterauth <master-password>

5.2. 说明

从机只读不写

主从复制设置完成并重启后，宿主机可以写，从机只读不能写。宿主机中所有信息和数据都会自动被从机保存。

宿主机：

127.0.0.1:6379> set k1 v1
OK
127.0.0.1:6379> get k1
"v1"

从机：

127.0.0.1:6380> keys *
1) "k1"
127.0.0.1:6380> get k1
"v1"
127.0.0.1:6380> set k2 v2
(error) READONLY You can't write against a read only replica.

主机断开连接，从机依旧连接到主机的，但是没有写操作，这个时候，主机如果回来了，从机依旧可以直接获取到主机写的信息！

如果是使用命令行来配置的主从，这个时候如果重启了，就会变回主机。只要变为从机，立马就会从主机中获取值。

复制原理

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

Master 接到命令，启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令，在后台进程执行完毕之后，master将传送整个数据文件到slave，并完成一次完全同步（全量复制）。

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

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

但是只要是重新连接master，一次完全同步（全量复制）将被自动执行！数据一定可以在从机中看到！

过程：（重新）连接->全量复制->增量复制

层层链路

层层链路指的是上一个Master连接下一个Slave的情况。即中间的Redis既是宿主机又是从机。

这个时候也可以完成主从复制。

启动多个Redis

同一服务器下启动多个Redis的方法：复制多个conf配置文件，修改对应信息。

需要修改的信息：端口、pid文件名、log文件名、dump.rdb文件名。

修改完毕后通过指定配置文件启动即可。

查看redis服务器运行状态的指令：ps -ef | grep redis.

5.3. 哨兵模式

主从切换技术的方法是：当主服务器宕机后，需要手动把一台从服务器切换为主服务器，这就需要人工干预，费事费力，还会造成一段时间内服务不可用。这不是一种推荐的方式，更多时候，优先考虑哨兵模式。Redis从2.8开始正式提供了Sentinel（哨兵）架构来解决这个问题。能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库。

哨兵模式是一种特殊的模式，首先Redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程，它会独立运行。其原理是 哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。

这里的哨兵有两个作用

通过发送命令，让Redis服务器执行并返回来监控其运行状态，包括主服务器和从服务器。
当哨兵监测到master宕机，会自动将slave切换成master，然后通过发布订阅模式通知其他的从服务器，修改配置文件，让它们切换主机。

一个哨兵进程对Redis服务器进行监控，可能会出现问题，为此，可以使用多个哨兵进行监控。各个哨兵之间还会进行监控，这样就形成了多哨兵模式。

假设主服务器宕机，哨兵 1 先检测到这个结果，系统并不会马上进行failover过程，仅仅是哨兵 1 主观的认为主服务器不可用，这个现象成为 主观下线 。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover[故障转移]操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为客观下线 。

配置

这里配置的是单哨兵模式的最简单的配置方式。

配置哨兵配置文件sentinel.conf.

# sentinel monitor 被监控的名称 host port 1
sentinel monitor myredis 127.0.0.1 6379 1

启动哨兵。

[root@localhost bin]# redis-sentinel redisConfig/sentinel.conf

如果Master节点断开了，这个时候就会从从机中随机选择一个服务器当作新的Master节点。如果原来的Master节点回来了，只能归并到新的Master节点下，当做Slave。

优点：

哨兵集群，基于主从复制模式。所以拥有主从复制模式的优点。
主从可以切换，故障可以转移，系统的可用性就会更好。
哨兵模式可以自动切换，使得程序更为健壮。

缺点：

Redis 在线扩容较难，集群容量一旦到达上限，在线扩容就比较麻烦！
实现哨兵模式的配置较为麻烦。

哨兵模式的完整配置文件：

# Example sentinel.conf
# 哨兵sentinel实例运行的端口 默认 26379
port 26379    # 多个哨兵集群就得配置端口

# 哨兵sentinel的工作目录
dir /tmp

# 哨兵sentinel监控的redis主节点的 ip port
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 配置多少个sentinel哨兵统一认为master主节点失联 那么这时客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127 .0.0.1 6379 2

# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都要提供密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster 123456

# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认 30 秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000

# 这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行 同步，这个数字越小，完成failover所需的时间就越长，但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1

# 故障转移的超时时间 failover-timeout 可以用在以下这些方面：
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。
#4.当进行failover时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超时，slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000

# SCRIPTS EXECUTION
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回 1 ，那么该脚本稍后将会被再次执行，重复次数目前默认为 10
#若脚本执行后返回 2 ，或者比 2 更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为 1 时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
#通知脚本
# shell编程
# sentinel notification-script <master-name> <script-path>
sentinel notification-script mymaster /var/redis/notify.sh

# 客户端重新配置主节点参数脚本
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。
# 以下参数将会在调用脚本时传给脚本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>总是“failover”,
# <role>是“leader”或者“observer”中的一个。
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的
# 这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh # 一般都是由运维来配置！

6. 插件

6.1. 布隆过滤器

cd /opt
wget https://github.com/RedisBloom/RedisBloom/archive/refs/tags/v2.2.14.zip
unzip v2.2.14.zip
cd /opt/RedisBloom-2.2.14
yum -y install gcc automake autoconf libtool make
make

vim /etc/redis.conf
/MODULES
+ loadmodule /opt/RedisBloom-2.2.14/redisbloom.so
:wq

# 关闭redis，传入配置文件重启
redis-server /etc/redis.conf
redis-cli 

bf.add key value

注意： key : 布隆过滤器的名称，item : 添加的元素。

BF.ADD：将元素添加到布隆过滤器中，如果该过滤器尚不存在，则创建该过滤器。格式：BF.ADD {key} {item}。
BF.MADD : 将一个或多个元素添加到“布隆过滤器”中，并创建一个尚不存在的过滤器。该命令的操作方式BF.ADD与之相同，只不过它允许多个输入并返回多个值。格式：BF.MADD {key} {item} [item ...] 。
BF.EXISTS : 确定元素是否在布隆过滤器中存在。格式：BF.EXISTS {key} {item}。
BF.MEXISTS ：确定一个或者多个元素是否在布隆过滤器中存在格式：BF.MEXISTS {key} {item} [item ...]。

另外， BF. RESERVE 命令需要单独介绍一下：

这个命令的格式如下：

BF. RESERVE {key} {error_rate} {capacity} [EXPANSION expansion] 。

下面简单介绍一下每个参数的具体含义：

key：布隆过滤器的名称
error_rate : 期望的误报率。该值必须介于 0 到 1 之间。例如，对于期望的误报率 0.1％（1000 中为 1），error_rate 应该设置为 0.001。该数字越接近零，则每个项目的内存消耗越大，并且每个操作的 CPU 使用率越高。
capacity: 过滤器的容量。当实际存储的元素个数超过这个值之后，性能将开始下降。实际的降级将取决于超出限制的程度。随着过滤器元素数量呈指数增长，性能将线性下降。

可选参数：

expansion：如果创建了一个新的子过滤器，则其大小将是当前过滤器的大小乘以expansion。默认扩展值为 2。这意味着每个后续子过滤器将是前一个子过滤器的两倍。

7. 其他

7.1. 配置允许远程连接

防火墙放行6379端口。

修改redis.conf并重启

# 允许远程访问
vim /etc/redis.conf
- bind 127.0.0.1
+ bind 0.0.0.0
- protected-mode yes
+ protected-mode no

# 关闭防火墙（慎用，主要还是以放行端口为主，这里作为测试。）
systemctl stop firewalld
systemctl disable firewalld

# 启动
redis-server /etc/redis.conf
# 允许启动后在后台运行，即关闭命令行窗口后仍能运行
daemonize no 改为 daemonize yes

redis-server ../redis.conf

7.2. 缓存雪崩

原本在数据库和系统之间会有一层Redis缓存，在某一个时刻出现大规模的缓存失效的情况（可能是Redis集群挂了等原因），系统的请求会直接打到数据库上，而数据库的压力剧增，最终可能导致数据库宕机，引起一连串连锁反应。

现象：同一时间大规模的Key失效。

可能的原因：

Redis单机或集群宕机。
大量的Key采用相同的过期时间。

措施：

对过期时间可以适当设置一个随机值，降低到期时间的重复率。
采取熔断机制，当请求过多时，进行限流等措施，减小数据库的压力。
提高数据库的容灾能力，可以使用分库分表，读写分离的策略。
搭建Redis集群，提高容灾性。
采取多级缓存架构：nginx缓存+redis缓存+其他缓存（ehcache等）。

7.3. 缓存击穿

某一个高并发集中访问的热点key失效，导致请求全部打到数据库上，数据库压力变大。

原因：某一热点key失效。

措施：

设置该key不过期（需考虑业务）。
采取熔断机制。
加互斥锁。保证每个key同时只有一个线程去查询后段服务，其他线程没有获得分布式锁的权限，因此只需要等待即可。（将高并发的压力转移到了分布式锁，对分布式锁的考验大）

7.4. 缓存穿透

在一般业务下，有在Redis缓存中查询不到对应key的值的情况，这时候可能会去查数据库，如果数据库也没有，返回空，Redis自然也不会保存这个值。如果这个现象被利用，发送大量恶意请求，就可以绕过缓存，将请求直接打到数据库上，给数据库带来压力。

原因：请求一个Redis和数据库中都不存在的key。

措施：

对结果初步判断，对不可能出现的key直接返回错误。
缓存空值结果，并设置较短的过期时间。
监控，设置黑白名单。
使用布隆过滤器。（哈希函数+位图）

7.5. 较为通用的解决方案

redis高可用

这个思想的含义是，既然redis有可能挂掉，那我多增设几台redis，这样一台挂掉之后其他的还可以继续工作，其实就是搭建的集群。（异地多活！）

限流降级

这个解决方案的思想是，在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。

数据预热

数据加热的含义就是在正式部署之前，我先把可能的数据先预先访问一遍，这样部分可能大量访问的数据就会加载到缓存中。在即将发生大并发访问前手动触发加载缓存不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀。

8. 工具类

判断是否为JSON字串。

import com.alibaba.fastjson.JSONObject;
import org.springframework.util.StringUtils;

/**
 * JSON处理工具类
 *
 * @author lieber
 */
public enum JsonUtil {

    /**
     * 实例
     */
    INSTANCE;

    /**
     * json对象字符串开始标记
     */
    private final static String JSON_OBJECT_START = "{";

    /**
     * json对象字符串结束标记
     */
    private final static String JSON_OBJECT_END = "}";

    /**
     * json数组字符串开始标记
     */
    private final static String JSON_ARRAY_START = "[";

    /**
     * json数组字符串结束标记
     */
    private final static String JSON_ARRAY_END = "]";


    /**
     * 判断字符串是否json对象字符串
     *
     * @param val 字符串
     * @return true/false
     */
    public boolean isJsonObj(String val) {
        if (!StringUtils.hasText(val)) {
            return false;
        }
        val = val.trim();
        if (val.startsWith(JSON_OBJECT_START) && val.endsWith(JSON_OBJECT_END)) {
            try {
               JSONObject.parseObject(val);
                return true;
            } catch (Exception e) {
                return false;
            }
        }
        return false;
    }

    /**
     * 判断字符串是否json数组字符串
     *
     * @param val 字符串
     * @return true/false
     */
    public boolean isJsonArr(String val) {
        if (!StringUtils.hasText(val)) {
            return false;
        }
        val = val.trim();
        if (!StringUtils.hasText(val)) {
            return false;
        }
        val = val.trim();
        if (val.startsWith(JSON_ARRAY_START) && val.endsWith(JSON_ARRAY_END)) {
            try {
                JSONObject.parseArray(val);
                return true;
            } catch (Exception e) {
                return false;
            }
        }
        return false;
    }

    /**
     * 判断对象是否是json对象
     *
     * @param obj 待判断对象
     * @return true/false
     */
    public boolean isJsonObj(Object obj) {
        String str = JSONObject.toJSONString(obj);
        return this.isJsonObj(str);
    }

    /**
     * 判断字符串是否json字符串
     *
     * @param str 字符串
     * @return true/false
     */
    public boolean isJson(String str) {
        if (!StringUtils.hasText(str)) {
            return false;
        }
        return this.isJsonObj(str) || this.isJsonArr(str);
    }
}

9. 其他

使用Redisson分布式锁

官网

特性
- 可重入的（使用hash数据结果保存，其中hash key是client id，value是重入次数）
- 不间断的（默认锁的到期时间是30秒，如果没有释放，则当到期时间为20秒时，再延长至30秒）

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.17.5</version>
</dependency>

@Resource
private RedissonClient redissonClient;

//创建锁
RLock lock = redissonClient.getLock("key");

//加锁
lock.lock();
try {
    // 加锁操作的内容
} finally {
    //释放锁
    helloLock.unlock();
}

/**
  * 普通非公平重入锁.
  */
RLock getLock(String name);

/**
  * 同时获取多把锁.
  */
RLock getMultiLock(RLock... locks);

/**
  * RedLock
  */
RLock getRedLock(RLock... locks);

/**
  * 公平锁
  */
RLock getFairLock(String name);

/**
  * 读写锁
  */
RReadWriteLock getReadWriteLock(String name);

默认配置

public Config() {
    this.transportMode = TransportMode.NIO;
    this.lockWatchdogTimeout = 30000L;
    this.checkLockSyncedSlaves = true;
    this.reliableTopicWatchdogTimeout = TimeUnit.MINUTES.toMillis(10L);
    this.keepPubSubOrder = true;
    this.useScriptCache = false;
    this.minCleanUpDelay = 5;
    this.maxCleanUpDelay = 1800;
    this.cleanUpKeysAmount = 100;
    this.nettyHook = new DefaultNettyHook();
    this.useThreadClassLoader = true;
    this.addressResolverGroupFactory = new DnsAddressResolverGroupFactory();
}

10. 遇到的问题

[Flag] 使用redisTemplate.convertAndSend(channel, message);发布消息和使用MessageListener监听消息的时候，观察到消息的顺序不一致的问题，暂时还没有深究。