中文 | English

1. Redis简介

Redis是一个远程内存数据库，具备以下特点：

• 每秒能处理上百万次请求
• 是非关系性数据库，以Key，Value的形式存储数据
• 数据都是存放在内存的
• 可构建Redis集群做到三高：高性能、高可用、高并发
• 单线程：同一个节点同一时间只能处理一个用户命令。（并不是说Redis就只有一个线程，后台还有很多其他任务线程，比如检测数据是否过期等）

Redis提供了5中数据结构：

• String：可以是字符串、整数或浮点数。
• List：链表。
• Set：无需集合，不可重复。
• Hash：Hash表，类似Java中的HashMap。
• ZSet：有序列表，使用分值(score)来控制顺序，越小越靠前。

Redis除了5中数据结构外，还提供了如下额外功能：

• Key自动过期：可以给Key设置过期时间，到时间自动被删除。
• 发布订阅：可以充当简易的MQ用。
• Lua脚本：可以自制Redis命令，实现复杂的功能。一个Lua脚本是一个原子操作。
• 简单的事务：保证一批redis命令以原字操作执行。
• Pipeline：一次发送一批命令，减小网络开销。不过这批命令各自之间互不影响，它们之间不是原子操作，一个命令失败也不影响后面命令的执行。

Redis的典型应用场景：

• 视频/微博等的点赞数：使用Redis的Set来存储点赞的用户ID。点赞就往Set里面加，取消赞就从Set里面删。当视频不再活跃时，就从Redis删除掉，持久化到数据库里。Redis内存占用：假设用户ID为10个字符, 平均一个视频5k点赞数，同时有10w个活跃视频，那占用内存为：10 * 5k * 10w / 1024 / 1024 / 1024 ≈ 4.65G，完全能够支撑。
• 存储用户session：用户登录系统后，将生成的token存到redis中，并设置失效时间。当用户请求数据时验证token是否失效，如果没失效则可以请求数据。若失效，提示用户重新登录。
• 数据缓存：数据一般读多写少，将经常读的数据缓存到Redis中。① 加快用户响应速度；② 减小数据库压力。不过会带来数据不一致问题。
• 分布式锁：使用setex命令尝试设置值，设置成功代表获取锁成功，设置失败表示锁已经被强占，就继续等待即可。设置成功的话，获取锁成功，建议设置个超时时间，避免死锁。当完成任务时，删除key来释放锁。
• 计数器：利用incr <key>命令记录点击次数等

2. Redis常用命令

全局常用命令：

```python
keys *  # 查看所有键。会遍历所有键，生产禁用。使用scan代替。

dbsize  # 查询键总数。时间复杂度O(1)，可放心使用

exists <key>  # 查询key是否存在

# 遍历与[match pattern]匹配的key，从<cursor>开始，遍历[count number]个。
# 例如：scan 3000 time* 1000  # 表示从第3000个key开始，遍历以time开头的key，遍历1000个。
# scan会返回遍历结果和最后一个数据的cursor，这样下次可以从这个位置开始遍历。
scan <cursor> [match pattern] [count number]  
```

String：

```python
get <key>  # 获取数据

set <key> <value>  # 写入数据

incr <key>  # 自增

decr <key>  # 自减
```

List:

```python
rpush <key> <value> [<value>, ...]  # 将一个或多个写入列表最后面

lpush <key> <value> [<value>, ...]  # 将一个或多个写入列表最前面

rpop <key>  # 移除并返回列表最后一个元素

lpop <key>  # 移除并返回列表第一个元素

lindex <key> <offset>  # 返回列表第offset个元素

lrange <key> <start> <end>  # 返回列表中的第start到end个元素，包含start和end
```

Set：

```python
sadd <key> <value> [<value>, ...]  # 增添一个或多个元素到集合中

srem <key> <value> [<value>, ...]  # 从集合移除一个或多个元素

sismember <key> <value>  # 检查value是否是集合中的元素

scard <key>  # 返回集合的元素数量
```

Hash：

```python
hmget <key> <k> [<k>, ...]  # 从hash表中获取一个或多个键的值

hmset <key> <k> <v> [<k> <v> ...]  # 为hash表写入一个或多个k,v

hdel <key> <k> [<k>, ...]  # 删除hash表中的一个或多个键

hlen <key>  # 返回hash表中的键数量
```

ZSet:

```python
zadd <key> <score> <value> [<score> <value>, ...]  # 为有序集合写入一个或多个包含分数的元素

zrem <key> <value> [<value>, ...]  # 从有序集合中移除一个或多个元素

scard <key>  # 返回集合的元素数量
```

3. Java操作Redis

Redis与客户端的交互基于TCP协议。在此基础上，给客户端制定了一套 RESP(REdis Serialization Protocol) 统一协议，简单来说就是对传输的TCP报文增加一些规定的格式，例如：若正常响应，则第一个字符为“+”，错误响应第一个字符为“-”。

Java有许多库可以访问Redis，接下来逐个讲解。

3.1 Jedis客户端

Java语言一般使用Jedis作为与Redis交互的客户端，其提供了基本的Redis命令。

简单使用样例：

```java
Jedis jedis = new Jedis("127.0.0.1", 6379);
jedis.set("hello", "world");
```

生产环境中，一般使用连接池来构建jedis对象，简单使用样例：

```java
GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); // 初始化连接池配置
JedisPool jedisPool = new JedisPool(poolConfig, "127.0.0.1", 6379);
Jedis jedis = jedisPool.getResource();
```

连接池常用配置：

• maxActive: 最大连接数。
• maxIdle：最大空闲连接数。
• minIdle：最小空闲连接数。
• maxWaitMillis：当连接池资源耗尽时，调用者的最大等待时间。
• minEvictableIdleTimeMillis：连接的最小空闲时间。连接空闲超过该时间会被释放。

若Redis是集群，则需要使用JedisCluster来操作Redis。样例代码如下：

```java
Set<HostAndPort> jedisClusterNodes = new HashSet<>();
// 配一个节点即可，不过建议全配上
jedisClusterNodes.add(new HostAndPort("127.0.0.1", 7000));
jedisClusterNodes.add(new HostAndPort("127.0.0.1", 7001));

JedisCluster jedisCluster = new JedisCluster(jedisClusterNodes);

jedisCluster.set("key", "value");
```

3.2 Lettuce客户端

Lettuce提供了全面的Redis操作API。此外，还提供了非阻塞（异步）的命令方式。相较于Jedis，Lettuce功能更全面，性能更高。

Lettuce使用样例：

```java
import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;

// 创建Redis客户端
RedisClient redisClient = RedisClient.create("redis://127.0.0.1:6379");
// 获取Redis连接
StatefulRedisConnection<String, String> connection = redisClient.connect();
// 使用同步方式访问
RedisCommands<String, String> syncCommands = connection.sync();

syncCommands.set("myKey", "Hello, Lettuce!");
String value = syncCommands.get("myKey");

// 关闭连接，管理Redis客户端
connection.close();
redisClient.shutdown();
```

3.3 Redission客户端

相比Jedis和Lettuce，Redission除了提供基础的Redis API外，还为用户封装了一些更高级的特性（例如：分布式锁）

```java
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("myLock");
try {
    // 获取锁
    lock.lock();
    // ... 执行逻辑

} finally {
    // 释放锁
    lock.unlock();
}
```

3.4 SpringBoot中使用RestTemplate

在SpringBoot项目中，可以使用RedisTemplate访问Redis。不过RedisTemplate是对Jedis、Lettuce等的进一步封装，我们可以自由选择基于哪个框架的Redis连接。

```java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class RedisExampleService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public void doSomething() {
        redisTemplate.opsForValue().set("key1", "value1");
        redisTemplate.opsForValue().get("key1");
        redisTemplate.delete("key1");
    }
}
```

若需要更改默认的RedisTemplate的连接，可以配置如下Bean：

```java
@Bean
public RedisConnectionFactory redisConnectionFactory() {
    LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory();
    lettuceConnectionFactory.setHostName("localhost");
    lettuceConnectionFactory.setPort(6379);
    // Additional configuration if needed
    return lettuceConnectionFactory;
}
```

RedisConnectionFactory是一个接口，共有Jedis、Lettuce和Reddision三种实现。

3.5 Jedis/Lettuce/Reddision的比较

	Jedis	Lettuce	Redission
性能	中	高	高
支持异步	×	√	√
上手难度	简单	较难	较难
Redis API	基础API	基础API	基础API+高级特性（例如分布式锁等）
线程安全	×	√	√

4. Redis的典型应用

4.1 SpringBoot使用Redis进行数据缓存

背景：在读多写少的场景下，若每次都访问数据库来获取最新的数据，那么可能导致数据库压力过大而宕机。因此，最好将不经常变化的数据写入缓存，减缓数据库压力。

在SpringBoot项目中，我们可以使用@Cacheable注解很方便的使用缓存。例如：

```java
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;

@Service
public class MyService {

    @Cacheable("myCache") // 缓存的名字
    public MyObject getData(String key) {
        // 正常的业务逻辑
        return new MyObject();
    }
}
```

在getData(...)方法上加入@Cacheable注解后，Spring就会自动写入和使用缓存。若有缓存，则直接返回，无需执行getData(...)方法中的查询逻辑。

SpingBoot使用缓存的详细配置流程如下：

(1) 引入Spring Redis依赖

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

(2) 在配置文件中（application.properties或application.yml）配置redis的相关配置。例如：

```properties
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=yourPassword
```

(3) 在启动类上增加@EnableCaching，开启缓存功能。例如：

```java
@SpringBootApplication
@EnableCaching
public class CachingApplication {

    public static void main(String[] args) {
        SpringApplication.run(CachingApplication.class, args);
    }
}
```

(4) 在响应的增删改查方法中增添相关的注解。Spring提供了@CachePut（增改）、@Cacheable（查）、@CacheEvict（删）三个注解。例如：

```java
@Service
public class MyService {

    private Map<String, String> database = new HashMap<>();

    // 执行业务逻辑，将返回结果写入缓存。
    // 例如：name为Amy，则存入Redis的key为`personCache::Amy`
    @CachePut("personCache") // personCache是缓存的名字
    public String addOrUpdatePerson(String name) {
        database.put(name, "hello, " + name);
        return database.get(name);
    }
    // 从缓存中获取，若没有缓存，则执行方法，然后存入缓存。
    // 若已经有缓存了，则直接返回缓存中的内容。
    @Cacheable("personCache")
    public String getPerson(String name) throws Exception {
        Thread.sleep(3000);
        return database.get(name);
    }

    // 删除缓存
    @CacheEvict("personCache")
    public void deletePerson(String name) {
        database.remove(name);
    }

    // 删除所有的personCache缓存
    @CacheEvict(value = "personCache", allEntries=true)
    public void deleteAllPerson() {
        database.clear();
    }
}
```

此外，Spring还提供了@Caching注解，可以同时组合上述的三个注解，适用于较为复杂的业务方法。

4.2 Redis实现分布式锁

背景：现代业务系统在部署时都是集群部署，若用户在操作时连续点击两次，会导致两个不同机器同时处理一个业务，进而导致数据异常。因此，需要使用分布式锁来控制不同机器之间的并发，而Redis单线程+高吞吐量非常适合用做分布式锁的实现。

Redis实现分布式锁的核心就是setnx <key> <value>命令：当key存在时，设置失败，当key不存在时，设置成功。

实际生产中，不建议自己写分布式锁，应当直接使用Redission的分布式锁

一个分布式锁至少需要有以下三个部分：

• 构建锁：创建锁对象，确定lockKey、lockValue、 timeout等基本属性。注意事项有：
  • key要唯一：用户使用时，锁的key要唯一，避免不同业务出现错误的竞争。不过这不是分布式锁的职责，需要用户来避免。
• 获取锁：尝试获取锁。若获取不到，则返回获取失败，或自旋等待。注意事项：
  • 最好有超时时间：锁最好要有超时时间，避免持有锁的机器宕机，导致锁释放不掉。
  • 原子操作：setnx操作（写入锁）和expire操作（设置过期时间）要保证是原子操作。否则，若setnx后宕机，锁还是会释放不掉。可以使用pipeline或lua实现两个命令的原子性。
  • 失败或自旋：当获取锁失败后，应当让业务选择是自旋尝试，还是直接返回。
• 释放锁：删除lockKey

使用RedisTemplate实现一个简单的分布式锁代码如下：

```java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.time.Duration;

@Component
public class RedisDistributedLockBuilder {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    // 构建锁。可以重写多个来简化这里
    public Lock build(String lockKey,
                      String lockValue,
                      boolean spin,
                      long timeout,
                      long waitTime
    ) {
        Lock lock = new Lock();
        lock.redisTemplate = redisTemplate;
        lock.lockKey = lockKey;
        lock.lockValue = lockValue;
        lock.spin = spin;
        lock.timeout = timeout;
        lock.waitTime = waitTime;

        return lock;
    }

    // 锁类
    public static class Lock {
        private RedisTemplate redisTemplate;
        private String lockKey;  // Redis的key
        private String lockValue;  // Redis的value，一般用“1”就行
        private boolean spin;  // 是否自旋等待
        private long timeout; // 锁的超时时间（毫秒）
        private long waitTime; // // 自旋时每次等待的时间（毫秒）

        // 获取锁
        public boolean acquireLock() throws InterruptedException {
            while (true) {
                // 尝试获取锁，该方法相当于"setnx+expire"命令，且使用管道保证了原子性
                boolean result = redisTemplate.opsForValue()
                        .setIfAbsent(lockKey, lockValue, Duration.ofMillis(timeout));

                if (!spin) {
                    // 若无需自旋等待，则返回获取结果。
                    return result;
                }

                if (result) {
                    // 若获取成功，则返回
                    return result;
                }

                // 未获取成功，等待waitTime毫秒后，再次尝试获取
                Thread.sleep(waitTime);
            }
        }

        // 释放锁
        public void releaseLock() {
            redisTemplate.delete(lockKey);
        }
    }
}
```

使用样例：

```java
@RestController
@RequestMapping("test")
public class TestController {

    @Autowired
    private RedisDistributedLockBuilder distributedLockBuilder;

    @RequestMapping("/createOrder")
    public void createOrder(@RequestParam String orderCode) throws Exception {
        RedisDistributedLockBuilder.Lock lock = distributedLockBuilder.build("lock_createOrder_" + orderCode, "1",
                false,  // 不自旋等待，直接返回失败
                10 * 1000, 0);

        boolean result = lock.acquireLock();
        if (!result) {
            throw new Exception("请勿重复点击！");
        }

        try {
            // ... 业务逻辑处理
            Thread.sleep(3000);
        } finally {
            lock.releaseLock();
        }
    }


    @RequestMapping("/printDoc")  // 打印机任务
    public void printDoc(@RequestParam String docId, String printerId) throws Exception {
        RedisDistributedLockBuilder.Lock lock = distributedLockBuilder.build("lock_printer_" + printerId, "1",
                true,  // 自旋等待
                60 * 1000, // 60秒超时，即一个打印任务最多占用锁60秒。
                10  // 自旋时，每次休息10毫秒
        );

        // 获取锁，若获取不到锁，则一直等待
        lock.acquireLock();
        try {
            // ... 处理打印任务
            Thread.sleep(3000);
        } finally {
            lock.releaseLock();
        }
    }
}
```

5. Redis常见的应用问题

Redis通常作为数据库的缓存来使用，但如果使用不当，会造成数据库压力过大甚至崩溃。

简单的读写缓存流程如下：

1. 检测读取数据是否在缓存中。若在，则直接读取缓存，返回结果。
2. 若缓存不存在，则读取数据库，然后将数据写入缓存，返回结果。

在这个简单的设计中存在许多漏洞，会造成一致性问题、特殊情况下数据库压力过大等。

5.1 缓存与数据库的一致性

背景：许多业务场景下，缓存的数据会同时存在读写的情况。例如：卖火车票。

风险：当“数据被写入数据库”到“覆盖原有缓存”之间的一小段时间内，请求会获取到旧的数据，导致数据库数据与缓存不一致。

根据“严苛程度”对缓存一致性进行分类，可分为：

• 强一致性：缓存与数据库中的数据一定要一致，不能有半点误差。适合业务有要求，且读多写少的情况。
• 弱一致性：数据写入后，系统不保证是什么时候才可以读到最新的数据。但如果系统可以保证：在某个时间级别后，一定可以读到最新的数据，那么就称为最终一致性（最终一致性是弱一致性的特例）。

如果业务没有强制要求，或者读写都多的场景，建议都使用弱一致性。例如：12306属于读写都多的场景，它就无法保证强一致性，经常显示有票，下单发现售完。

强一致性解决方案：一般不保证强一致性。但对于读特别多，写特别少，可以通过加锁的方式实现。“删除缓存，更新数据库”（该动作全程加锁），在该期间，所有的查询都访问数据库，且不写入缓存。即保证数据更新期间，其他线程不能读写缓存

最终一致性解决方案：

• 写数据后删除缓存，等下次读取的时候再重新插入缓存：适合读多写少的情况
• 仅通过过期时间删除缓存：适合读多写多的场景。例如：12306抢票，都卖完多久了，查询还是显示有票。

不推荐使用的有风险的一致性策略：

• 先更新缓存，再更新数据库：若更新数据库失败，则数据不一致。
• 先删除缓存，再更新数据库：若中间有数据读取，则数据不一致。
• 先更新数据库，再更新缓存：若更新缓存失败，则数据不一致。

5.2 缓存击穿（Cache Breakdown）

背景：业务中存在某一个热点key，读取量特别大。

风险：若该key过期，从过期到下次被写入数据库之间会有一小段时间内Redis中是没有该key的缓存的。由于该Key的读取量特别大，就会造成一瞬间会有大量的请求访问数据库，造成数据库出问题。

击穿：在某一瞬间被突破。缓存击穿：缓存在某一瞬间被突破，全部涌向了数据库。

解决方案：

1. 对热点Key不设置过期时间，只在写的时候更新缓存（对于写，可以加入分布式锁，保证同一时间只有一个线程更新缓存）。
2. 缓存预热：在上线前，先将热点key放入缓存，避免上线时大量请求涌入。

5.3 缓存穿透（Cache Penetration）

背景：业务中存在一个接口，可以根据ID查询数据

风险：若有大量的请求都查询了不存在的ID，由于ID数据不存在，所以无法命中缓存，最后这些请求都去访问了数据库。常见场景：① 恶意攻击，遍历数据。② 网络爬虫，遍历数据。

穿透（渗透）：一些杂质穿透过滤网渗透下去了。缓存穿透：Redis就是请求的过滤网，但有些杂质（无ID的请求）穿透了这张过滤网，到达了数据库。

解决方案：

1. 对空数据也进行缓存。例如：ID=1234无数据，那么就缓存一条ID_1234: None
2. 使用布隆过滤器。其特点为：通过布隆过滤器，可以知道数据可能存在，或数据一定不存在

5.4 缓存雪崩（Cache Avalanche）

背景：项目中有大量数据都用到了缓存，且设置了过期时间。

风险：① 如果大量的数据写入和过期时间都一致，就会导致同一时间大量缓存过期，导致大量请求涌入数据库。② Redis宕机，导致大量请求访问数据库。

雪崩：本来一片宁静，突然山上的雪全部涌下来。缓存雪崩：本来一片宁静，突然大量缓存同时过期，大量请求涌下来，砸到数据库上。

解决方案：

1. 不同的业务尽量根据需求设置不同的过期时间。
2. 对于同一个业务，设置过期时间时，也需要在给定的基础上，增加一个随机时间来分散过期时间
3. 增加熔断和降级机制：对数据库的请求量做限制，避免大量请求涌入数据库。对于被拒绝的请求，采用降级处理，例如：返回错误，使用冷缓存（见双层缓存）
4. 使用双层缓存：准备两套Redis，第一套过期时间短，与数据库保持高度一致（热缓存数据）。第二套过期时间长，与数据库一致性较差（冷缓存数据）。若数据库连接被熔断，则降级处理时，返回冷缓存数据。
5. 数据预热：上线前将数据放入缓存，避免一上线后，大量请求访问数据库。
6. 采用高可用架构：针对Redis宕机问题，可采用集群主从模式来防止Redis宕机。

6. Redis的高级数据结构

Redis除了最基本的5种数据结构，后续版本还支持一些高级的数据结构。

6.1 Bitmap位运算

Bitmaps：存储了一串0/1，可以很方便高效的做位运算。常见应用：布隆过滤器

Bitmaps的基本使用：

```python
# 设置<key>的第<offset>位为0或1
setbit <key> <offset> <0|1>

# 获取<key>的第<offset>位的值，返回0或1。（若key不存在或该位置没被设置为1，则返回0）
getbit <key> <offset>

# 获取<key>中1的数量
bitcount <key>

# 对key1,key2,...做与/或/异或/非操作，将结果赋值给<destkey>
bitop <and|or|xor|not> <destkey> key1 [key2, ...]
```

Bitmaps底层原理：本质存储的还是字符串。只不过把字符串映射成了二进制进行处理。你甚至可以用混用string的命令和bigmap的命令，例如：

```python
> set k1 "ab"
OK

> bitcount k1
6   
```

返回 6 的原因是，ab 的ascii码分别是“97”和“98”，对应到二进制就是“0110 0001”和“0110 0010”，一共6个1

6.2 HyperLogLog海量数据统计

HyperLogLog用于海量数据的基数估计。集合的基数就是该集合中不重复元素的个数。

Redis中，只需要12K内存，就可以估算2^64个不同元素的基数。

HyperLogLog的基本使用：

```python
# 将一个或多个<element>增添到<key>中
pfadd <key> <element> [<element>, ...]

# 估算一个或多个<key>的基数
pfcount <key> [<key>, ...]

# 将一个或多个<sourcekey>合并为一个新的<destkey>
pfmerge <destkey> <sourcekey> [<sourcekey>, ...]
```

应用举例：统计月活数量。假设B站每月要统计月活用户数，那么每个用户请求一次，就pfadd <key-month> <userid>一下，最后月底pfcount <key-month>就行了。因为HyperLogLog底层不会真的去记录每个用户的userid，所以耗内存很小。

6.3 GEO地图信息

GEO 主要用于存储地理位置信息，并对存储的信息进行操作

GEO 的基本使用：

```python
# 增添地理位置信息：经度、维度、位置名称。可以一次增添多个
geoadd <key> <longitude> <latitude> <member> [<longitude> <latitude> <member> ...]

# 获取地理位置坐标。可以一次获取多个地点的
geopos <key> <member> [<member> ...]

# 获取两个位置之间的距离，可以指定单位
geodist <key> <member1> <member2> [m|km|ft|mi]

# 根据经纬度，获取半径以内的地点
georadius <key> <longitude> <latitude> <radius> <m|km|ft|mi>

# 根据地点名称，获取半径以内的地点
georadiusbymember <key> <member> <radius> <m|km|ft|mi>
```

使用样例：

```python
# 增添北京市的部分建筑
> geoadd Beijing 116.397469 39.908821 TianAnMen  # 天安门
(integer) 1

# 增添北大和清华
> geoadd Beijing 116.316833 39.998877 PKU 116.337180 39.971874 THU
(integer) 2

# 获取北大的经纬度
> geopos Beijing PKU
1) 1) "116.31683439016342"
   2) "39.998877029375571"

# 获取北大和清华的距离
> geodist Beijing PKU THU km
"3.4680"

# 根据北大经纬度，获取半径10km之内的地理位置
> georadius Beijing 116.310547 39.992828 10 km
1) "THU"
2) "PKU"

# 根据北大名称，获取半径10km之内的地理位置
> georadiusbymember Beijing PKU 10 km
1) "THU"
2) "PKU"
```

---

参考资料

• Redis实战（Josiah L. Carlson）

• Redis开发与运维（付磊）