Java 并发之分工工具

发表于 2020-07-14 更新于 2025-09-13 分类于 Java ， JavaCore ，并发本文字数： 12k 阅读时长 ≈ 11 分钟

Java 并发之分工工具

对于简单的并行任务，你可以通过“线程池 + Future”的方案来解决；如果任务之间有聚合关系，无论是 AND 聚合还是 OR 聚合，都可以通过 CompletableFuture 来解决；而批量的并行任务，则可以通过 CompletionService 来解决。

FutureTask

FutureTask 有两个构造函数：

1 2	FutureTask(Callable<V> callable); FutureTask(Runnable runnable, V result);

FutureTask 实现了 Runnable 和 Future 接口。由于实现了 Runnable 接口，所以可以将 FutureTask 对象作为任务提交给 ThreadPoolExecutor 去执行，也可以直接被 Thread 执行；又因为实现了 Future 接口，所以也能用来获得任务的执行结果。

下面，通过一组示例来展示 FutureTask 如何分别交给线程池、线程执行。

::: tabs#创建 FutureTask 示例

@tab FutureTask 交给线程池执行

【示例】FutureTask 交给线程池执行

public class FutureTaskDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建 FutureTask
        Task task = new Task();
        FutureTask<String> f1 = new FutureTask<>(task);
        FutureTask<String> f2 = new FutureTask<>(task);

        // 创建线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        executor.submit(f1);
        executor.submit(f2);
        System.out.println(f1.get());
        System.out.println(f2.get());
        executor.shutdown();
    }

    static class Task implements Callable<String> {

        @Override
        public String call() {
            return Thread.currentThread().getName() + " 执行成功！";
        }

    }

}
// 输出
// pool-1-thread-1 执行成功！
// pool-1-thread-2 执行成功！

@tab FutureTask 交给线程执行

【示例】FutureTask 交给线程执行

public class FutureTaskDemo2 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        // 创建 FutureTask
        Task task = new Task();
        FutureTask<String> f1 = new FutureTask<>(task);
        FutureTask<String> f2 = new FutureTask<>(task);

        // 创建线程
        new Thread(f1).start();
        new Thread(f2).start();
        System.out.println(f1.get());
        System.out.println(f2.get());
    }

    static class Task implements Callable<String> {

        @Override
        public String call() {
            return Thread.currentThread().getName() + " 执行成功！";
        }

    }

}
// 输出
// Thread-0 执行成功！
// Thread-1 执行成功！

@tab 用 FutureTask 完成并行计算

【示例】用 FutureTask 完成并行计算

public class FutureTaskDemo3 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        // 创建一个线程池来执行任务
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 创建两个 Callable 对象
        Callable<Integer> t1 = () -> {
            int result = 0;
            for (int i = 1; i <= 100; i++) {
                result += i;
            }
            return result;
        };
        Callable<Integer> t2 = () -> {
            int result = 0;
            for (int i = 101; i <= 200; i++) {
                result += i;
            }
            return result;
        };

        // 创建两个 FutureTask 对象
        FutureTask<Integer> f1 = new FutureTask<>(t1);
        FutureTask<Integer> f2 = new FutureTask<>(t2);

        // 提交任务到线程池执行
        executor.execute(f1);
        executor.execute(f2);

        // 获取任务的结果
        Integer value1 = f1.get();
        Integer value2 = f2.get();
        System.out.println("total = " + value1 + value2);

        // 关闭线程池
        executor.shutdown();
    }

}

:::

CompletableFuture

JDK8 提供了 CompletableFuture 来支持异步编程。

CompletableFuture 提供了四个静态方法来创建一个异步操作。

// 使用默认线程池
public static CompletableFuture<Void> runAsync(Runnable runnable) { // 省略 }
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { // 省略 }

// 使用自定义线程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) { // 省略 }
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) { // 省略 }

上面的 4 个静态方法中，有 2 个 runAsync 方法，2 个 supplyAsync 方法，它们的区别是：

runAsync 方法没有返回值。
supplyAsync 方法有返回值。

默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option: -Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置 ForkJoinPool 线程池的线程数）。如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰。

CompletionStage

CompletionStage 接口可以清晰地描述任务之间的时序关系，如串行关系、并行关系、汇聚关系等。

串行关系

CompletionStage 接口里面描述串行关系，主要是 thenApply、thenAccept、thenRun 和 thenCompose 这四个系列的接口。

thenApply 系列函数里参数 fn 的类型是接口 Function<T, R>，这个接口里与 CompletionStage 相关的方法是 R apply(T t)，这个方法既能接收参数也支持返回值，所以 thenApply 系列方法返回的是CompletionStage<R>。

thenAccept 系列方法里参数 consumer 的类型是接口 Consumer<T>，这个接口里与 CompletionStage 相关的方法是 void accept(T t)，这个方法虽然支持参数，但却不支持回值，所以 thenAccept 系列方法返回的是CompletionStage<Void>。

thenRun 系列方法里 action 的参数是 Runnable，所以 action 既不能接收参数也不支持返回值，所以 thenRun 系列方法返回的也是 CompletionStage<Void>。

这些方法里面 Async 代表的是异步执行 fn、consumer 或者 action。其中，需要你注意的是 thenCompose 系列方法，这个系列的方法会新创建出一个子流程，最终结果和 thenApply 系列是相同的。

CompletionStage<R> thenApply(fn);
CompletionStage<R> thenApplyAsync(fn);
CompletionStage<Void> thenAccept(consumer);
CompletionStage<Void> thenAcceptAsync(consumer);
CompletionStage<Void> thenRun(action);
CompletionStage<Void> thenRunAsync(action);
CompletionStage<R> thenCompose(fn);
CompletionStage<R> thenComposeAsync(fn);

描述 AND 汇聚关系

CompletionStage 接口里面描述 AND 汇聚关系，主要是 thenCombine、thenAcceptBoth 和 runAfterBoth 系列的接口，这些接口的区别也是源自 fn、consumer、action 这三个核心参数不同。

CompletionStage<R> thenCombine(other, fn);
CompletionStage<R> thenCombineAsync(other, fn);
CompletionStage<Void> thenAcceptBoth(other, consumer);
CompletionStage<Void> thenAcceptBothAsync(other, consumer);
CompletionStage<Void> runAfterBoth(other, action);
CompletionStage<Void> runAfterBothAsync(other, action);

描述 OR 汇聚关系

CompletionStage 接口里面描述 OR 汇聚关系，主要是 applyToEither、acceptEither 和 runAfterEither 系列的接口，这些接口的区别也是源自 fn、consumer、action 这三个核心参数不同。

CompletionStage applyToEither(other, fn);
CompletionStage applyToEitherAsync(other, fn);
CompletionStage acceptEither(other, consumer);
CompletionStage acceptEitherAsync(other, consumer);
CompletionStage runAfterEither(other, action);
CompletionStage runAfterEitherAsync(other, action);

下面的示例代码展示了如何使用 applyToEither() 方法来描述一个 OR 汇聚关系。

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});

CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});

CompletableFuture<String> f3 = f1.applyToEither(f2, s -> s);
System.out.println(f3.join());

异常处理

虽然上面我们提到的 fn、consumer、action 它们的核心方法都不允许抛出可检查异常，但是却无法限制它们抛出运行时异常，例如下面的代码，执行 7/0 就会出现除零错误这个运行时异常。非异步编程里面，我们可以使用 try {} catch {} 来捕获并处理异常，那在异步编程里面，异常该如何处理呢？

1
2
3

CompletableFuture<Integer> f = CompletableFuture.supplyAsync(() -> (7 / 0))
                                                 .thenApply(r -> r * 10);
System.out.println(f.join());

CompletionStage 接口给我们提供的方案非常简单，比 try {} catch {} 还要简单，下面是相关的方法，使用这些方法进行异常处理和串行操作是一样的，都支持链式编程方式。

CompletionStage exceptionally(fn);
CompletionStage<R> whenComplete(consumer);
CompletionStage<R> whenCompleteAsync(consumer);
CompletionStage<R> handle(fn);
CompletionStage<R> handleAsync(fn);

下面的示例代码展示了如何使用 exceptionally() 方法来处理异常，exceptionally() 的使用非常类似于 try {} catch {} 中的 catch {}，但是由于支持链式编程方式，所以相对更简单。既然有 try {} catch {}，那就一定还有 try {} catch {}，whenComplete() 和 handle() 系列方法就类似于 try {} catch {} 中的 finally {}，无论是否发生异常都会执行 whenComplete() 中的回调函数 consumer 和 handle() 中的回调函数 fn。whenComplete() 和 handle() 的区别在于 whenComplete() 不支持返回结果，而 handle() 是支持返回结果的。

CompletableFuture<Integer> f = CompletableFuture.supplyAsync(() -> 7 / 0)
                                                .thenApply(r -> r * 10)
                                                .exceptionally(e -> 0);
System.out.println(f.join());

CompletionService

CompletionService 接口的实现类是 ExecutorCompletionService，这个实现类的构造方法有两个，分别是：

ExecutorCompletionService(Executor executor)；
ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)。

这两个构造方法都需要传入一个线程池，如果不指定 completionQueue，那么默认会使用无界的 LinkedBlockingQueue。任务执行结果的 Future 对象就是加入到 completionQueue 中。

下面的示例代码完整地展示了如何利用 CompletionService 来实现高性能的询价系统。其中，我们没有指定 completionQueue，因此默认使用无界的 LinkedBlockingQueue。之后通过 CompletionService 接口提供的 submit() 方法提交了三个询价操作，这三个询价操作将会被 CompletionService 异步执行。最后，我们通过 CompletionService 接口提供的 take() 方法获取一个 Future 对象（前面我们提到过，加入到阻塞队列中的是任务执行结果的 Future 对象），调用 Future 对象的 get() 方法就能返回询价操作的执行结果了。

// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 创建 CompletionService
CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
// 异步向电商 S1 询价
cs.submit(()->getPriceByS1());
// 异步向电商 S2 询价
cs.submit(()->getPriceByS2());
// 异步向电商 S3 询价
cs.submit(()->getPriceByS3());
// 将询价结果异步保存到数据库
for (int i=0; i<3; i++) {
  Integer r = cs.take().get();
  executor.execute(()->save(r));
}

CompletionService 接口提供的方法有 5 个，这 5 个方法的方法签名如下所示。

其中，submit() 相关的方法有两个。一个方法参数是Callable<V> task，前面利用 CompletionService 实现询价系统的示例代码中，我们提交任务就是用的它。另外一个方法有两个参数，分别是Runnable task和V result，这个方法类似于 ThreadPoolExecutor 的 <T> Future<T> submit(Runnable task, T result) ，这个方法在《23 | Future：如何用多线程实现最优的“烧水泡茶”程序？》中我们已详细介绍过，这里不再赘述。

CompletionService 接口其余的 3 个方法，都是和阻塞队列相关的，take()、poll() 都是从阻塞队列中获取并移除一个元素；它们的区别在于如果阻塞队列是空的，那么调用 take() 方法的线程会被阻塞，而 poll() 方法会返回 null 值。 poll(long timeout, TimeUnit unit) 方法支持以超时的方式获取并移除阻塞队列头部的一个元素，如果等待了 timeout unit 时间，阻塞队列还是空的，那么该方法会返回 null 值。

Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
Future<V> take() throws InterruptedException;
Future<V> poll();
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;

当需要批量提交异步任务的时候建议你使用 CompletionService。CompletionService 将线程池 Executor 和阻塞队列 BlockingQueue 的功能融合在了一起，能够让批量异步任务的管理更简单。除此之外，CompletionService 能够让异步任务的执行结果有序化，先执行完的先进入阻塞队列，利用这个特性，你可以轻松实现后续处理的有序性，避免无谓的等待，同时还可以快速实现诸如 Forking Cluster 这样的需求。

CompletionService 的实现类 ExecutorCompletionService，需要你自己创建线程池，虽看上去有些啰嗦，但好处是你可以让多个 ExecutorCompletionService 的线程池隔离，这种隔离性能避免几个特别耗时的任务拖垮整个应用的风险。

ForkJoinPool

Fork/Join 是一个并行计算的框架，主要就是用来支持分治任务模型的，这个计算框架里的** Fork 对应的是分治任务模型里的任务分解，Join 对应的是结果合并。Fork/Join 计算框架主要包含两部分，一部分是分治任务的线程池 ForkJoinPool，另一部分是分治任务 ForkJoinTask**。这两部分的关系类似于 ThreadPoolExecutor 和 Runnable 的关系，都可以理解为提交任务到线程池，只不过分治任务有自己独特类型 ForkJoinTask。

ForkJoinTask 是一个抽象类，它的方法有很多，最核心的是 fork() 方法和 join() 方法，其中 fork() 方法会异步地执行一个子任务，而 join() 方法则会阻塞当前线程来等待子任务的执行结果。ForkJoinTask 有两个子类——RecursiveAction 和 RecursiveTask，通过名字你就应该能知道，它们都是用递归的方式来处理分治任务的。这两个子类都定义了抽象方法 compute()，不过区别是 RecursiveAction 定义的 compute() 没有返回值，而 RecursiveTask 定义的 compute() 方法是有返回值的。这两个子类也是抽象类，在使用的时候，需要你定义子类去扩展。

Fork/Join 并行计算的核心组件是 ForkJoinPool，所以下面我们就来简单介绍一下 ForkJoinPool 的工作原理。

ForkJoinPool 本质上也是一个生产者 - 消费者的实现，但是更加智能，你可以参考下面的 ForkJoinPool 工作原理图来理解其原理。ThreadPoolExecutor 内部只有一个任务队列，而 ForkJoinPool 内部有多个任务队列，当我们通过 ForkJoinPool 的 invoke() 或者 submit() 方法提交任务时，ForkJoinPool 根据一定的路由规则把任务提交到一个任务队列中，如果任务在执行过程中会创建出子任务，那么子任务会提交到工作线程对应的任务队列中。

如果工作线程对应的任务队列空了，是不是就没活儿干了呢？不是的，ForkJoinPool 支持一种叫做“任务窃取”的机制，如果工作线程空闲了，那它可以“窃取”其他工作任务队列里的任务，例如下图中，线程 T2 对应的任务队列已经空了，它可以“窃取”线程 T1 对应的任务队列的任务。如此一来，所有的工作线程都不会闲下来了。

ForkJoinPool 中的任务队列采用的是双端队列，工作线程正常获取任务和“窃取任务”分别是从任务队列不同的端消费，这样能避免很多不必要的数据竞争。我们这里介绍的仅仅是简化后的原理，ForkJoinPool 的实现远比我们这里介绍的复杂，如果你感兴趣，建议去看它的源码。

【示例】模拟 MapReduce 统计单词数量


static void main(String[] args) {
    String[] fc = { "hello world",
        "hello me",
        "hello fork",
        "hello join",
        "fork join in world" };
    //创建 ForkJoin 线程池
    ForkJoinPool fjp = new ForkJoinPool(3);
    //创建任务
    MR mr = new MR(fc, 0, fc.length);
    //启动任务
    Map<String, Long> result = fjp.invoke(mr);
    //输出结果
    result.forEach((k, v) -> System.out.println(k + ":" + v));
}

//MR 模拟类
static class MR extends RecursiveTask<Map<String, Long>> {

    private String[] fc;
    private int start, end;

    //构造函数
    MR(String[] fc, int fr, int to) {
        this.fc = fc;
        this.start = fr;
        this.end = to;
    }

    @Override
    protected Map<String, Long> compute() {
        if (end - start == 1) {
            return calc(fc[start]);
        } else {
            int mid = (start + end) / 2;
            MR mr1 = new MR(fc, start, mid);
            mr1.fork();
            MR mr2 = new MR(fc, mid, end);
            //计算子任务，并返回合并的结果
            return merge(mr2.compute(),
                mr1.join());
        }
    }

    //合并结果
    private Map<String, Long> merge(Map<String, Long> r1, Map<String, Long> r2) {
        Map<String, Long> result = new HashMap<>();
        result.putAll(r1);
        //合并结果
        r2.forEach((k, v) -> {
            Long c = result.get(k);
            if (c != null) { result.put(k, c + v); } else { result.put(k, v); }
        });
        return result;
    }

    //统计单词数量
    private Map<String, Long> calc(String line) {
        Map<String, Long> result = new HashMap<>();
        //分割单词
        String[] words = line.split("\\s+");
        //统计单词数量
        for (String w : words) {
            Long v = result.get(w);
            if (v != null) { result.put(w, v + 1); } else { result.put(w, 1L); }
        }
        return result;
    }
}

参考资料

Redis 面试之应用篇

发表于 2020-07-13 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 5.8k 阅读时长 ≈ 5 分钟

Redis 面试之应用篇

缓存

【中等】如何避免缓存雪崩、缓存击穿、缓存穿透？

缓存击穿：指某个热点数据在缓存中失效，导致大量请求直接访问数据库。此时，由于瞬间的高并发，可能导致数据库崩溃。
- 使用互斥锁，确保同一时间只有一个请求可以去数据库查询并更新缓存。
- 热点数据永不过期。
缓存穿透：指查询一个不存在的数据，缓存中没有相应的记录，每次请求都会去数据库查询，造成数据库负担加重。
- 使用布隆过滤器，过滤掉不存在的请求，避免直接访问数据库。
- 对查询结果进行缓存，即使是不存在的数据，也可以缓存一个标识，以减少对数据库的请求。
缓存雪崩：指多个缓存数据在同一时间过期，导致大量请求同时访问数据库，从而造成数据库瞬间负载激增。
- 采用随机过期时间策略，避免多个数据同时过期。
- 使用双缓存策略，将数据同时存储在两层缓存中，减少数据库直接请求。

【中等】如何保证缓存与数据库的数据一致性？

一般来说，系统如果不是严格要求缓存和数据库保持一致性的话，尽量不要将读请求和写请求串行化。串行化可以保证一定不会出现数据不一致的情况，但是它会导致系统的吞吐量大幅度下降。

一般来说缓存的更新有两种情况：

先删除缓存，再更新数据库。
先更新数据库，再删除缓存。

为什么是删除缓存，而不是更新缓存呢？

你可以想想当有多个并发的请求更新数据，你并不能保证更新数据库的顺序和更新缓存的顺序一致，那就会出现数据库中和缓存中数据不一致的情况。所以一般来说考虑删除缓存。

【中等】有哪些常见的内存淘汰算法？

缓存淘汰的类型：

基于空间 - 设置缓存空间大小。
基于容量 - 设置缓存存储记录数。
基于时间
- TTL（Time To Live，即存活期）缓存数据从创建到过期的时间。
- TTI（Time To Idle，即空闲期）缓存数据多久没被访问的时间。

缓存淘汰算法：

FIFO - 先进先出。在这种淘汰算法中，先进入缓存的会先被淘汰。这种可谓是最简单的了，但是会导致我们命中率很低。试想一下我们如果有个访问频率很高的数据是所有数据第一个访问的，而那些不是很高的是后面再访问的，那这样就会把我们的首个数据但是他的访问频率很高给挤出。
LRU - 最近最少使用算法。在这种算法中避免了上面的问题，每次访问数据都会将其放在我们的队尾，如果需要淘汰数据，就只需要淘汰队首即可。但是这个依然有个问题，如果有个数据在 1 个小时的前 59 分钟访问了 1 万次（可见这是个热点数据）, 再后一分钟没有访问这个数据，但是有其他的数据访问，就导致了我们这个热点数据被淘汰。
LFU - 最近最少频率使用。在这种算法中又对上面进行了优化，利用额外的空间记录每个数据的使用频率，然后选出频率最低进行淘汰。这样就避免了 LRU 不能处理时间段的问题。

这三种缓存淘汰算法，实现复杂度一个比一个高，同样的命中率也是一个比一个好。而我们一般来说选择的方案居中即可，即实现成本不是太高，而命中率也还行的 LRU。

分布式锁

【中等】实现分布式锁需要解决哪些问题？

分布式锁的解决方案大致有以下几种：

基于数据库实现
基于缓存（Redis，Memcached 等）实现
基于 Zookeeper 实现

分布式锁的实现要点大同小异，仅在实现细节上有所不同。

实现分布式锁需要解决以下目标：

互斥 - 分布式锁必须是独一无二的，表现形式为：向数据存储插入一个唯一的 key，一旦有一个线程插入这个 key，其他线程就不能再插入了。
避免死锁 - 在分布式锁的场景中，部分失败和异步网络这两个问题是同时存在的。如果一个进程获得了锁，但是这个进程与锁服务之间的网络出现了问题，导致无法通信，那么这个情况下，如果锁服务让它一直持有锁，就会导致死锁的发生。
- 常见的解决思路是引入超时机制，即成功申请锁后，超过一定时间，锁失效（删除 key）。
- 超时机制解锁了死锁问题，但又引入了一个新问题：如果应用加锁时，对于操作共享资源的时长估计不足，可能会出现：操作尚未执行完，但是锁没了的尴尬情况。为了解决这个问题，需要引入锁续期机制：当持有锁的线程尚未执行完操作前，不断周期性检测锁的超时时间，一旦发现快要过期，就自动为锁续期。
- ZooKeeper 分布式锁避免死锁采用了另外一种思路—— Watch 机制。
可重入 - 可重入指的是：同一个线程在没有释放锁之前，能否再次获得该锁。其实现方案是：只需在加锁的时候，记录好当前获取锁的节点 + 线程组合的唯一标识，然后在后续的加锁请求时，如果当前请求的节点 + 线程的唯一标识和当前持有锁的相同，那么就直接返回加锁成功；如果不相同，则按正常加锁流程处理。
公平性 - 当多个线程请求同一锁时，它们必须按照请求的顺序来获取锁，即先来先得的原则。锁的公平性的实现也非常简单，对于被阻塞的加锁请求，我们只要先记录好它们的顺序，在锁被释放后，按顺序颁发就可以了。
重试 - 有时候，加锁失败可能只是由于网络波动、请求超时等原因，稍候就可以成功获取锁。为了应对这种情况，加锁操作需要支持重试机制。常见的做法是，设置一个加锁超时时间，在该时间范围内，不断自旋重试加锁操作，超时后再判定加锁失败。
容错 - 分布式锁若存储在单一节点，一旦该节点宕机或失联，就会导致锁失效。将分布式锁存储在多数据库实例中，加锁时并发写入 N 个节点，只要 N / 2 + 1 个节点写入成功即视为加锁成功。代表有：Redis 官方提供的 RedLock。

【中等】Redis 中如何实现分布式锁？

基础实现

加锁：SET key val NX PX 30000（NX：仅当 key 不存在时写入，PX：超时时间，毫秒）。
解锁：DEL key（直接删除 key 释放锁）。

避免死锁

问题：节点宕机或异常导致锁无法释放。
方案：加锁时设置超时时间（PX/EX），到期自动删除。
原子性：使用 SET key val NX PX 替代 setnx + expire，避免组合命令非原子性问题。

超时续期（WatchDog）

问题：业务未执行完，锁已过期。
方案：启动定时任务检测锁剩余时间，自动续期（如 Redisson 的 WatchDog 机制）。

安全解锁

问题：其他节点误删锁。
方案：
- 加锁时写入唯一标识（如 UUID）。
- 解锁时校验标识，匹配才删除（使用 Lua 脚本保证原子性）：
  1
  2
  3
  4
  5
  if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])
  else
  return 0
  end

自旋重试

问题：网络波动导致加锁失败。
方案：在超时时间内循环重试加锁（伪代码示例见原内容）。

其他问题

不可重入：同一线程无法重复获取同一把锁（可通过 Redisson 等客户端支持）。
单点问题：主从切换可能导致锁失效（可用 RedLock 算法，但存在争议）。

【中等】Red Lock 分布式锁的原理是什么？

核心机制：

多节点写入：同时向多个 Redis 主节点（≥5 个）加锁
多数派原则：半数以上节点加锁成功才算成功
耗时控制：总加锁时间必须小于锁的过期时间
强制清理：解锁时向所有节点发起请求

注意事项

网络延迟：多节点操作会增加耗时风险
性能代价：相比单节点锁性能更低
实现复杂度：需要精确控制时间和节点状态

【中等】Redisson 分布式锁的原理是什么？

Redisson 是一个流行的 Redis Java 客户端，它基于 Netty 开发，并提供了丰富的扩展功能，如：分布式计数器、分布式集合、分布式锁等。

Redisson 支持的分布式锁有多种：Lock, FairLock, MultiLock, RedLock, ReadWriteLock, Semaphore, PermitExpirableSemaphore, CountDownLatch，可以根据场景需要去选择，非常方便。一般而言，使用 Redis 分布式锁，推荐直接使用 Redisson 提供的 API，功能全面且较为可靠。

Redisson 分布式锁的实现要点：

锁的获取：Redisson 使用 Lua 脚本，利用 exists + hexists + hincrby 命令来保证只有一个线程能成功设置键（表示获得锁）。同时，Redisson 会通过 pexpire 命令为锁设置过期时间，防止因宕机等原因导致锁无法释放（即死锁问题）。
锁的续期：为了防止锁在持有过程中过期导致其他线程抢占锁，Redisson 实现了一种叫做 Watch Dog（看门狗） 的锁自动续期的功能。持有锁的线程会定期续期，即更新锁的过期时间，确保任务没有完成时锁不会失效。
锁的释放：锁释放时，Redisson 也是通过 Lua 脚本保证释放操作的原子性。利用 hexists + del 确保只有持有锁的线程才能释放锁，防止误释放锁的情况。Lua 脚本同时利用 publish 命令，广播唤醒其它等待的线程。
可重入锁：Redisson 支持可重入锁，持有锁的线程可以多次获取同一把锁而不会被阻塞。具体是利用 Redis 中的哈希结构，哈希中的 key 为线程 ID，如果重入则 value +1，如果释放则 value -1，减到 0 说明锁被释放了，则 del 锁。

消息队列

【中等】Redis 如何实现消息队列？

Redis 可以做消息队列吗？

Redis 有哪些实现消息队列的方式？

先说结论：可以是可以，但不建议使用 Redis 来做消息队列。和专业的消息队列相比，还是有很多欠缺的地方。

基于 List 实现消息队列

Redis 2.0 之前，如果想要使用 Redis 来做消息队列的话，只能通过 List 来实现。

通过 RPUSH/LPOP 或者 LPUSH/RPOP即可实现简易版消息队列：

# 生产者生产消息
> RPUSH myList msg1 msg2
(integer) 2
> RPUSH myList msg3
(integer) 3
# 消费者消费消息
> LPOP myList
"msg1"

不过，通过 RPUSH/LPOP 或者 LPUSH/RPOP这样的方式存在性能问题，我们需要不断轮询去调用 RPOP 或 LPOP 来消费消息。当 List 为空时，大部分的轮询的请求都是无效请求，这种方式大量浪费了系统资源。

因此，Redis 还提供了 BLPOP、BRPOP 这种阻塞式读取的命令（带 B-Blocking 的都是阻塞式），并且还支持一个超时参数。如果 List 为空，Redis 服务端不会立刻返回结果，它会等待 List 中有新数据后再返回或者是等待最多一个超时时间后返回空。如果将超时时间设置为 0 时，即可无限等待，直到弹出消息

# 超时时间为 10s
# 如果有数据立刻返回，否则最多等待 10 秒
> BRPOP myList 10
null

List 实现消息队列功能太简单，像消息确认机制等功能还需要我们自己实现，最要命的是没有广播机制，消息也只能被消费一次。

基于发布订阅功能实现消息队列

Redis 2.0 引入了发布订阅 (pub/sub) 功能，解决了 List 实现消息队列没有广播机制的问题。

Redis 发布订阅 (pub/sub) 功能

pub/sub 中引入了一个概念叫 channel（频道），发布订阅机制的实现就是基于这个 channel 来做的。

pub/sub 涉及发布者（Publisher）和订阅者（Subscriber，也叫消费者）两个角色：

发布者通过 PUBLISH 投递消息给指定 channel。
订阅者通过SUBSCRIBE订阅它关心的 channel。并且，订阅者可以订阅一个或者多个 channel。

pub/sub 既能单播又能广播，还支持 channel 的简单正则匹配。不过，消息丢失（客户端断开连接或者 Redis 宕机都会导致消息丢失）、消息堆积（发布者发布消息的时候不会管消费者的具体消费能力如何）等问题依然没有一个比较好的解决办法。

基于 Stream 实现消息队列

为此，Redis 5.0 新增加的一个数据结构 Stream 来做消息队列。Stream 支持：

发布 / 订阅模式
按照消费者组进行消费（借鉴了 Kafka 消费者组的概念）
消息持久化（ RDB 和 AOF）
ACK 机制（通过确认机制来告知已经成功处理了消息）
阻塞式获取消息

Stream 的结构如下：

这是一个有序的消息链表，每个消息都有一个唯一的 ID 和对应的内容。ID 是一个时间戳和序列号的组合，用来保证消息的唯一性和递增性。内容是一个或多个键值对（类似 Hash 基本数据类型），用来存储消息的数据。

这里再对图中涉及到的一些概念，进行简单解释：

Consumer Group：消费者组用于组织和管理多个消费者。消费者组本身不处理消息，而是再将消息分发给消费者，由消费者进行真正的消费
last_delivered_id：标识消费者组当前消费位置的游标，消费者组中任意一个消费者读取了消息都会使 last_delivered_id 往前移动。
pending_ids：记录已经被客户端消费但没有 ack 的消息的 ID。

Stream 使用起来相对要麻烦一些，这里就不演示了。

总的来说，Stream 已经可以满足一个消息队列的基本要求了。不过，Stream 在实际使用中依然会有一些小问题不太好解决比如在 Redis 发生故障恢复后不能保证消息至少被消费一次。

综上，和专业的消息队列相比，使用 Redis 来实现消息队列还是有很多欠缺的地方比如消息丢失和堆积问题不好解决。因此，我们通常建议不要使用 Redis 来做消息队列，你完全可以选择市面上比较成熟的一些消息队列比如 RocketMQ、Kafka。不过，如果你就是想要用 Redis 来做消息队列的话，那我建议你优先考虑 Stream，这是目前相对最优的 Redis 消息队列实现。

相关阅读：Redis 消息队列发展历程 - 阿里开发者 - 2022

延时任务

【中等】如何基于 Redis 实现延时任务？

基于 Redis 实现延时任务的功能无非就下面两种方案：

Redis 过期事件监听
Redisson 内置的延时队列

Redis 过期事件监听的存在时效性较差、丢消息、多服务实例下消息重复消费等问题，不被推荐使用。

Redisson 内置的延时队列具备下面这些优势：

减少了丢消息的可能：DelayedQueue 中的消息会被持久化，即使 Redis 宕机了，根据持久化机制，也只可能丢失一点消息，影响不大。当然了，你也可以使用扫描数据库的方法作为补偿机制。
消息不存在重复消费问题：每个客户端都是从同一个目标队列中获取任务的，不存在重复消费的问题。

参考资料

Redis 面试

发表于 2020-07-13 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 20k 阅读时长 ≈ 19 分钟

Redis 面试

Redis 简介

【简单】什么是 Redis？

什么是 Redis？

Redis 是一个开源的、数据存于内存中的 K-V 数据库。由于，Redis 的读写操作都是在内存中完成，因此其读写速度非常快。

高性能 - 由于，Redis 的读写操作都是在内存中完成，因此性能极高。
高并发 - Redis 单机 QPS 能达到 10w+，将近是 Mysql 的 10 倍。

Redis 常被用于缓存，消息队列、分布式锁等场景。

Redis 有什么功能和特性？

Redis 的功能和特性：

Redis 支持多种数据类型。如：String（字符串）、Hash（哈希）、 List （列表）、Set（集合）、Zset（有序集合）、Bitmaps（位图）、HyperLogLog（基数统计）、GEO（地理空间）、Stream（流）。
Redis 的读写采用“单线程”模型，因此，其操作天然就具有原子性。需要注意的是，Redis 6.0 后在其网络模块中引入了多线程 I/O 机制。
Redis 支持两种持久化策略：RDB 和 AOF。
Redis 有多种高可用方案：主从复制模式、哨兵模式、集群模式。
Redis 支持很多丰富的特性，如：事务、Lua 脚本、发布订阅、过期删除、内存淘汰等等。

图来自 Redis Explained

【简单】Redis 有哪些应用场景？

Redis 常见应用场景如下：

缓存 - 将热点数据放到内存中，设置内存的最大使用量以及过期淘汰策略来保证缓存的命中率。
计数器 - Redis 这种内存数据库能支持计数器频繁的读写操作。
应用限流 - 限制一个网站访问流量。
消息队列 - 使用 List 数据类型，它是双向链表。
查找表 - 使用 HASH 数据类型。
聚合运算 - 使用 SET 类型，例如求两个用户的共同好友。
排行榜 - 使用 ZSET 数据类型。
分布式 Session - 多个应用服务器的 Session 都存储到 Redis 中来保证 Session 的一致性。
分布式锁 - 除了可以使用 SETNX 实现分布式锁之外，还可以使用官方提供的 RedLock 分布式锁实现。

【简单】Redis 有哪些里程碑版本？

Redis 里程碑版本如下：

Redis 1.0（2010 年） - Redis 1.0 发布，采用单机架构，一般作为业务应用的缓存。但是 Redis 的数据是存在内存中的，重启 Redis 时，数据会全部丢失，流量直接打到数据库。
Redis 2.8（2013 年）
- 持久化 - Redis 引入了 RDB 内存快照来持久化数据。它还支持 AOF（仅追加文件），其中每个写入命令都写入 AOF 文件。
- 复制 - 添加了复制功能以提高可用性。主实例处理实时读写请求，而副本同步主实例的数据。
- 哨兵 - 引入了 Sentinel 来实时监控 Redis 实例。Sentinel 是一个旨在帮助管理 Redis 实例的系统。它执行以下四个任务：监控、通知、自动故障转移和共享配置。
Redis 3.0（2015 年） - 官方提供了 redis-cluster。redis-cluster 是一种分布式数据库解决方案，通过分片管理数据。数据被分成 16384 个槽，每个节点负责槽的一部分。
Redis 5.0（2017 年） - 新增 Stream 数据类型。
Redis 6.0（2020 年） - 在网络模块中引入了多线程 I/O。Redis 模型分为网络模块和主处理模块。特别注意：Redis 不再完全是单线程架构。

【简单】对比一下 Redis 和 Memcached？

Redis 和 Memcached 有什么相同点？

Redis 和 Memcached 有什么差异？

分布式缓存技术选型，选 Redis 还是 Memcached，为什么？

Redis 与 Memcached 的共性：

都是内存数据库，因此性能都很高。
都有过期策略。

因为以上两点，所以常被作为缓存使用。

Redis 与 Memcached 的差异：

核心差异对比：

	Memcached	Redis
数据类型	只支持 String 类型	支持多种数据类型：String、Hash、List、Set、ZSet 等
持久化	不支持持久化，一旦重启或宕机就会丢失数据	支持两种持久化策略：RDB 和 AOF
分布式	本身不支持分布式，只能通过在客户端使用像一致性哈希这样的分布式算法来实现分布式存储，这种方式在存储和查询时都需要先在客户端计算一次数据所在的节点	支持分布式
线程模型	采用多线程+IO 多路复用。在 100k 以上的数据中，Memcached 性能要高于 Redis	读写采用单线程+IO 多路复用。因此存储小数据时比 Memcached 性能更高
其他功能	不支持	支持发布订阅模型、Lua 脚本、事务等功能

通过以上分析，可以看出，Redis 在很多方面都占有优势。因此，绝大多数情况下，优先选择 Redis 作为分布式缓存。

扩展：《脚踏两只船的困惑 - Memcached 与 Redis》

【简单】Redis 有哪些 Java 客户端？各有什么优劣？

Redis 的主流 Java 客户端有三种，对比如下：

客户端	线程安全	自动重连	编程模型	适用场景
Jedis	❌	❌	同步	简单应用、快速开发
Lettuce	✅	✅	同步/异步/响应式	高并发、Spring Boot 项目
Redisson	✅	✅	同步/异步	分布式系统、高级功能需求

推荐选择：

基础需求 → Jedis（简单直接）。
高并发/Spring 项目 → Lettuce（默认选择）。
分布式锁/队列等 → Redisson（功能强大）。

Jedis

✅ 优点

简单易用：API 直观，适合快速上手。
广泛使用：社区支持丰富，文档齐全。
性能良好：常规操作高效。
功能全面：支持字符串、哈希、列表等基础数据结构。

❌ 缺点

非线程安全：需为每个线程创建独立实例。
无自动重连：网络异常需手动处理。
同步阻塞：高并发时可能成为性能瓶颈。

Lettuce

✅ 优点

线程安全：多线程共享同一连接。
高性能：基于 Netty 实现，支持高并发。
自动重连：网络中断后自动恢复。
多编程模型：支持同步、异步、响应式（如 Reactive API）。

❌ 缺点

API 较复杂：学习成本高于 Jedis。
资源消耗：异步模式可能占用更多内存/CPU。

Redisson

✅ 优点

分布式支持：内置分布式锁、队列、缓存等高级功能。
线程安全：天然适配多线程场景。
集群友好：完善支持 Redis 集群模式。
稳定性高：企业级应用验证。

❌ 缺点

学习曲线陡峭：需掌握分布式概念。
依赖兼容性：可能与其他库冲突需调优。

Redis 内存管理

【中等】Redis 支持哪些过期删除策略？

Redis 支持哪些过期删除策略？

常见的过期策略有哪些，Redis 的选择考量是什么？

Redis 采用的过期策略是：定期删除+惰性删除。

定时删除 - 在设置 key 的过期时间的同时，创建一个定时器，让定时器在 key 的过期时间来临时，立即执行 key 的删除操作。
- 优点 - 保证过期 key 被尽可能快的删除，释放内存。
- 缺点 - 如果过期 key 较多，可能会占用相当一部分的 CPU，从而影响服务器的吞吐量和响应时延。
惰性删除 - 放任 key 过期不管，但是每次访问 key 时，都检查 key 是否过期，如果过期的话，就删除该 key ；如果没有过期，就返回该 key。
- 优点 - 占用 CPU 最少。程序只会在读写键时，对当前键进行过期检查，因此不会有额外的 CPU 开销。
- 缺点 - 过期的 key 可能因为没有被访问，而一直无法释放，造成内存的浪费，有内存泄漏的风险。
定期删除 - 每隔一段时间，程序就对数据库进行一次检查，删除里面的过期 key。至于要删除多少过期 key ，以及要检查多少个数据库，则由算法决定。定期删除是前两种策略的一种折中方案。定期删除策略的难点是删除操作执行的时长和频率。
- 执行太频或执行时间过长，就会出现和定时删除相同的问题；
- 执行太少或执行时间过短，就会出现和惰性删除相同的问题；

【中等】Redis 有哪些内存淘汰策略？

Redis 内存不足时，怎么办？

Redis 有哪些内存淘汰策略？

如何选择内存淘汰策略？

Redis 内存淘汰要点

失效时间 - 作为一种定期清理无效数据的重要机制，在 Redis 提供的诸多命令中，EXPIRE、EXPIREAT、PEXPIRE、PEXPIREAT 以及 SETEX 和 PSETEX 均可以用来设置一条键值对的失效时间。而一条键值对一旦被关联了失效时间就会在到期后自动删除（或者说变得无法访问更为准确）。
最大缓存 - Redis 允许通过 maxmemory 参数来设置内存最大值。当内存达设定的阀值，就会触发内存淘汰。
内存淘汰 - 内存淘汰是为了更好的利用内存——清理部分缓存，以此换取内存的利用率，即尽量保证 Redis 缓存中存储的是热点数据。

Redis 内存淘汰策略

不淘汰
- noeviction - 当内存使用达到阈值的时候，所有引起申请内存的命令会报错。这是 Redis 默认的策略。
在过期键中进行淘汰
- volatile-random - 在设置了过期时间的键空间中，随机移除某个 key。
- volatile-ttl - 在设置了过期时间的键空间中，具有更早过期时间的 key 优先移除。
- volatile-lru - 在设置了过期时间的键空间中，优先移除最近未使用的 key。
- volatile-lfu （Redis 4.0 新增）- 淘汰所有设置了过期时间的键值中，最少使用的键值。
在所有键中进行淘汰
- allkeys-random - 在主键空间中，随机移除某个 key。
- allkeys-lru - 在主键空间中，优先移除最近未使用的 key。
- allkeys-lfu (Redis 4.0 新增） - 淘汰整个键值中最少使用的键值。

如何选择内存淘汰策略

如果数据呈现正态分布，也就是一部分数据访问频率高，一部分数据访问频率低，则使用 allkeys-lru 或 allkeys-lfu。
如果数据呈现平均分布，也就是所有的数据访问频率都相同，则使用 allkeys-random。
若 Redis 既用于缓存，也用于持久化存储时，适用 volatile-lru 、volatile-lfu、volatile-random。但是，这种情况下，也可以部署两个 Redis 集群来达到同样目的。
为 key 设置过期时间实际上会消耗更多的内存。因此，如果条件允许，建议使用 allkeys-lru 或 allkeys-lfu，从而更高效的使用内存。

【中等】Redis 持久化时，对过期键会如何处理？

RDB 持久化

RDB 文件生成阶段 - 从内存状态持久化成 RDB（文件）的时候，会对 key 进行过期检查，过期的键“不会”被保存到新的 RDB 文件中，因此 Redis 中的过期键不会对生成新 RDB 文件产生任何影响。
RDB 加载阶段 - RDB 加载阶段时，要看服务器是主服务器还是从服务器，分别对应以下两种情况：
- 如果 Redis 是“主服务器”运行模式的话，在载入 RDB 文件时，程序会对文件中保存的键进行检查，过期键“不会”被载入到数据库中。所以过期键不会对载入 RDB 文件的主服务器造成影响；
- 如果 Redis 是“从服务器”运行模式的话，在载入 RDB 文件时，不论键是否过期都会被载入到数据库中。但由于主从服务器在进行数据同步时，从服务器的数据会被清空。所以一般来说，过期键对载入 RDB 文件的从服务器也不会造成影响。

AOF 持久化

AOF 文件写入阶段 - 当 Redis 以 AOF 模式持久化时，如果数据库某个过期键还没被删除，那么 AOF 文件会保留此过期键，当此过期键被删除后，Redis 会向 AOF 文件追加一条 DEL 命令来显式地删除该键值。
AOF 重写阶段 - 执行 AOF 重写时，会对 Redis 中的键值对进行检查，已过期的键不会被保存到重写后的 AOF 文件中，因此不会对 AOF 重写造成任何影响。

【中等】Redis 主从复制时，对过期键会如何处理？

当 Redis 运行在主从模式下时，从库不会进行过期扫描，从库对过期的处理是被动的。也就是即使从库中的 key 过期了，如果有客户端访问从库时，依然可以得到 key 对应的值，像未过期的键值对一样返回。

从库的过期键处理依靠主服务器控制，主库在 key 到期时，会在 AOF 文件里增加一条 del 指令，同步到所有的从库，从库通过执行这条 del 指令来删除过期的 key。

【中等】Redis 中的内存碎片化是什么？如何进行优化？

Redis 内存碎片化是指已分配的内存无法被有效利用，导致内存浪费的现象。

可以通过 Redis 的 INFO memory 命令查看：

1	mem_fragmentation_ratio: 1.86 # 大于 1.5 表示碎片较多

内存碎片的原因：

内存分配器机制 - Redis 使用 Jemalloc 或 glibc 的 malloc 等内存分配器，这些分配器为了性能不会总是精确分配请求的大小，可能分配稍大的块。
键值对频繁修改 - 当键值对被频繁修改（特别是大小变化时），旧的内存空间可能无法重用。例如：字符串值从 128B 改为 256B，原空间不够需要新分配。
键过期/删除 - 删除键释放的内存块可能无法与相邻空闲块合并，这些碎片空间可能无法满足新的大内存请求。
不同大小的数据混合存储 - Redis 存储各种大小的键值对，导致内存中出现大小不一的空闲块。

内存碎片的影响：

内存分配策略：不同的分配器 (Jemalloc/libc 等）碎片率不同
工作负载模式：频繁修改和删除操作会增加碎片
数据大小分布：大小差异大的数据混合存储更容易产生碎片

内存碎片的解决：

重启 Redis（会丢失数据）
使用 MEMORY PURGE 命令（需要特定分配器支持）
配置合理的 maxmemory 和淘汰策略
对于高碎片环境，可考虑使用 Redis 4.0+ 的主动碎片整理功能

Redis 持久化

【中等】Redis 支持哪些持久化方式？

为了追求性能，Redis 的读写都是在内存中完成的。一旦重启，内存中的数据就会清空，为了保证数据不丢失，Redis 支持持久化机制。

Redis 有三种持久化方式

RDB 快照
AOF 日志
混合持久化

RDB

RDB 的实现原理是什么？

生成 RDB 快照时，Redis 可以响应请求吗？

有两个 Redis 命令可以用于生成 RDB 文件：SAVE 和 BGSAVE。

SAVE 命令由服务器进程直接执行保存操作，直到 RDB 创建完成为止。所以该命令“会阻塞”服务器，在阻塞期间，服务器不能响应任何命令请求。

BGSAVE 命令会派生（fork）一个子进程，由子进程负责创建 RDB 文件，服务器进程继续处理命令请求，所以该命令“不会阻塞”服务器。

🔔 【注意】

BGSAVE 命令的实现采用的是写时复制技术（Copy-On-Write，缩写为 CoW）。

BGSAVE 命令执行期间，SAVE、BGSAVE、BGREWRITEAOF 三个命令会被拒绝，以免与当前的 BGSAVE 操作产生竞态条件，降低性能。

AOF

AOF 的实现原理是什么？

为什么先执行命令，再把数据写入日志呢？

Redis 命令请求会先保存到 AOF 缓冲区，再定期写入并同步到 AOF 文件。

AOF 的实现可以分为命令追加（append）、文件写入、文件同步（sync）三个步骤。

命令追加 - 当 Redis 服务器开启 AOF 功能时，服务器在执行完一个写命令后，会以 Redis 命令协议格式将被执行的写命令追加到 AOF 缓冲区的末尾。
文件写入和文件同步
- Redis 的服务器进程就是一个事件循环，这个循环中的文件事件负责接收客户端的命令请求，以及向客户端发送命令回复。而时间事件则负责执行想 serverCron 这样的定时运行的函数。
- 因为服务器在处理文件事件时可能会执行写命令，这些写命令会被追加到 AOF 缓冲区，服务器每次结束事件循环前，都会根据 appendfsync 选项来判断 AOF 缓冲区内容是否需要写入和同步到 AOF 文件中。

先执行命令，再把数据写入 AOF 日志有两个好处：

避免额外的检查开销
不会阻塞当前写操作命令的执行

当然，这样做也会有弊端：

数据可能会丢失：
可能阻塞其他操作：

Redis 命令请求会先保存到 AOF 缓冲区，再定期写入并同步到 AOF 文件。

appendfsync 不同选项决定了不同的持久化行为：

always - 将 AOF 缓冲区中所有内容写入并同步到 AOF 文件。这种方式是最数据最安全的，但也是性能最差的。
no - 将 AOF 缓冲区所有内容写入到 AOF 文件，但并不对 AOF 文件进行同步，何时同步由操作系统决定。这种方式是数据最不安全的，一旦出现故障，未来得及同步的所有数据都会丢失。
everysec - appendfsync 默认选项。将 AOF 缓冲区所有内容写入到 AOF 文件，如果上次同步 AOF 文件的时间距离现在超过一秒钟，那么再次对 AOF 文件进行同步，这个同步操作是有一个线程专门负责执行的。这张方式是前面两种的这种方案——性能足够好，且即使出现故障，仅丢失一秒钟内的数据。

appendfsync 选项的不同值对 AOF 持久化功能的安全性、以及 Redis 服务器的性能有很大的影响。

混合持久化

Redis 4.0 提出了混合使用 AOF 日志和内存快照，也叫混合持久化，既保证了 Redis 重启速度，又降低数据丢失风险。

混合持久化的工作机制

触发时机：在 AOF 重写过程中启用。
执行流程：
1. 子进程将共享内存数据以 RDB 格式写入 AOF 文件（全量数据）。
2. 主线程将操作命令记录到重写缓冲区，再以 AOF 格式追加到 AOF 文件（增量数据）。
3. 替换旧 AOF 文件，新文件包含 RDB（前半部分） + AOF（后半部分）。

混合持久化的优点

重启速度快：优先加载 RDB 部分（全量数据恢复快）。
数据丢失少：后续加载 AOF 部分（增量数据补充）。

混合持久化的缺点

可读性差：AOF 文件包含二进制 RDB 数据，不易阅读。
兼容性差：仅支持 Redis 4.0+ 版本，旧版本无法识别。

【中等】AOF 的重写机制是怎样的？

AOF 日志过大时，怎么办？
AOF 重写流程是怎样的？
AOF 重写时，可以处理请求吗？

知识点

当 AOF 日志过大时，恢复过程就会很久。为了避免此问题，Redis 提供了 AOF 重写机制，即 AOF 日志大小超过所设阈值后，启动 AOF 重写，压缩 AOF 文件。

AOF 重写机制是，读取当前数据库中的所有键值对，然后将每一个键值对用一条命令记录到新的 AOF 日志中，等到全部记录完成后，就使用新的 AOF 日志替换现有的 AOF 日志。

作为一种辅助性功能，显然 Redis 并不想在 AOF 重写时阻塞 Redis 服务接收其他命令。因此，Redis 决定通过 BGREWRITEAOF 命令创建一个子进程，然后由子进程负责对 AOF 文件进行重写，这与 BGSAVE 原理类似。

在执行 BGREWRITEAOF 命令时，Redis 服务器会维护一个 AOF 重写缓冲区。当 AOF 重写子进程开始工作后，Redis 每执行完一个写命令，会同时将这个命令发送给 AOF 缓冲区和 AOF 重写缓冲区。
由于彼此不是在同一个进程中工作，AOF 重写不影响 AOF 写入和同步。当子进程完成创建新 AOF 文件的工作之后，服务器会将重写缓冲区中的所有内容追加到新 AOF 文件的末尾，使得新旧两个 AOF 文件所保存的数据库状态一致。
最后，服务器用新的 AOF 文件替换就的 AOF 文件，以此来完成 AOF 重写操作。

Redis 批处理

【中等】Redis 支持事务吗？

::: info Redis 支持事务吗？

:::

Redis 支持事务。MULTI、EXEC、DISCARD 和 WATCH 是 Redis 事务相关的命令。

MULTI 命令用于开启一个事务，它总是返回 OK。MULTI 执行之后，客户端可以继续向服务器发送任意多条命令，这些命令不会立即被执行，而是被放到一个队列中，当 EXEC 命令被调用时，所有队列中的命令才会被执行。

EXEC 命令负责触发并执行事务中的所有命令。

如果客户端在使用 MULTI 开启了一个事务之后，却因为断线而没有成功执行 EXEC ，那么事务中的所有命令都不会被执行。
另一方面，如果客户端成功在开启事务之后执行 EXEC ，那么事务中的所有命令都会被执行。

当执行 DISCARD 命令时，事务会被放弃，事务队列会被清空，并且客户端会从事务状态中退出。

WATCH 命令可以为 Redis 事务提供 check-and-set （CAS）行为。被 WATCH 的键会被监视，并会发觉这些键是否被改动过了。如果有至少一个被监视的键在 EXEC 执行之前被修改了，那么整个事务都会被取消，EXEC 返回 nil-reply 来表示事务已经失败。

WATCH 可以用于创建 Redis 没有内置的原子操作。

举个例子，以下代码实现了原创的 ZPOP 命令，它可以原子地弹出有序集合中分值（score）最小的元素：

WATCH zset
element = ZRANGE zset 0 0
MULTI
ZREM zset element
EXEC

::: info Redis 事务是严格意义的事务吗？

:::

ACID 是数据库事务正确执行的四个基本要素。

原子性（Atomicity）
- 事务被视为不可分割的最小单元，事务中的所有操作要么全部提交成功，要么全部失败回滚。
- 回滚可以用日志来实现，日志记录着事务所执行的修改操作，在回滚时反向执行这些修改操作即可。
一致性（Consistency）
- 数据库在事务执行前后都保持一致性状态。
- 在一致性状态下，所有事务对一个数据的读取结果都是相同的。
隔离性（Isolation）
- 一个事务所做的修改在最终提交以前，对其它事务是不可见的。
持久性（Durability）
- 一旦事务提交，则其所做的修改将会永远保存到数据库中。即使系统发生崩溃，事务执行的结果也不能丢失。
- 可以通过数据库备份和恢复来实现，在系统发生奔溃时，使用备份的数据库进行数据恢复。

一个支持事务（Transaction）中的数据库系统，必需要具有这四种特性，否则在事务过程（Transaction processing）当中无法保证数据的正确性。

Redis 仅支持“非严格”的事务。所谓“非严格”是指：

Redis 事务保证全部执行命令 - Redis 事务中的多个命令会被打包到事务队列中，然后按先进先出（FIFO）的顺序执行。事务在执行过程中不会被中断，当事务队列中的所有命令都被执行完毕之后，事务才会结束。
Redis 事务不支持回滚 - 如果命令执行失败不会回滚，而是会继续执行下去。

Redis 官方的事务特性文档给出的不支持回滚的理由是：

Redis 命令只会因为错误的语法而失败，或是命令用在了错误类型的键上面。
因为不需要对回滚进行支持，所以 Redis 的内部可以保持简单且快速。

【中等】Redis Pipeline 能保证原子性吗？

先说结论：Redis Pipeline 不保证原子性。

Redis Pipeline（管道）是一种客户端技术，用于将多个 Redis 命令批量发送到服务器，减少网络往返时间（RTT），提高吞吐量。

传统模式：客户端发送一条命令 → 等待响应 → 再发送下一条（高延迟）。
Pipeline 模式：客户端一次性发送多条命令 → 服务器按顺序执行 → 一次性返回所有结果（低延迟）。

核心特性

特性	说明
批量发送	客户端打包多条命令，一次性发送，减少网络开销
非原子性	Pipeline 只是批量发送，不保证所有命令连续执行（可能被其他客户端命令打断）
高性能	相比单条命令模式，吞吐量可提升 5~10 倍
无回滚	如果某条命令失败，不会影响其他命令的执行

注意事项

命令数量控制：避免单次 Pipeline 发送过多命令（建议每批 ≤ 1 万条），否则可能阻塞 Redis。
集群模式限制：Redis Cluster 要求 Pipeline 的所有 Key 必须在同一个 Slot（可用 {hash_tag} 确保，如 user:{123}:name）。
错误处理：Pipeline 返回的是一个列表，需逐条检查命令是否成功。
与事务的区别：Pipeline 不保证原子性。

适用场景

✅ 适合：

批量写入（如日志上报、缓存预热）
批量查询（如获取多个 Key 的值）
对原子性无要求的高并发场景

❌ 不适合：

需要事务保证原子性的操作（改用 MULTI/EXEC）
命令之间有依赖关系（如后一条命令依赖前一条的结果）

【中等】Redis Lua 脚本有什么用？

Redis 从 2.6 版本开始支持执行 Lua 脚本，它的功能和事务非常类似。Redis 的 Lua 脚本提供了一种原子性执行多个命令的方式。也就是说，一段 Lua 脚本执行过程中不会有其他脚本或 Redis 命令同时执行，保证了操作不会被其他指令插入或打扰，这是 pipeline 所不具备的。

并且，Lua 脚本中支持一些简单的逻辑处理比如使用命令读取值并在 Lua 脚本中进行处理，这同样是 pipeline 所不具备的。

不过， Lua 脚本依然存在下面这些缺陷：

如果 Lua 脚本运行时出错并中途结束，之后的操作不会进行，但是之前已经发生的写操作不会撤销，所以即使使用了 Lua 脚本，也不能实现类似数据库回滚的原子性。
Redis Cluster 下 Lua 脚本的原子操作也无法保证，原因同样是无法保证所有的 key 都在同一个 hash slot（哈希槽）上。

另外，Redis 7.0 新增了 Redis functions 特性，你可以将 Redis functions 看作是比 Lua 更强大的脚本。

Redis 高可用

【中等】Redis 如何实现主从复制？

Redis 复制的工作原理？Redis 旧版复制和新版复制有何不同？

Redis 主从节点间如何复制数据？

Redis 的数据一致性是强一致性吗？

（1）旧版复制基于 SYNC 命令实现。分为同步（sync）和命令传播（command propagate）两个操作。这种方式存在缺陷：不能高效处理断线重连后的复制情况。

（2）新版复制基于 PSYNC 命令实现。同步操作分为了两块：

完整重同步（full resychronization） 用于初次复制；
部分重同步（partial resychronization） 用于断线后重复制。
- 主从服务器的复制偏移量（replication offset）
- 主服务器的复制积压缓冲区（replication backlog）
- 服务器的运行 ID

（3）Redis 集群主从节点复制的工作流程：

设置主从服务器
主从服务器建立 TCP 连接。
发送 PING 检查通信状态。
身份验证。
发送端口信息。
同步。
命令传播。

（4）由于主从复制是异步的，具体来说，在主从服务器命令传播阶段，主服务器收到新的写命令后，会发送给从服务器。但是，主服务器并不会等到从服务器实际执行完命令后，再把结果返回给客户端，而是主服务器自己在本地执行完命令后，就会向客户端返回结果了。如果从服务器还没有执行主服务器同步过来的命令，主从服务器间的数据就不一致了。所以，无法实现强一致性保证（主从数据时时刻刻保持一致），数据不一致是难以避免的。

【中等】Redis 哨兵是如何工作的？

Redis 哨兵的功能？

Redis 哨兵的原理？

Redis 哨兵如何选举 Leader？

Redis 如何实现故障转移？

（1）Redis 主从复制模式无法自动故障转移，也就是说，一旦主服务器宕机，需要手动恢复。为了解决此问题，Redis 增加了哨兵模式（Sentinel）。

（2）由一个或多个 Sentinel 实例组成的 Sentinel 系统可以监视任意多个主服务器，以及这些主服务器的所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求。

【中等】Redis 集群是如何工作的？

集群架构

节点类型：
- 主节点 (master)：负责处理槽
- 从节点 (slave)：复制主节点，主节点下线时可接管
节点通信：使用 Gossip 协议 (PING/PONG/MEET/FAIL/PUBLISH 消息）
数据分区：
- 哈希槽分配：CRC16(key) mod 16384
- 槽指派命令：CLUSTER ADDSLOTS
- 所有槽必须分配，否则集群下线

关键机制

请求路由：
- 计算 key 所属槽
- MOVED 错误重定向（永久重定向）
- ASK 错误重定向（临时重定向，迁移过程中使用）
故障转移：
- 故障检测：PING 消息+半数以上主节点确认为 FAIL 状态
- 选主流程：基于 Raft 协议，需要获得 N/2+1 票
- 转移步骤：从节点升级为主节点并接管槽
重新分区：
- 在线迁移槽和键值对
- 使用 redis-trib 工具操作

使用限制

功能限制：
- 批量操作/事务/Pipeline/lua 仅支持相同 slot 的 key
- 不支持多数据库（仅 db0)
- 复制结构只支持一层
规模限制：
- 适合中小规模集群（几个到几十个节点）
- Gossip 协议在大规模集群中传播效率低

【困难】Redis 中的脑裂问题是如何产生的？

在 Redis 主从架构中，部署方式一般是“一主多从”，主节点提供写操作，从节点提供读操作。如果主节点的网络突然发生了问题，它与所有的从节点都失联了，但是此时的主节点和客户端的网络是正常的，这个客户端并不知道 Redis 内部已经出现了问题，还在照样的向这个失联的主节点写数据（过程 A），此时这些数据被旧主节点缓存到了缓冲区里，因为主从节点之间的网络问题，这些数据都是无法同步给从节点的。

这时，哨兵也发现主节点失联了，它就认为主节点挂了（但实际上主节点正常运行，只是网络出问题了），于是哨兵就会在“从节点”中选举出一个 leader 作为主节点，这时集群就有两个主节点了 —— 脑裂出现了。

然后，网络突然好了，哨兵因为之前已经选举出一个新主节点了，它就会把旧主节点降级为从节点（A），然后从节点（A）会向新主节点请求数据同步，因为第一次同步是全量同步的方式，此时的从节点（A）会清空掉自己本地的数据，然后再做全量同步。所以，之前客户端在过程 A 写入的数据就会丢失了，也就是集群产生脑裂数据丢失的问题。

总结一句话就是：由于网络问题，集群节点之间失去联系。主从数据不同步；重新平衡选举，产生两个主服务。等网络恢复，旧主节点会降级为从节点，再与新主节点进行同步复制的时候，由于从节点会清空自己的缓冲区，所以导致之前客户端写入的数据丢失了。

分布式系统的脑裂问题（Split-Brain Problem）是一个严重的一致性问题，通常发生在分布式系统中的节点之间失去通信或部分通信时。这个问题的名称源自脑裂的比喻，就像一个分布式系统被分成多个部分的”脑”，每个部分独立运行，而没有协调一致的方式。

脑裂问题通常发生在以下情况下：

网络分区：当分布式系统中的网络发生问题，导致节点之间无法互相通信或只能部分通信时。这可能是由于网络故障、硬件故障、防火墙配置问题等原因引起的。
节点故障：当分布式系统的某个节点崩溃或出现故障，但其他节点无法确定该节点的状态，可能导致脑裂问题。

脑裂问题的典型情况是，在网络分区或节点故障后，分布式系统的一部分节点认为另一部分节点已经不可用，因此开始采取某种措施，比如选举新的领袖或切换到备份模式。然而，在某些情况下，网络分区可能会解除，或者节点故障可能会自行修复，导致系统中存在多个独立运行的子系统，每个子系统都认为自己是正确的。

这种情况下，脑裂问题可能导致以下问题：

数据不一致性：不同子系统可能具有不同的数据状态，这可能会导致数据不一致性和冲突。
资源冲突：如果不同的子系统尝试访问相同的资源，可能会发生资源冲突和竞争条件。
性能问题：系统中的资源可能被多次分配，从而浪费了资源并降低了性能。

为了解决脑裂问题，分布式系统通常需要采用一些机制，如投票算法、选举协议、心跳检测等，以确保在出现网络分区或节点故障时，系统能够正确地识别和处理问题，并维护一致性。这些机制可以帮助系统中的节点协同工作，避免脑裂问题的发生。然而，脑裂问题是分布式系统设计和管理中的复杂挑战之一，需要细致的规划和测试来确保系统的可靠性和稳定性。

【困难】如何解决 Redis 中的脑裂问题？

当主节点发现从节点下线或者通信超时的总数量小于阈值时，那么禁止主节点进行写数据，直接把错误返回给客户端。

在 Redis 的配置文件中有两个参数我们可以设置：

min-slaves-to-write x，主节点必须要有至少 x 个从节点连接，如果小于这个数，主节点会禁止写数据。
min-slaves-max-lag x，主从数据复制和同步的延迟不能超过 x 秒，如果超过，主节点会禁止写数据。

我们可以把 min-slaves-to-write 和 min-slaves-max-lag 这两个配置项搭配起来使用，分别给它们设置一定的阈值，假设为 N 和 T。

这两个配置项组合后的要求是，主库连接的从库中至少有 N 个从库，和主库进行数据复制时的 ACK 消息延迟不能超过 T 秒，否则，主库就不会再接收客户端的写请求了。

即使原主库是假故障，它在假故障期间也无法响应哨兵心跳，也不能和从库进行同步，自然也就无法和从库进行 ACK 确认了。这样一来，min-slaves-to-write 和 min-slaves-max-lag 的组合要求就无法得到满足，原主库就会被限制接收客户端写请求，客户端也就不能在原主库中写入新数据了。

等到新主库上线时，就只有新主库能接收和处理客户端请求，此时，新写的数据会被直接写到新主库中。而原主库会被哨兵降为从库，即使它的数据被清空了，也不会有新数据丢失。

再来举个例子。

假设我们将 min-slaves-to-write 设置为 1，把 min-slaves-max-lag 设置为 12s，把哨兵的 down-after-milliseconds 设置为 10s，主库因为某些原因卡住了 15s，导致哨兵判断主库客观下线，开始进行主从切换。

同时，因为原主库卡住了 15s，没有一个从库能和原主库在 12s 内进行数据复制，原主库也无法接收客户端请求了。

这样一来，主从切换完成后，也只有新主库能接收请求，不会发生脑裂，也就不会发生数据丢失的问题了。

Redis 架构

【中等】Redis 为什么快？

根据 Redis 官方 Benchmark 文档的描述，Redis 单机 QPS 能达到 10w+，将近是 Mysql 的 10 倍。

Redis 官方 Benchmark QPS 图

Redis 是单线程模型（Redis 6.0 已经支持多线程模型），为什么还能有这么高的并发？

Redis 读写基于内存
IO 多路复用 + 读写单线程模型
- IO 多路复用是利用 select、poll、epoll 可以同时监察多个流的 I/O 事件的能力，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有 I/O 事件时，就从阻塞态中唤醒，于是程序就会轮询一遍所有的流（epoll 是只轮询那些真正发出了事件的流），并且只依次顺序的处理就绪的流，这种做法就避免了大量的无用操作。
- 单线程模型避免了由于并发而产生的线程切换、锁竞争等开销。
- 由于，Redis 读写基于内存，性能很高，所以 CPU 并不是制约 Redis 性能表现的瓶颈所在。更多情况下是受到内存大小和网络 I/O 的限制，所以 Redis 核心网络模型使用单线程并没有什么问题。
高效的数据结构

图来自 Why is redis so fast?

扩展：【视频】Why is Redis so FAST

【中等】Redis 单线程模式是怎样的？

Redis 单线程模式指的是其核心网络模型为单线程模式。这个模式为 IO 多路复用+单线程读写请求，其中，IO 多路复用使得 Redis 可以同时处理多个客户端连接。

Redis 真的只有单线程吗？

Redis 并非真的只有单线。

Redis 的主要工作包括接收客户端请求、解析请求和进行数据读写等操作，是由单线程来执行的，这也是常说 Redis 是单线程程序的原因。
Redis 还启动了 3 个线程来执行文件关闭、AOF 同步写和惰性删除等操作。
此外，Redis 6.0 版本之后引入了多线程来处理网络请求（提高网络 IO 读写性能）。

【中等】Redis 6.0 之后为什么引入了多线程？

随着网络硬件的性能提升，Redis 的性能瓶颈有时会出现在网络 IO 的处理上，也就是说，单个主线程处理网络请求的速度跟不上底层网络硬件的速度。

为了提高网络 I/O 的并行度，Redis 6.0 对于网络 I/O 采用多线程来处理。但是，对于命令的执行，Redis 仍然使用单线程来处理。

Redis 官方表示，Redis 6.0 版本引入的多线程 I/O 特性对性能提升至少是一倍以上。

【中等】什么是 Redis 模块？有什么用？

Redis 从 4.0 版本开始，支持通过 Module 来扩展其功能以满足特殊的需求。这些 Module 以动态链接库（so 文件）的形式被加载到 Redis 中，这是一种非常灵活的动态扩展功能的实现方式，值得借鉴学习！

我们每个人都可以基于 Redis 去定制化开发自己的 Module，比如实现搜索引擎功能、自定义分布式锁和分布式限流。

目前，被 Redis 官方推荐的 Module 有：

RediSearch：用于实现搜索引擎的模块。
RedisJSON：用于处理 JSON 数据的模块。
RedisGraph：用于实现图形数据库的模块。
RedisTimeSeries：用于处理时间序列数据的模块。
RedisBloom：用于实现布隆过滤器的模块。
RedisAI：用于执行深度学习/机器学习模型并管理其数据的模块。
RedisCell：用于实现分布式限流的模块。

关于 Redis 模块的详细介绍，可以查看官方文档：https://redis.io/modules。

【困难】Redis 有哪些巧妙的设计？

Redis 的巧妙设计体现在：

单线程 + 非阻塞 I/O → 高吞吐、低延迟。
精细优化的数据结构 → 节省内存，提升访问速度。
异步化处理（持久化、删除、复制）→ 减少主线程阻塞。
扩展性设计（模块化、集群）→ 适应不同场景需求。

Redis 作为高性能的内存数据库，有许多巧妙的设计理念和实现细节，使其在性能、简洁性和功能性之间取得平衡。以下是 Redis 的一些巧妙设计：

单线程模型（核心命令处理）

Redis 采用单线程处理命令（6.0+ 后支持多线程 I/O，但核心逻辑仍单线程），避免了锁竞争和上下文切换的开销，同时利用以下优化：

非阻塞 I/O：基于 epoll/kqueue 实现高效事件驱动模型。
纯内存操作：绝大多数操作在内存中完成，单线程即可高效处理。
原子性保证：单线程天然支持原子操作，简化了事务、Lua 脚本等实现。

巧妙点：牺牲多线程并行性换取无锁设计的简单性和高性能。

高效数据结构实现

Redis 的核心数据结构经过高度优化：

**SDS (Simple Dynamic String)**：
- 预分配内存、惰性释放，减少内存重分配。
- 二进制安全（可存储任意数据，不像 C 字符串以 \0 结尾）。
**压缩列表 (ziplist)**：
- 对小数据（如短列表、小哈希）使用紧凑存储，节省内存。
**快速列表 (quicklist)**：
- 结合 ziplist 和双向链表，优化 List 的内存和访问效率。
**跳跃表 (skiplist)**：
- 实现 ZSET，支持 O(logN) 范围查询和高效插入。
渐进式 Rehash：
- Hash 扩容时不阻塞服务，分批次迁移数据。

巧妙点：针对不同场景选择最优底层结构，平衡内存和速度。

异步持久化

Redis 提供两种持久化方式：

RDB （快照）：
- fork() 子进程生成快照，主进程继续服务。
- 使用 Copy-On-Write (COW) 机制减少内存开销。
AOF （日志追加）：
- 先执行命令再记录日志，避免日志错误影响数据。
- 支持 AOF Rewrite 压缩日志（类似 RDB 的快照逻辑）。

巧妙点：通过 fork() + COW 实现后台持久化，避免阻塞主线程。

多路复用+零拷贝

I/O 多路复用：
- 使用 epoll/kqueue 监听大量连接，避免线程/进程切换。
零拷贝优化：
- 网络发送数据时，直接引用内存缓冲区，减少拷贝（如 sendfile）。

巧妙点：最大化利用系统调用，减少 CPU 和内存开销。

惰性删除 (Lazy Free)

DEL 命令不立即释放内存，而是异步回收（避免大 Key 删除卡住主线程）。
适用于 UNLINK、FLUSHDB ASYNC 等场景。

巧妙点：用空间换时间，避免同步删除导致服务延迟。

过期键的混合淘汰策略

定期删除：随机抽查部分 Key，清理已过期的。
惰性删除：访问 Key 时检查是否过期，再决定删除。

巧妙点：平衡 CPU 和内存，避免全局扫描影响性能。

模块化设计 (Redis Modules)

支持动态加载模块（如 RedisSearch、RedisGraph），扩展功能而不改核心代码。

巧妙点：保持核心精简，通过插件机制扩展能力。

集群分片的无中心化设计

Gossip 协议：节点间自动发现和状态同步。
**哈希槽 (Hash Slot)**：数据分片到 16384 个槽，而非一致性哈希，简化迁移。

巧妙点：去中心化设计，避免单点瓶颈，支持动态扩缩容。

Redis 优化

【中等】为什么会有慢查询命令？

一个 Redis 命令的执行可以简化为以下 4 步：

发送命令
命令排队
命令执行
返回结果

Redis 慢查询统计的是命令执行这一步骤的耗时，慢查询命令也就是那些命令执行时间较长的命令。

Redis 为什么会有慢查询命令呢？

Redis 中的大部分命令都是 O(1) 时间复杂度，但也有少部分 O(n) 时间复杂度的命令，例如：

KEYS *：会返回所有符合规则的 key。
HGETALL：会返回一个 Hash 中所有的键值对。
LRANGE：会返回 List 中指定范围内的元素。
SMEMBERS：返回 Set 中的所有元素。
SINTER/SUNION/SDIFF：计算多个 Set 的交集/并集/差集。
……

由于这些命令时间复杂度是 O(n)，有时候也会全表扫描，随着 n 的增大，执行耗时也会越长。不过，这些命令并不是一定不能使用，但是需要明确 N 的值。另外，有遍历的需求可以使用 HSCAN、SSCAN、ZSCAN 代替。

除了这些 O(n) 时间复杂度的命令可能会导致慢查询之外，还有一些时间复杂度可能在 O(N) 以上的命令，例如：

ZRANGE/ZREVRANGE：返回指定 Sorted Set 中指定排名范围内的所有元素。时间复杂度为 O(log(n)+m)，n 为所有元素的数量，m 为返回的元素数量，当 m 和 n 相当大时，O(n) 的时间复杂度更小。
ZREMRANGEBYRANK/ZREMRANGEBYSCORE：移除 Sorted Set 中指定排名范围/指定 score 范围内的所有元素。时间复杂度为 O(log(n)+m)，n 为所有元素的数量，m 被删除元素的数量，当 m 和 n 相当大时，O(n) 的时间复杂度更小。
……

【中等】如何找到慢查询命令？

在 redis.conf 文件中，我们可以使用 slowlog-log-slower-than 参数设置耗时命令的阈值，并使用 slowlog-max-len 参数设置耗时命令的最大记录条数。

当 Redis 服务器检测到执行时间超过 slowlog-log-slower-than阈值的命令时，就会将该命令记录在慢查询日志 (slow log) 中，这点和 MySQL 记录慢查询语句类似。当慢查询日志超过设定的最大记录条数之后，Redis 会把最早的执行命令依次舍弃。

⚠️注意：由于慢查询日志会占用一定内存空间，如果设置最大记录条数过大，可能会导致内存占用过高的问题。

slowlog-log-slower-than和slowlog-max-len的默认配置如下（可以自行修改）：

# The following time is expressed in microseconds, so 1000000 is equivalent
# to one second. Note that a negative number disables the slow log, while
# a value of zero forces the logging of every command.
slowlog-log-slower-than 10000

# There is no limit to this length. Just be aware that it will consume memory.
# You can reclaim memory used by the slow log with SLOWLOG RESET.
slowlog-max-len 128

除了修改配置文件之外，你也可以直接通过 CONFIG 命令直接设置：

# 命令执行耗时超过 10000 微妙（即 10 毫秒）就会被记录
CONFIG SET slowlog-log-slower-than 10000
# 只保留最近 128 条耗时命令
CONFIG SET slowlog-max-len 128

获取慢查询日志的内容很简单，直接使用SLOWLOG GET 命令即可。

127.0.0.1:6379> SLOWLOG GET #慢日志查询
 1) 1) (integer) 5
   2) (integer) 1684326682
   3) (integer) 12000
   4) 1) "KEYS"
      2) "*"
   5) "172.17.0.1:61152"
   6) ""
  // ...

慢查询日志中的每个条目都由以下六个值组成：

唯一渐进的日志标识符。
处理记录命令的 Unix 时间戳。
执行所需的时间量，以微秒为单位。
组成命令参数的数组。
客户端 IP 地址和端口。
客户端名称。

SLOWLOG GET 命令默认返回最近 10 条的的慢查询命令，你也自己可以指定返回的慢查询命令的数量 SLOWLOG GET N。

下面是其他比较常用的慢查询相关的命令：

# 返回慢查询命令的数量
127.0.0.1:6379> SLOWLOG LEN
(integer) 128
# 清空慢查询命令
127.0.0.1:6379> SLOWLOG RESET
OK

【中等】Redis 中的 Big Key 问题是什么？如何解决？

什么是 Redis Big Key？

Big Key 并不是指 key 的值很大，而是 key 对应的 value 很大。

一般而言，下面这两种情况被称为 Big Key：

String 类型的值大于 10 KB；
Hash、List、Set、ZSet 类型的元素的个数超过 5000 个，或总大小超过 10MB

Big Key 会造成什么问题？

Big Key 会带来以下四种影响：

内存分布不均：集群模型在 slot 分片均匀情况下，会出现数据和查询倾斜情况，部分有 Big Key 的 Redis 节点占用内存多，QPS 也会比较大。
命令阻塞：Redis 单线程模型，操作大 Key 耗时，阻塞其他命令。
网络传输压力：每次获取 Big Key 产生的网络流量较大，如果一个 key 的大小是 1 MB，每秒访问量为 1000，那么每秒会产生 1000MB 的流量，这对于普通千兆网卡的服务器来说是灾难性的。
客户端超时：由于 Redis 执行命令是单线程处理，然后在操作 Big Key 时会比较耗时，那么就会阻塞 Redis，从客户端这一视角看，就是很久很久都没有响应。

解决方案

（1）开发优化

数据压缩：存储前压缩 Value（如 Gzip、Snappy）。
拆分大对象：将单个大 Key 拆分为多个小 Key（如 user:1000:info → user:1000:basic + user:1000:details）。
优化数据结构：
- 避免巨型 String，改用 Hash、List 等分片存储。
- 使用 ziplist、quicklist 等紧凑结构。

（2）业务调整

精简存储数据：仅保留高频访问字段（如不存用户全部信息，只存 ID + 核心字段）。
逻辑优化：避免业务层生成大 Key（如限制缓存数据大小、分页查询）。

（3）数据分布优化

集群分片：通过 Redis Cluster 分散大 Key 到不同节点。
本地缓存：对冷数据使用本地缓存（如 Caffeine），减少 Redis 压力。

关键点

预防优于治理：在设计和开发阶段规避大 Key。
监控与巡检：通过 redis-cli --bigkeys 或自定义脚本定期检测大 Key。

如何查找 Redis Big Key？

（1）使用 redis-cli --bigkeys

命令：

1	redis-cli -h 127.0.0.1 -p 6379 -a "password" --bigkeys

注意事项：

推荐在从节点执行（主节点执行可能阻塞业务）
低峰期执行 或 加 -i 参数控制扫描间隔（如 -i 0.1 表示每 100ms 扫描一次）

缺点：

只能返回每种数据类型最大的 1 个 Key（无法获取 Top N）
对集合类型只统计元素个数，而非实际内存占用

（2）使用 SCAN + 内存分析命令

遍历所有 Key（避免 KEYS * 阻塞 Redis）：

1	redis-cli --scan --pattern "*" \| while read key; do ...; done

分析 Key 大小：

String：STRLEN $key（字节数）
集合类型（List/Hash/Set/ZSet）：
- 方法 1：LLEN/HLEN/SCARD/ZCARD（元素个数 × 预估元素大小）
- 方法 2（Redis 4.0+）：MEMORY USAGE $key（精确内存占用）

优点：

可自定义筛选条件（如大小 Top 10）
精确计算内存占用

（3）使用 RdbTools 分析 RDB 文件

命令：

1	rdb dump.rdb -c memory --bytes 10240 -f redis.csv # 导出 >10KB 的 Key 到 CSV

适用场景：

离线分析，不影响线上 Redis
精准统计所有 Key 的内存分布

缺点：需要 Redis 生成 RDB 快照

总结：3 种方法对比

方法	适用场景	优点	缺点
`--bigkeys`	快速找出最大 Key	简单易用	结果不全面
`SCAN`+命令	精确分析内存	可定制化	需脚本支持
`RdbTools`	离线全面分析	精准无遗漏	依赖 RDB 文件

推荐组合：

日常监控用 --bigkeys（低峰期执行）
深度分析用 RdbTools（定期检查 RDB）
排查问题时用 SCAN+MEMORY USAGE（实时精准定位）

【中等】如何解决 Redis 中的热点 key 问题？

解决 Redis 中的热点 key 问题的方法：

热点 Key 拆分
- 垂直分片：user:123 → user:123:base + user:123:detail
- 水平分片：product:views → product:views:shard1/shard2
多级缓存：CDN → 本地缓存（Caffeine/Guava） → Redis → DB
读写分离：读请求分流到从节点（配置 replica-read-only yes）

流量控制：

Sentinel / Hystrix 等流控中间件
Redis + Lua 限流

-- 示例：每秒限 100 次访问
local count = redis.call('INCR', KEYS[1])
if count == 1 then redis.call('EXPIRE', KEYS[1], 1) end
return count <= 100

数据预热：定时任务提前加载热点数据到缓存
负载均衡：
- Redis Cluster：分散热点 Key 到不同节点
- 代理层：Twemproxy / Redis Proxy / Nginx 实现负载均衡

参考资料

Redis 面试之数据类型篇

发表于 2020-07-13 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 9.3k 阅读时长 ≈ 8 分钟

Redis 面试之数据类型篇

Redis 数据类型

【简单】Redis 支持哪些数据类型？

Redis 支持五种基本数据类型：String（字符串）、Hash（哈希）、List（列表）、Set（集合）、Zset（有序集合）。
随着 Redis 版本升级，又陆续支持以下数据类型： BitMap（2.2 版新增）、HyperLogLog（2.8 版新增）、GEO（3.2 版新增）、Stream（5.0 版新增）。

扩展：What Redis data structures look like

【简单】Redis 基础数据类型的常见命令有哪些？

String 命令

命令	说明
`SET`	存储一个字符串值
`SETNX`	仅当键不存在时，才存储字符串值
`GET`	获取指定 key 的值
`MGET`	获取一个或多个指定 key 的值
`INCRBY`	将 key 中储存的数字加上指定的增量值
`DECRBY`	将 key 中储存的数字减去指定的减量值

扩展；Redis String 类型官方命令文档

Hash 命令

命令	行为
`HSET`	将指定字段的值设为 value
`HGET`	获取指定字段的值
`HGETALL`	获取所有键值对
`HMSET`	设置多个键值对
`HMGET`	获取所有指定字段的值
`HDEL`	删除指定字段
`HINCRBY`	为指定字段的整数值加上增量
`HKEYS`	获取所有字段

扩展；Redis Hash 类型官方命令文档

List 命令

命令	行为
`LPUSH`	将给定值推入列表的右端。
`RPUSH`	将给定值推入列表的右端。
`LPOP`	从列表的左端弹出一个值，并返回被弹出的值。
`RPOP`	从列表的右端弹出一个值，并返回被弹出的值。
`LRANGE`	获取列表在给定范围上的所有值。
`LINDEX`	获取列表在给定位置上的单个元素。
`LREM`	从列表的左端弹出一个值，并返回被弹出的值。
`LTRIM`	只保留指定区间内的元素，删除其他元素。

扩展；Redis List 类型官方命令文档

Set 命令

命令	行为
`SADD`	将给定元素添加到集合。
`SMEMBERS`	返回集合包含的所有元素。
`SISMEMBER`	检查给定元素是否存在于集合中。
`SREM`	如果给定的元素存在于集合中，那么移除这个元素。

扩展；Redis Set 类型官方命令文档

Zset 命令

命令	行为
`ZADD`	将一个带有给定分值的成员添加到有序集合里面
`ZRANGE`	顺序排序，并返回指定排名区间的成员
`ZREVRANGE`	反序排序，并返回指定排名区间的成员
`ZRANGEBYSCORE`	顺序排序，并返回指定排名区间的成员及其分值
`ZREVRANGEBYSCORE`	反序排序，并返回指定排名区间的成员及其分值
`ZREM`	移除指定的成员
`ZSCORE`	返回指定成员的分值
`ZCARD`	返回所有成员数

扩展；Redis ZSet 类型官方命令文档

【中等】Redis 各数据类型的应用场景？

String（字符串） - 缓存对象、分布式 Session、分布式锁、计数器、限流器、分布式 ID 等。
Hash（哈希） - 缓存对象、购物车等。
List（列表） - 消息队列
Set（集合） - 聚合计算（并集、交集、差集），如点赞、共同关注、抽奖活动等。
Zset（有序集合） - 排序场景，如排行榜、电话和姓名排序等。
BitMap（2.2 版新增） - 二值状态统计的场景，比如签到、判断用户登陆状态、连续签到用户总数等；
HyperLogLog（2.8 版新增） - 海量数据基数统计的场景，比如百万级网页 UV 计数等；
GEO（3.2 版新增） - 存储地理位置信息的场景，比如滴滴叫车；
Stream（5.0 版新增） - 消息队列，相比于基于 List 类型实现的消息队列，有这两个特有的特性：自动生成全局唯一消息 ID，支持以消费组形式消费数据。

【困难】Redis 基础数据类型的底层实现是怎样的？

String 类型 - String 类型的底层数据结构是 SDS。SDS 是 Redis 针对字符串类型的优化，具有以下特性：
- 常数复杂度获取字符串长度
- 杜绝缓冲区溢出
- 减少修改字符串长度时所需的内存重分配次数
List 类型 - 列表对象的编码可以是 ziplist 或者 linkedlist。当列表对象可以同时满足以下两个条件时，列表对象使用 ziplist 编码；否则，使用 linkedlist 编码。
- 列表对象保存的所有字符串元素的长度都小于 64 字节；
- 列表对象保存的元素数量小于 512 个；
Hash 类型 - 哈希对象的编码可以是 ziplist 或者 hashtable。当哈希对象同时满足以下两个条件时，使用 ziplist 编码；否则，使用 hashtable 编码。
- 哈希对象保存的所有键值对的键和值的字符串长度都小于 64 字节；
- 哈希对象保存的键值对数量小于 512 个；
Set 类型 - 集合对象的编码可以是 intset 或者 hashtable。当集合对象可以同时满足以下两个条件时，集合对象使用 intset 编码；否则，使用 hashtable 编码。
- 集合对象保存的所有元素都是整数值；
- 集合对象保存的元素数量不超过 512 个；
Zset 类型 - 有序集合的编码可以是 ziplist 或者 skiplist。当有序集合对象可以同时满足以下两个条件时，有序集合对象使用 ziplist 编码；否则，使用 skiplist 编码。
- 有序集合保存的元素数量小于 128 个；
- 有序集合保存的所有元素成员的长度都小于 64 字节；

【困难】Redis 为什么用 `listpack` 替代 `ziplist`？

**listpack 是 Redis 5.0 引入的优化结构，用来替代 ziplist**，作为 hash、list、zset 数据类型的实现编码之一。

二者对比如下：

特性	`ziplist`	`listpack`
级联更新	可能发生（最坏 `O(n)`）	完全避免（稳定 `O(1)`）
内存占用	预留空间可能浪费	按需分配，更紧凑
安全性	需手动校验边界	内置长度校验，防溢出
版本	Redis 旧版本	Redis 5.0+ 的 Hash、ZSet 等

ziplist 的缺陷

级联更新问题 - 当修改或删除中间某个元素时，可能引发后续所有节点的内存重分配（因为 ziplist 用 前驱节点长度 定位数据）。最坏情况下时间复杂度从 O(1) 退化到 O(n)，影响性能。
内存浪费 - ziplist 为每个节点预留 1~5 字节 存储前驱节点长度（即使实际不需要这么多空间）。对于短小的数据（如小整数），存储开销比例过高。
安全性风险 - ziplist 对内存布局的强依赖可能导致 缓冲区溢出（需严格校验边界）。

listpack 的改进

消除级联更新 - 每个节点 独立存储自身长度（不再依赖前驱节点）。修改任意节点仅影响当前节点，时间复杂度稳定为 O(1)。
更紧凑的内存布局 - 节点长度字段采用 变长编码（类似 Protobuf 的 Varint），根据实际需求分配 1~5 字节。存储小整数时，长度字段仅需 1 字节。
更强的安全性 - 每个节点记录 总长度 和 校验字段，避免解析越界。
兼容性与平滑替换 - listpack 的 API 设计兼容 ziplist，Redis 内部可无缝迁移（如 Hash、ZSet 的底层实现）。

【困难】为什么 Zset 用跳表实现而不是红黑树、B+树？

实现简单性
- 跳表的实现比红黑树简单得多，代码更易于维护和调试。
- 红黑树需要处理复杂的旋转和重新平衡操作，而跳表的平衡是通过概率实现的。
范围查询效率
- 跳表在范围查询（如 zrange）上表现优异，因为它是基于链表的结构，可以线性遍历。
- 红黑树进行范围查询需要中序遍历，相对复杂。
- B+树虽然也擅长范围查询，但实现复杂度更高。
并发性能
- 跳表更容易实现无锁并发操作 (Redis 虽然是单线程，但考虑未来扩展）
- 红黑树的平衡操作涉及大量指针修改，难以实现高效的并发控制
内存效率
- 跳表不需要像 B+ 树那样维护严格的树形结构，内存使用更灵活
- B+树的节点通常设计为填满一定比例，可能造成内存浪费
性能平衡
- 跳表的查询、插入、删除操作时间复杂度都是 O(logN)，与红黑树相当
- 跳表的实际性能在实践中表现良好，特别是对于内存数据结构
Redis 的特殊需求
- Redis 的 Zset 需要同时支持按 score 和按 member 查询，跳表+哈希表的组合完美满足这一需求
- Redis 是内存数据库，不需要考虑 B+树针对磁盘 I/O 优化的特性

【困难】跳表的实现原理是什么？

跳表是一种可以实现二分查找的有序链表，通过多级索引提升查找效率。跳表的查找、插入、删除操作的时间复杂度均为 O(log n)，与平衡二叉树（如红黑树）接近。

对于一个有序数组，可以使用高效的二分查找法，其时间复杂度为 O(log n)。

但是，即使是有序的链表，也只能使用低效的顺序查找，其时间复杂度为 O(n)。

如何提高链表的查找效率呢？

我们可以对链表加一层索引。具体来说，可以每两个结点提取一个结点到上一级，我们把抽出来的那一级叫作索引或索引层。索引节点中通过一个 down 指针，指向下一级结点。通过这样的改造，就可以支持类似二分查找的算法。我们把改造之后的数据结构叫作跳表（Skip list）。

随着数据的不断增长，一级索引层也变得越来越长。此时，我们可以为一级索引再增加一层索引层：二级索引层。

随着数据的膨胀，当二级索引层也变得很长时，我们可以继续为其添加新的索引层。这种链表加多级索引的结构，就是跳表。

跳表的时间复杂度

查找：从最高级索引开始逐层下沉，每层最多遍历 3 个节点，时间复杂度为 O(log n)。
插入：先查找插入位置（O(log n)），再随机生成索引层级（O(log n)），总时间复杂度为 O(log n)。
删除：类似查找过程，删除节点及其索引（O(log n)）。

跳表的空间复杂度

索引节点总数为 n/2 + n/4 + n/8 + … ≈ n，空间复杂度为 **O(n)**。
可通过调整索引密度（如每 3 个节点抽 1 个）减少空间占用，但会牺牲部分查找效率。

【困难】Redis 利用什么机制来实现各种数据结构？

Redis 并没有直接使用这些数据结构来实现键值对数据库，而是基于这些数据结构创建了一个对象系统，这个系统包含字符串对象、列表对象、哈希对象、集合对象和有序集合对象这五种类型的对象。

Redis 数据库中的每个键值对的键和值都是一个对象。Redis 共有字符串、列表、哈希、集合、有序集合五种类型的对象，每种类型的对象至少都有两种或以上的编码方式，不同的编码可以在不同的使用场景上优化对象的使用效率。

服务器在执行某些命令之前，会先检查给定键的类型能否执行指定的命令，而检查一个键的类型就是检查键的值对象的类型。

对象的类型

Redis 使用对象来表示数据库中的键和值。每次当我们在 Redis 的数据库中新创建一个键值对时，我们至少会创建两个对象，一个对象用作键值对的键（键对象），另一个对象用作键值对的值（值对象）。

Redis 中的每个对象都由一个 redisObject 结构表示，该结构中和保存数据有关的三个属性分别是 type 属性、 encoding 属性和 ptr 属性：

typedef struct redisObject {

    // 类型
    unsigned type:4;

    // 编码
    unsigned encoding:4;

    // 指向底层实现数据结构的指针
    void *ptr;

    // ...

} robj;

对象的 type 属性记录了对象的类型，有以下类型：

对象	对象 `type` 属性的值	TYPE 命令的输出
字符串对象	`REDIS_STRING`	`"string"`
列表对象	`REDIS_LIST`	`"list"`
哈希对象	`REDIS_HASH`	`"hash"`
集合对象	`REDIS_SET`	`"set"`
有序集合对象	`REDIS_ZSET`	`"zset"`

Redis 数据库保存的键值对来说，键总是一个字符串对象，而值则可以是字符串对象、列表对象、哈希对象、集合对象或者有序集合对象的其中一种。

对象的编码

对象的 ptr 指针指向对象的底层实现数据结构，而这些数据结构由对象的 encoding 属性决定。

encoding 属性记录了对象所使用的编码，也即是说这个对象使用了什么数据结构作为对象的底层实现。

Redis 中每种类型的对象都至少使用了两种不同的编码，不同的编码可以在不同的使用场景上优化对象的使用效率。

Redis 支持的编码如下所示：

类型	编码	对象	OBJECT ENCODING 命令输出
`REDIS_STRING`	`REDIS_ENCODING_INT`	使用整数值实现的字符串对象。	“int”
`REDIS_STRING`	`REDIS_ENCODING_EMBSTR`	使用 `embstr` 编码的简单动态字符串实现的字符串对象。	“embstr”
`REDIS_STRING`	`REDIS_ENCODING_RAW`	使用简单动态字符串实现的字符串对象。	“raw”
`REDIS_LIST`	`REDIS_ENCODING_ZIPLIST`	使用压缩列表实现的列表对象。	“ziplist”
`REDIS_LIST`	`REDIS_ENCODING_LINKEDLIST`	使用双端链表实现的列表对象。	“linkedlist”
`REDIS_HASH`	`REDIS_ENCODING_ZIPLIST`	使用压缩列表实现的哈希对象。	“ziplist”
`REDIS_HASH`	`REDIS_ENCODING_HT`	使用字典实现的哈希对象。	“hashtable”
`REDIS_SET`	`REDIS_ENCODING_INTSET`	使用整数集合实现的集合对象。	“intset”
`REDIS_SET`	`REDIS_ENCODING_HT`	使用字典实现的集合对象。	“hashtable”
`REDIS_ZSET`	`REDIS_ENCODING_ZIPLIST`	使用压缩列表实现的有序集合对象。	“ziplist”
`REDIS_ZSET`	`REDIS_ENCODING_SKIPLIST`	使用跳表和字典实现的有序集合对象。	“skiplist”

内存回收

由于 C 语言不支持内存回收，Redis 内部实现了一套基于引用计数的内存回收机制。

每个对象的引用计数信息由 redisObject 结构的 refcount 属性记录。当对象的引用计数值变为 0 时，对象所占用的内存会被释放。

对象共享

在 Redis 中，让多个键共享同一个值对象需要执行以下两个步骤：

将数据库键的值指针指向一个现有的值对象；
将被共享的值对象的引用计数增一。

共享对象机制对于节约内存非常有帮助，数据库中保存的相同值对象越多，对象共享机制就能节约越多的内存。

Redis 会在初始化服务器时，共享值为 0 到 9999

对象的空转时长

redisObject 的 lru 属性记录了对象最后一次被命令程序访问的时间。

如果服务器打开了 maxmemory 选项，并且服务器用于回收内存的算法为 volatile-lru 或者 allkeys-lru ，那么当服务器占用的内存数超过了 maxmemory 选项所设置的上限值时，空转时长较高的那部分键会优先被服务器释放，从而回收内存。

【中等】如何使用 Redis 实现排行榜？

各种排行榜，如：内容平台（视频、歌曲、文章）的播放量/收藏量/评分排行榜；电商网站的销售排行榜等等，都可以基于 Redis zset 类型来实现。

我们以博文点赞排名为例，dunwu 发表了五篇博文，分别获得赞为 200、40、100、50、150。

# article:1 文章获得了 200 个赞
> ZADD user:dunwu:ranking 200 article:1
(integer) 1
# article:2 文章获得了 40 个赞
> ZADD user:dunwu:ranking 40 article:2
(integer) 1
# article:3 文章获得了 100 个赞
> ZADD user:dunwu:ranking 100 article:3
(integer) 1
# article:4 文章获得了 50 个赞
> ZADD user:dunwu:ranking 50 article:4
(integer) 1
# article:5 文章获得了 150 个赞
> ZADD user:dunwu:ranking 150 article:5
(integer) 1

文章 article:4 新增一个赞，可以使用 ZINCRBY 命令（为有序集合 key 中元素 member 的分值加上 increment）：

1 2	> ZINCRBY user:dunwu:ranking 1 article:4 "51"

查看某篇文章的赞数，可以使用 ZSCORE 命令（返回有序集合 key 中元素个数）：

1 2	> ZSCORE user:dunwu:ranking article:4 "50"

获取 dunwu 文章赞数最多的 3 篇文章，可以使用 ZREVRANGE 命令（倒序获取有序集合 key 从 start 下标到 stop 下标的元素）：

# WITHSCORES 表示把 score 也显示出来
> ZREVRANGE user:dunwu:ranking 0 2 WITHSCORES
1) "article:1"
2) "200"
3) "article:5"
4) "150"
5) "article:3"
6) "100"

获取 dunwu 100 赞到 200 赞的文章，可以使用 ZRANGEBYSCORE 命令（返回有序集合中指定分数区间内的成员，分数由低到高排序）：

> ZRANGEBYSCORE user:dunwu:ranking 100 200 WITHSCORES
1) "article:3"
2) "100"
3) "article:5"
4) "150"
5) "article:1"
6) "200"

【中等】如何使用 Redis 实现百万级网页 UV 计数？

Redis HyperLogLog 是 Redis 2.8.9 版本新增的数据类型，是一种用于“统计基数”的数据集合类型，基数统计就是指统计一个集合中不重复的元素个数。但要注意，**HyperLogLog 是统计规则是基于概率完成的，不是非常准确，标准误算率是 0.81%**（统计结果为 100 万时，实际可能在 99.19 万~100.81 万之间）。

核心优势

极低内存占用：仅需 12 KB 内存，即可统计接近 2^64 个元素的基数（如 UV 统计）。
适合海量数据：相比 Set/Hash（元素越多内存消耗越大），HyperLogLog 在百万级以上数据场景中优势显著。

适用场景

网页 UV 统计：统计独立访客数（如 page1:uv），尤其适合高并发、大数据量场景。
容忍误差的基数统计：如热门活动页面访问量、广告点击去重等。

如果需要精确统计，则需要转用 Set / Hash，并且不得不消耗

基本命令

添加元素：PFADD key element [element...]

1	PFADD page1:uv user1 user2 user3 # 将用户添加到 UV 统计

获取统计值：PFCOUNT key

1	PFCOUNT page1:uv # 返回近似 UV 数（如 100 万）

【中等】如何使用 Redis 实现布隆过滤器？

布隆过滤器是一种高效的概率数据结构，常用于检测一个元素是否在一个集合中，可以有效减少数据库的查询次数，解决缓存穿透等问题。

可以通过以下两种方式实现布隆过滤器：

使用位图（Bitmap）实现布隆过滤器：

使用 Redis 的位图结构 SETBIT 和 GETBIT 操作来实现布隆过滤器。位图本质上是一个比特数组，用于标识元素是否存在。
对于给定的数据，通过多个哈希函数计算位置索引，将位图中的相应位置设置为 1，表示该元素可能存在。

使用 RedisBloom 模块：

Redis 提供了一个官方模块 RedisBloom，封装了哈希函数、位图大小等操作，可以直接用于创建和管理布隆过滤器。
使用 BF.ADD 来向布隆过滤器添加元素，使用 BF.EXISTS 来检查某个元素是否可能存在。

BitMap

Bitmap，即位图，是一串连续的二进制数组（0 和 1），可以通过偏移量（offset）定位元素。由于 bit 是计算机中最小的单位，使用它进行储存将非常节省空间，特别适合一些数据量大且使用二值统计的场景。例如在一个系统中，不同的用户使用单调递增的用户 ID 表示。40 亿（$$2^{32}$$ = $$410241024*1024$$ ≈ 40 亿）用户只需要 512M 内存就能记住某种状态，例如用户是否已登录。

实际上，BitMap 不是真实的数据结构，而是针对 String 实现的一组位操作。

由于 STRING 是二进制安全的，并且其最大长度是 512 MB，所以 BitMap 能最大设置 $$2^{32}$$ 个不同的 bit。

【示例】判断用户是否登录

Bitmap 提供了 GETBIT、SETBIT 操作，通过一个偏移值 offset 对 bit 数组的 offset 位置的 bit 位进行读写操作，需要注意的是 offset 从 0 开始。

只需要一个 key = login_status 表示存储用户登陆状态集合数据，将用户 ID 作为 offset，在线就设置为 1，下线设置 0。通过 GETBIT判断对应的用户是否在线。50000 万用户只需要 6 MB 的空间。

假如我们要判断 ID = 10086 的用户的登陆情况：

第一步，执行以下指令，表示用户已登录。

1	SETBIT login_status 10086 1

第二步，检查该用户是否登陆，返回值 1 表示已登录。

1	GETBIT login_status 10086

第三步，登出，将 offset 对应的 value 设置成 0。

1	SETBIT login_status 10086 0

RedisBloom

RedisBloom 是 Redis 官方提供的模块，是一种简化的布隆过滤器实现。它提供了更高性能和更低误判率控制。

RedisBloom 常用命令

BF.RESERVE key error_rate capacity：创建布隆过滤器（指定误判率、容量）
BF.ADD key item：添加元素
BF.EXISTS key item：检查元素是否存在（可能误判）
自动扩容：可动态调整数据结构以适应数据增长

RedisBloom 操作示例

创建：

1	BF.RESERVE myBloomFilter 0.01 1000000 # 误判率 1%，容量 100 万

添加元素：

1	BF.ADD myBloomFilter "item1"

检查元素：

1 2	BF.EXISTS myBloomFilter "item1" # 返回 1（可能存在） BF.EXISTS myBloomFilter "item2" # 返回 0（一定不存在）

适用场景

RedisBloom 适合海量数据判断，且允许误判的场景：

爬虫：URL 去重
黑名单：反垃圾邮件（可能误杀）
分布式系统：优化数据查找（如 Hadoop、Cassandra）
推荐系统：避免重复推荐

核心特点

空间高效：节省存储空间
快速查询：O(1) 时间复杂度
误判率可控：通过参数调整

参考资料

MySQL 存储引擎

发表于 2020-07-13 更新于 2025-09-13 分类于数据库，关系型数据库， mysql 本文字数： 6.6k 阅读时长 ≈ 6 分钟

MySQL 存储引擎

::: info 概述

大体来说，MySQL 可以分为 Server 层和存储引擎层两部分。

Server 层包括连接器、查询缓存、解析器、优化器、执行器等，涵盖 MySQL 的大多数核心服务功能，以及所有的内置函数（如日期、时间、数学和加密函数等），所有跨存储引擎的功能都在这一层实现，比如存储过程、触发器、视图等。

存储引擎层负责数据的存储和提取。MySQL 的存储引擎采用了插拔式架构，可以根据需要替换。MySQL 内置了 InnoDB、MyISAM、Memory 等多个存储引擎，现在最常用的存储引擎是 InnoDB，它从 MySQL 5.5.5 版本开始成为了默认存储引擎。

:::

存储引擎简介

在文件系统中，MySQL 将每个数据库（也可以成为 schema）保存为数据目录下的一个子目录。创建表示，MySQL 会在数据库子目录下创建一个和表同名的 .frm 文件保存表的定义。因为 MySQL 使用文件系统的目录和文件来保存数据库和表的定义，大小写敏感性和具体平台密切相关。Windows 中大小写不敏感；类 Unix 中大小写敏感。不同的存储引擎保存数据和索引的方式是不同的，但表的定义则是在 MySQL 服务层统一处理的。

MySQL 的存储引擎采用了插件的形式，每个存储引擎都面向一种特定的数据库应用环境。同时开源的 MySQL 还允许开发人员设置自己的存储引擎。

MySQL 内置的存储引擎

MySQL 内置了以下存储引擎：

InnoDB - InnoDB 是 MySQL 5.5 版本以后的默认存储引擎。
- 优点：支持事务，支持行级锁，支持外键约束等，并发性能不错且支持自动故障恢复。
MyISAM - MyISAM 是 MySQL 5.5 版本以前的默认存储引擎。
- 优点：速度快，占用资源少。
- 缺点：不支持事务，不支持行级锁，不支持外键约束，也不支持自动故障恢复功能。
Memory - 使用系统内存作为存储介质，以便得到更快的响应速度。不过，如果 mysqld 进程崩溃，则会导致所有的数据丢失。因此，Memory 引擎常用于临时表。
NDB - 也被称为 NDB Cluster 存储引擎，主要用于 MySQL Cluster 分布式集群环境，类似于 Oracle 的 RAC 集群。
Archive - Archive 存储引擎有很好的压缩机制，非常适合用于归档数据。
- Archive 存储引擎只支持 INSERT 和 SELECT 操作。
- Archive 存储引擎采用 zlib 算法压缩数据，压缩比可达到 1: 10。
CSV - 可以将 CSV 文件作为 MySQL 的表来处理，但这种表不支持索引。

如何选择合适的存储引擎

大多数情况下，InnoDB 都是正确的选择，除非需要用到 InnoDB 不具备的特性。

如果应用需要选择 InnoDB 以外的存储引擎，可以考虑以下因素：

事务：如果业务场景是 OLTP，则 InnoDB 是首选存储引擎。如果不需要支持事务，且主要是 SELECT 和 INSERT 操作，MyISAM 是不错的选择。所以，如果 MySQL 部署方式为主备模式，并进行读写分离。那么可以这么做：主节点只支持写操作，默认引擎为 InnoDB；备节点只支持读操作，默认引擎为 MyISAM。
并发：MyISAM 只支持表级锁，而 InnoDB 还支持行级锁。所以，InnoDB 并发性能更高。
外键：InnoDB 支持外键。
备份：InnoDB 支持在线热备份。
崩溃恢复：MyISAM 崩溃后发生损坏的概率比 InnoDB 高很多，而且恢复的速度也更慢。
其它特性：MyISAM 支持压缩表和空间数据索引。

存储引擎相关操作

查看存储引擎命令

-- 查看支持的存储引擎
SHOW ENGINES;

-- 查看默认的存储引擎
SHOW VARIABLES LIKE 'storage_engine';

-- 查看某表所使用的存储引擎
SHOW CREATE TABLE `table_name`;

-- 查看某数据库中的某表所使用的存储引擎
SHOW TABLE STATUS LIKE 'table_name';
SHOW TABLE STATUS FROM `database_name` WHERE `name` = "table_name";

设置存储引擎命令

-- 建表时指定存储引擎，如果不显示指定，默认是 INNODB
CREATE TABLE t1 (i INT) ENGINE = INNODB;
CREATE TABLE t2 (i INT) ENGINE = CSV;
CREATE TABLE t3 (i INT) ENGINE = MEMORY;

-- 修改存储引擎
ALTER TABLE t ENGINE = InnoDB;

-- 修改默认存储引擎，也可以在配置文件 my.cnf 中修改默认引擎
SET default_storage_engine=NDBCLUSTER;

默认情况下，每当 CREATE TABLE 或 ALTER TABLE 不能使用默认存储引擎时，都会生成一个警告。为了防止在所需的引擎不可用时出现令人困惑的意外行为，可以启用 NO_ENGINE_SUBSTITUTION SQL 模式。如果所需的引擎不可用，则此设置将产生错误而不是警告，并且不会创建或更改表

InnoDB

InnoDB 简介

InnoDB 是 MySQL 5.5 版本以后的默认存储引擎。只有在需要 InnoDB 不支持的特性时，才考虑使用其它存储引擎。

InnoDB 也使用 B+Tree 作为索引结构，但具体实现方式却与 MyISAM 截然不同。MyISAM 索引文件和数据文件是分离的，索引文件仅保存数据记录的地址。而在 InnoDB 中，表数据文件本身就是按 B+Tree 组织的一个索引结构，这棵树的叶节点 data 域保存了完整的数据记录。这个索引的 key 是数据表的主键，因此InnoDB 表数据文件本身就是主索引。

InnoDB 采用 MVCC 来支持高并发，并且实现了四个标准的隔离级别。其默认级别是可重复读（REPEATABLE READ），并且通过间隙锁（next-key locking）防止幻读。

InnoDB 是基于聚簇索引建立的，与其他存储引擎有很大不同。在索引中保存了数据，从而避免直接读取磁盘，因此对查询性能有很大的提升。

内部做了很多优化，包括从磁盘读取数据时采用的可预测性读、能够加快读操作并且自动创建的自适应哈希索引、能够加速插入操作的插入缓冲区等。

支持真正的在线热备份。其它存储引擎不支持在线热备份，要获取一致性视图需要停止对所有表的写入，而在读写混合场景中，停止写入可能也意味着停止读取。

InnoDB 物理文件结构为：

.frm 文件：与表相关的元数据信息都存放在 frm 文件，包括表结构的定义信息等。
.ibd 文件或 .ibdata 文件：这两种文件都是存放 InnoDB 数据的文件，之所以有两种文件形式存放 InnoDB 的数据，是因为 InnoDB 的数据存储方式能够通过配置来决定是使用共享表空间存放存储数据，还是用独享表空间存放存储数据。
- 独享表空间存储方式使用.ibd文件，并且每个表一个.ibd文件
- 共享表空间存储方式使用.ibdata文件，所有表共同使用一个.ibdata文件（或多个，可自己配置）

InnoDB 存储架构

InnoDB 存储架构分为内存结构和磁盘结构。

InnoDB 内存结构的核心组件有：

Buffer Pool
Change Buffer
Adaptive Hash Index
Log Buffer

InnoDB 磁盘结构的核心组件有：

Tablespace
Doublewrite Buffer
redo log
undo log

InnoDB 表空间

行（row）

数据库表中的记录都是按行（row）进行存放的，每行记录根据不同的行格式，有不同的存储结构。

后面我们详细介绍 InnoDB 存储引擎的行格式，也是本文重点介绍的内容。

页（page）

记录是按照行来存储的，但是数据库的读取并不以「行」为单位，否则一次读取（也就是一次 I/O 操作）只能处理一行数据，效率会非常低。

因此，InnoDB 的数据是按「页」为单位来读写的，也就是说，当需要读一条记录的时候，并不是将这个行记录从磁盘读出来，而是以页为单位，将其整体读入内存。

默认每个页的大小为 16KB，也就是最多能保证 16KB 的连续存储空间。

页是 InnoDB 存储引擎磁盘管理的最小单元，意味着数据库每次读写都是以 16KB 为单位的，一次最少从磁盘中读取 16K 的内容到内存中，一次最少把内存中的 16K 内容刷新到磁盘中。

页的类型有很多，常见的有数据页、undo 日志页、溢出页等等。数据表中的行记录是用「数据页」来管理的，数据页的结构这里我就不讲细说了，之前文章有说过，感兴趣的可以去看这篇文章：换一个角度看 B+ 树(opens new window)

总之知道表中的记录存储在「数据页」里面就行。

区（extent）

我们知道 InnoDB 存储引擎是用 B+ 树来组织数据的。

B+ 树中每一层都是通过双向链表连接起来的，如果是以页为单位来分配存储空间，那么链表中相邻的两个页之间的物理位置并不是连续的，可能离得非常远，那么磁盘查询时就会有大量的随机 I/O，随机 I/O 是非常慢的。

解决这个问题也很简单，就是让链表中相邻的页的物理位置也相邻，这样就可以使用顺序 I/O 了，那么在范围查询（扫描叶子节点）的时候性能就会很高。

那具体怎么解决呢？

在表中数据量大的时候，为某个索引分配空间的时候就不再按照页为单位分配了，而是按照区（extent）为单位分配。每个区的大小为 1MB，对于 16KB 的页来说，连续的 64 个页会被划为一个区，这样就使得链表中相邻的页的物理位置也相邻，就能使用顺序 I/O 了。

段（segment）

表空间是由各个段（segment）组成的，段是由多个区（extent）组成的。段一般分为数据段、索引段和回滚段等。

索引段：存放 B + 树的非叶子节点的区的集合；
数据段：存放 B + 树的叶子节点的区的集合；
回滚段：存放的是回滚数据的区的集合。

好了，终于说完表空间的结构了。接下来，就具体讲一下 InnoDB 的行格式了。

之所以要绕一大圈才讲行记录的格式，主要是想让大家知道行记录是存储在哪个文件，以及行记录在这个表空间文件中的哪个区域，有一个从上往下切入的视角，这样理解起来不会觉得很抽象。

InnoDB 内存结构

Buffer Pool

Buffer Pool 用于加速数据的访问和修改，通过将热点数据缓存在内存的方法，最大限度地减少磁盘 IO，加速热点数据的读和写。

Buffer Pool 中数据以页为存储单位，其实现数据结构是以页为单位的单链表。

由于内存的空间限制，Buffer Pool 仅能容纳最热点的数据。Buffer Pool 使用最近最少使用算法（Least Recent Used，LRU）算法淘汰非热点数据页。

依据时间局部性原理与空间局部性原理，Buffer Pool 在存储当前活动数据页的时候，会以预读 Read-ahead 的方式缓存目标数据页临近的数据页。

预读机制带来预读失败的问题，InnoDB 采用分代机制解决预读失败问题：将 Buffer Pool 分为 New SubList 和 Old SubList 两部分，将最新读取的数据页置于 Old SubList 头部，Old SubList 中的数据再次被访问到才会置于 New SubList 头部；预读失败的冷数据将更快地从 Old SubList 中淘汰，而不会影响到 New SubList 中原有的热数据。

预读失败问题可以引申到缓冲池污染问题，InnoDB 采用时间窗口（Time Window）机制解决缓冲池污染问题：对于 Old SubList 中的数据页，必须在 Old SubList 中停留到达指定时间之后再次被访问到，才能转移到 New SubList 中，默认窗口大小是 1s。

对于 Buffer Pool 中数据的查询，InnoDB 直接读取返回；对于 Buffer Pool 中数据的修改，InnoDB 直接在 Buffer Pool 中修改，并将修改写入 redo Log 中，当数据页被 LRU 算法淘汰时写入磁盘，若持久化前系统崩溃，则在重启后使用 redo Log 进行恢复。

Change Buffer

Change Buffer 用于加速非热点数据中二级索引的写入操作。由于二级索引数据的不连续性，导致修改二级索引时需要进行频繁的磁盘 IO 消耗大量性能，Change Buffer 缓冲对二级索引的修改操作，同时将写操作录入 redo log 中，在缓冲到一定量或系统较空闲时进行 ibuf merge 操作将修改写入磁盘中。Change Buffer 在系统表空间中有相应的持久化区域。

Change Buffer 大小默认占 Buffer Pool 的 25%，在引擎启动时便初始化完成。其物理结构为一棵名为 ibuf 的 B Tree。Change Buffer 的使用条件为：

InnoDB 开启 innodb_change_buffering，且该表当前没有 flush 操作。
仅对二级索引树的叶子节点进行修改，且该索引页不在 Buffer Pool 中。
对于 Unique 二级索引，仅删除操作可以缓冲。

其 ibuf merge 时机为：

用户使用该二级索引进行查询时。
缓存插入操作时，预估到 page 空间不足可能导致索引页分裂时。
本次缓存操作将导致 ibuf btree 页分裂，且分类后 Change Buffer 大小将超出限制时。
master 线程发起 merge 命令时。
用户对该表进行 flush 操作时。

Adaptive Hash Index

自适应哈希索引（Adaptive Hash Index）用于实现对于热数据页的一次查询。使用聚簇索引进行数据页定位的时候需要根据索引树的高度从根节点走到叶子节点，通常需要 3 到 4 次查询才能定位数据。InnoDB 根据对索引使用情况的分析和索引字段的分析，通过自调优 Self-tuning 的方式为索引页建立或者删除哈希索引。

AHI 所作用的目标是频繁查询的数据页和索引页，而由于数据页是聚簇索引的一部分，因此 AHI 是建立在索引之上的索引，对于二级索引，若命中 AHI，则将直接从 AHI 获取二级索引页的记录指针，再根据主键沿着聚簇索引查找数据；若聚簇索引查询同样命中 AHI，则直接返回目标数据页的记录指针，此时就可以根据记录指针直接定位数据页。

AHI 的大小为 Buffer Pool 的 1/64，再 MySql 5.7 之后支持分区，以减少对于全局 AHI 锁的竞争，默认分区数为 8。

Log Buffer

Log Buffer 是用于缓冲待写入磁盘日志文件的数据。InnoDB 的所有修改操作都会被写入 redo log、undo log 等日志文件，如果每次都直接写入磁盘，会引发大量 IO。Log Buffer 正是针对此进行了优化：先将修改操作缓冲于此内存区域，然后定期批量刷新到磁盘。

日志缓冲区大小可以由配置 innodb_log_buffer_size 控制，默认大小为 16MB。

MyISAM

MyISAM 是 MySQL 5.5 版本以前的默认存储引擎。

MyISAM 设计简单，数据以紧密格式存储。对于只读数据，或者表比较小、可以容忍修复操作，则依然可以使用 MyISAM。

MyISAM 引擎使用 B+Tree 作为索引结构，叶节点的 data 域存放的是数据记录的地址。

MyISAM 提供了大量的特性，包括：全文索引、压缩表、空间函数等。但是，MyISAM 不支持事务和行级锁。并且 MyISAM 不支持崩溃后的安全恢复。

MyISAM 物理文件结构为：

.frm文件：与表相关的元数据信息都存放在 frm 文件，包括表结构的定义信息等。
.MYD (MYData) 文件：MyISAM 存储引擎专用，用于存储 MyISAM 表的数据。
.MYI (MYIndex)文件：MyISAM 存储引擎专用，用于存储 MyISAM 表的索引相关信息。

InnoDB vs. MyISAM

InnoDB 和 MyISAM 的对比：

对比项	MyISAM	InnoDB
外键	不支持	支持
事务	不支持	支持四种事务隔离级别
锁粒度	支持表级锁	支持表级锁、行级锁
索引	采用 B+ 树索引（非聚簇索引）	采用 B+ 树索引（聚簇索引）
表空间	小	大
关注点	性能	事务
计数器	维护了计数器，`SELECT COUNT(*)` 效率为 `O(1)`	没有维护计数器，需要全表扫描
自动故障恢复	不支持	支持（依赖于 redo log）

参考资料

Java I/O 之工具类

发表于 2020-06-30 更新于 2025-09-13 分类于 Java ， JavaCore ， IO 本文字数： 20k 阅读时长 ≈ 19 分钟

Java I/O 之 BIO

BIO

BIO（blocking IO）即阻塞 IO。指的主要是传统的 java.io 包，它基于流模型实现。流从概念上来说是一个连续的数据流。当程序需要读数据的时候就需要使用输入流读取数据，当需要往外写数据的时候就需要输出流。

java.io 包提供了我们最熟知的一些 IO 功能，比如 File 抽象、输入输出流等。交互方式是同步、阻塞的方式，也就是说，在读取输入流或者写入输出流时，在读、写动作完成之前，线程会一直阻塞在那里，它们之间的调用是可靠的线性顺序。很多时候，人们也把 java.net 下面提供的部分网络 API，比如 Socket、ServerSocket、HttpURLConnection 也归类到同步阻塞 IO 类库，因为网络通信同样是 IO 行为。

BIO 中操作的流主要有两大类，字节流和字符流，两类根据流的方向都可以分为输入流和输出流。

字节流
- 输入字节流：InputStream
- 输出字节流：OutputStream
字符流
- 输入字符流：Reader
- 输出字符流：Writer

字节流

字节流主要操作字节数据或二进制对象。

字节流有两个核心抽象类：InputStream 和 OutputStream。所有的字节流类都继承自这两个抽象类。

InputStream

InputStream用于从源头（通常是文件）读取数据（字节信息）到内存中，java.io.InputStream抽象类是所有字节输入流的父类。

InputStream 常用方法：

read()：返回输入流中下一个字节的数据。返回的值介于 0 到 255 之间。如果未读取任何字节，则代码返回 -1 ，表示文件结束。
read(byte b[ ]) : 从输入流中读取一些字节存储到数组 b 中。如果数组 b 的长度为零，则不读取。如果没有可用字节读取，返回 -1。如果有可用字节读取，则最多读取的字节数最多等于 b.length ，返回读取的字节数。这个方法等价于 read(b, 0, b.length)。
read(byte b[], int off, int len)：在read(byte b[ ]) 方法的基础上增加了 off 参数（偏移量）和 len 参数（要读取的最大字节数）。
skip(long n)：忽略输入流中的 n 个字节 , 返回实际忽略的字节数。
available()：返回输入流中可以读取的字节数。
close()：关闭输入流释放相关的系统资源。

从 Java 9 开始，InputStream 新增加了多个实用的方法：

readAllBytes()：读取输入流中的所有字节，返回字节数组。
readNBytes(byte[] b, int off, int len)：阻塞直到读取 len 个字节。
transferTo(OutputStream out)：将所有字节从一个输入流传递到一个输出流。

OutputStream

OutputStream 用于将数据（字节信息）写入到目的地（通常是文件），java.io.OutputStream抽象类是所有字节输出流的父类。

OutputStream 常用方法：

write(int b)：将特定字节写入输出流。
write(byte b[ ]) : 将数组b 写入到输出流，等价于 write(b, 0, b.length) 。
write(byte[] b, int off, int len) : 在write(byte b[ ]) 方法的基础上增加了 off 参数（偏移量）和 len 参数（要读取的最大字节数）。
flush()：刷新此输出流并强制写出所有缓冲的输出字节。
close()：关闭输出流释放相关的系统资源。

文件字节流

FileOutputStream 和 FileInputStream 提供了读写字节到文件的能力。

文件流操作一般步骤：

使用 File 类绑定一个文件。
把 File 对象绑定到流对象上。
进行读或写操作。
关闭流

FileOutputStream 和 FileInputStream 示例：

public class FileStreamDemo {

    private static final String FILEPATH = "temp.log";

    public static void main(String[] args) throws Exception {
        write(FILEPATH);
        read(FILEPATH);
    }

    public static void write(String filepath) throws IOException {
        // 第 1 步、使用 File 类找到一个文件
        File f = new File(filepath);

        // 第 2 步、通过子类实例化父类对象
        OutputStream out = new FileOutputStream(f);
        // 实例化时，默认为覆盖原文件内容方式；如果添加 true 参数，则变为对原文件追加内容的方式。
        // OutputStream out = new FileOutputStream(f, true);

        // 第 3 步、进行写操作
        String str = "Hello World\n";
        byte[] bytes = str.getBytes();
        out.write(bytes);

        // 第 4 步、关闭输出流
        out.close();
    }

    public static void read(String filepath) throws IOException {
        // 第 1 步、使用 File 类找到一个文件
        File f = new File(filepath);

        // 第 2 步、通过子类实例化父类对象
        InputStream input = new FileInputStream(f);

        // 第 3 步、进行读操作
        // 有三种读取方式，体会其差异
        byte[] bytes = new byte[(int) f.length()];
        int len = input.read(bytes); // 读取内容
        System.out.println("读入数据的长度：" + len);

        // 第 4 步、关闭输入流
        input.close();
        System.out.println("内容为：\n" + new String(bytes));
    }

}

内存字节流

ByteArrayInputStream 和 ByteArrayOutputStream 是用来完成内存的输入和输出功能。

内存操作流一般在生成一些临时信息时才使用。如果临时信息保存在文件中，还需要在有效期过后删除文件，这样比较麻烦。

ByteArrayInputStream 和 ByteArrayOutputStream 示例：

public class ByteArrayStreamDemo {

    public static void main(String[] args) {
        String str = "HELLOWORLD"; // 定义一个字符串，全部由大写字母组成
        ByteArrayInputStream bis = new ByteArrayInputStream(str.getBytes());
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        // 准备从内存 ByteArrayInputStream 中读取内容
        int temp = 0;
        while ((temp = bis.read()) != -1) {
            char c = (char) temp; // 读取的数字变为字符
            bos.write(Character.toLowerCase(c)); // 将字符变为小写
        }
        // 所有的数据就全部都在 ByteArrayOutputStream 中
        String newStr = bos.toString(); // 取出内容
        try {
            bis.close();
            bos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        System.out.println(newStr);
    }

}

管道流

管道流的主要作用是可以进行两个线程间的通信。

如果要进行管道通信，则必须把 PipedOutputStream 连接在 PipedInputStream 上。为此，PipedOutputStream 中提供了 connect() 方法。

public class PipedStreamDemo {

    public static void main(String[] args) {
        Send s = new Send();
        Receive r = new Receive();
        try {
            s.getPos().connect(r.getPis()); // 连接管道
        } catch (IOException e) {
            e.printStackTrace();
        }
        new Thread(s).start(); // 启动线程
        new Thread(r).start(); // 启动线程
    }

    static class Send implements Runnable {

        private PipedOutputStream pos = null;

        Send() {
            pos = new PipedOutputStream(); // 实例化输出流
        }

        @Override
        public void run() {
            String str = "Hello World!!!";
            try {
                pos.write(str.getBytes());
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                pos.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        /**
         * 得到此线程的管道输出流
         */
        PipedOutputStream getPos() {
            return pos;
        }

    }

    static class Receive implements Runnable {

        private PipedInputStream pis = null;

        Receive() {
            pis = new PipedInputStream();
        }

        @Override
        public void run() {
            byte[] b = new byte[1024];
            int len = 0;
            try {
                len = pis.read(b);
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                pis.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("接收的内容为：" + new String(b, 0, len));
        }

        /**
         * 得到此线程的管道输入流
         */
        PipedInputStream getPis() {
            return pis;
        }

    }

}

对象字节流

ObjectInputStream 和 ObjectOutputStream 是对象输入输出流，一般用于对象序列化。

这里不展开叙述，想了解详细内容和示例可以参考：[Java 序列化](03.Java 序列化。md)

数据操作流

数据操作流提供了格式化读入和输出数据的方法，分别为 DataInputStream 和 DataOutputStream。

DataInputStream 和 DataOutputStream 格式化读写数据示例：

public class DataStreamDemo {

    public static final String FILEPATH = "temp.log";

    public static void main(String[] args) throws IOException {
        write(FILEPATH);
        read(FILEPATH);
    }

    private static void write(String filepath) throws IOException {
        // 1. 使用 File 类绑定一个文件
        File f = new File(filepath);

        // 2. 把 File 对象绑定到流对象上
        DataOutputStream dos = new DataOutputStream(new FileOutputStream(f));

        // 3. 进行读或写操作
        String[] names = { "衬衣", "手套", "围巾" };
        float[] prices = { 98.3f, 30.3f, 50.5f };
        int[] nums = { 3, 2, 1 };
        for (int i = 0; i < names.length; i++) {
            dos.writeChars(names[i]);
            dos.writeChar('\t');
            dos.writeFloat(prices[i]);
            dos.writeChar('\t');
            dos.writeInt(nums[i]);
            dos.writeChar('\n');
        }

        // 4. 关闭流
        dos.close();
    }

    private static void read(String filepath) throws IOException {
        // 1. 使用 File 类绑定一个文件
        File f = new File(filepath);

        // 2. 把 File 对象绑定到流对象上
        DataInputStream dis = new DataInputStream(new FileInputStream(f));

        // 3. 进行读或写操作
        String name = null; // 接收名称
        float price = 0.0f; // 接收价格
        int num = 0; // 接收数量
        char[] temp = null; // 接收商品名称
        int len = 0; // 保存读取数据的个数
        char c = 0; // '\u0000'
        try {
            while (true) {
                temp = new char[200]; // 开辟空间
                len = 0;
                while ((c = dis.readChar()) != '\t') { // 接收内容
                    temp[len] = c;
                    len++; // 读取长度加 1
                }
                name = new String(temp, 0, len); // 将字符数组变为 String
                price = dis.readFloat(); // 读取价格
                dis.readChar(); // 读取、t
                num = dis.readInt(); // 读取 int
                dis.readChar(); // 读取、n
                System.out.printf("名称：%s；价格：%5.2f；数量：%d\n", name, price, num);
            }
        } catch (EOFException e) {
            System.out.println("结束");
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 4. 关闭流
        dis.close();
    }

}

合并流

合并流的主要功能是将多个 InputStream 合并为一个 InputStream 流。合并流的功能由 SequenceInputStream 完成。

public class SequenceInputStreamDemo {

    public static void main(String[] args) throws Exception {

        InputStream is1 = new FileInputStream("temp1.log");
        InputStream is2 = new FileInputStream("temp2.log");
        SequenceInputStream sis = new SequenceInputStream(is1, is2);

        int temp = 0; // 接收内容
        OutputStream os = new FileOutputStream("temp3.logt");
        while ((temp = sis.read()) != -1) { // 循环输出
            os.write(temp); // 保存内容
        }

        sis.close(); // 关闭合并流
        is1.close(); // 关闭输入流 1
        is2.close(); // 关闭输入流 2
        os.close(); // 关闭输出流
    }

}

字符流

字符流主要操作字符，一般用于处理文本数据。

字符流有两个核心类：Reader 类和 Writer 。所有的字符流类都继承自这两个抽象类。

文件字符流

文件字符流 FileReader 和 FileWriter 可以向文件读写文本数据。

FileReader 和 FileWriter 读写文件示例：

public class FileReadWriteDemo {

    private static final String FILEPATH = "temp.log";

    public static void main(String[] args) throws IOException {
        write(FILEPATH);
        System.out.println("内容为：" + new String(read(FILEPATH)));
    }

    public static void write(String filepath) throws IOException {
        // 1. 使用 File 类绑定一个文件
        File f = new File(filepath);

        // 2. 把 File 对象绑定到流对象上
        Writer out = new FileWriter(f);
        // Writer out = new FileWriter(f, true); // 追加内容方式

        // 3. 进行读或写操作
        String str = "Hello World!!!\r\n";
        out.write(str);

        // 4. 关闭流
        // 字符流操作时使用了缓冲区，并在关闭字符流时会强制将缓冲区内容输出
        // 如果不关闭流，则缓冲区的内容是无法输出的
        // 如果想在不关闭流时，将缓冲区内容输出，可以使用 flush 强制清空缓冲区
        out.flush();
        out.close();
    }

    public static char[] read(String filepath) throws IOException {
        // 1. 使用 File 类绑定一个文件
        File f = new File(filepath);

        // 2. 把 File 对象绑定到流对象上
        Reader input = new FileReader(f);

        // 3. 进行读或写操作
        int temp = 0; // 接收每一个内容
        int len = 0; // 读取内容
        char[] c = new char[1024];
        while ((temp = input.read()) != -1) {
            // 如果不是-1 就表示还有内容，可以继续读取
            c[len] = (char) temp;
            len++;
        }
        System.out.println("文件字符数为：" + len);

        // 4. 关闭流
        input.close();

        return c;
    }

}

字节流转换字符流

我们可以在程序中通过 InputStream 和 Reader 从数据源中读取数据，然后也可以在程序中将数据通过 OutputStream 和 Writer 输出到目标媒介中

使用 InputStreamReader 可以将输入字节流转化为输入字符流；使用OutputStreamWriter可以将输出字节流转化为输出字符流。

OutputStreamWriter 示例：

public class OutputStreamWriterDemo {

    public static void main(String[] args) throws IOException {
        File f = new File("temp.log");
        Writer out = new OutputStreamWriter(new FileOutputStream(f));
        out.write("hello world!!");
        out.close();
    }

}

InputStreamReader 示例：

public class InputStreamReaderDemo {

    public static void main(String[] args) throws IOException {
        File f = new File("temp.log");
        Reader reader = new InputStreamReader(new FileInputStream(f));
        char[] c = new char[1024];
        int len = reader.read(c);
        reader.close();
        System.out.println(new String(c, 0, len));
    }

}

字节流 vs. 字符流

相同点：

字节流和字符流都有 read()、write()、flush()、close() 这样的方法，这决定了它们的操作方式近似。

不同点：

数据类型
- 字节流的数据是字节（二进制对象）。主要核心类是 InputStream 类和 OutputStream 类。
- 字符流的数据是字符。主要核心类是 Reader 类和 Writer 类。
缓冲区
- 字节流在操作时本身不会用到缓冲区（内存），是文件直接操作的。
- 字符流在操作时是使用了缓冲区，通过缓冲区再操作文件。

选择：

所有的文件在硬盘或传输时都是以字节方式保存的，例如图片，影音文件等都是按字节方式存储的。字符流无法读写这些文件。

所以，除了纯文本数据文件使用字符流以外，其他文件类型都应该使用字节流方式。

I/O 工具类

File

File 类是 java.io 包中唯一对文件本身进行操作的类。它可以对文件、目录进行增删查操作。

createNewFille

可以使用 createNewFille() 方法创建一个新文件。

注：

Windows 中使用反斜杠表示目录的分隔符 \。~~~~~~~~

Linux 中使用正斜杠表示目录的分隔符 /。

最好的做法是使用 File.separator 静态常量，可以根据所在操作系统选取对应的分隔符。

【示例】创建文件

1 2	File f = new File(filename); boolean flag = f.createNewFile();

mkdir

可以使用 mkdir() 来创建文件夹，但是如果要创建的目录的父路径不存在，则无法创建成功。

如果要解决这个问题，可以使用 mkdirs()，当父路径不存在时，会连同上级目录都一并创建。

【示例】创建目录

1 2	File f = new File(filename); boolean flag = f.mkdir();

delete

可以使用 delete() 来删除文件或目录。

需要注意的是，如果删除的是目录，且目录不为空，直接用 delete() 删除会失败。

【示例】删除文件或目录

1 2	File f = new File(filename); boolean flag = f.delete();

list 和 listFiles

File 中给出了两种列出文件夹内容的方法：

list(): 列出全部名称，返回一个字符串数组。
listFiles(): 列出完整的路径，返回一个 File 对象数组。

list() 示例：

1 2	File f = new File(filename); String str[] = f.list();

listFiles() 示例：

1 2	File f = new File(filename); File files[] = f.listFiles();

RandomAccessFile

注：RandomAccessFile 类虽然可以实现对文件内容的读写操作，但是比较复杂。所以一般操作文件内容往往会使用字节流或字符流方式。

RandomAccessFile 类是随机读取类，它是一个完全独立的类。

它适用于由大小已知的记录组成的文件，所以我们可以使用 seek() 将记录从一处转移到另一处，然后读取或者修改记录。

文件中记录的大小不一定都相同，只要能够确定哪些记录有多大以及它们在文件中的位置即可。

RandomAccessFile 写操作

当用 rw 方式声明 RandomAccessFile 对象时，如果要写入的文件不存在，系统将自行创建。

r 为只读；w 为只写；rw 为读写。

【示例】文件随机读写

public class RandomAccessFileDemo01 {

    public static void main(String args[]) throws IOException {
        File f = new File("d:" + File.separator + "test.txt"); // 指定要操作的文件
        RandomAccessFile rdf = null; // 声明 RandomAccessFile 类的对象
        rdf = new RandomAccessFile(f, "rw");// 读写模式，如果文件不存在，会自动创建
        String name = null;
        int age = 0;
        name = "zhangsan"; // 字符串长度为 8
        age = 30; // 数字的长度为 4
        rdf.writeBytes(name); // 将姓名写入文件之中
        rdf.writeInt(age); // 将年龄写入文件之中
        name = "lisi    "; // 字符串长度为 8
        age = 31; // 数字的长度为 4
        rdf.writeBytes(name); // 将姓名写入文件之中
        rdf.writeInt(age); // 将年龄写入文件之中
        name = "wangwu  "; // 字符串长度为 8
        age = 32; // 数字的长度为 4
        rdf.writeBytes(name); // 将姓名写入文件之中
        rdf.writeInt(age); // 将年龄写入文件之中
        rdf.close(); // 关闭
    }
}

RandomAccessFile 读操作

读取是直接使用 r 的模式即可，以只读的方式打开文件。

读取时所有的字符串只能按照 byte 数组方式读取出来，而且长度必须和写入时的固定大小相匹配。

public class RandomAccessFileDemo02 {

    public static void main(String args[]) throws IOException {
        File f = new File("d:" + File.separator + "test.txt");    // 指定要操作的文件
        RandomAccessFile rdf = null;        // 声明 RandomAccessFile 类的对象
        rdf = new RandomAccessFile(f, "r");// 以只读的方式打开文件
        String name = null;
        int age = 0;
        byte b[] = new byte[8];    // 开辟 byte 数组
        // 读取第二个人的信息，意味着要空出第一个人的信息
        rdf.skipBytes(12);        // 跳过第一个人的信息
        for (int i = 0; i < b.length; i++) {
            b[i] = rdf.readByte();    // 读取一个字节
        }
        name = new String(b);    // 将读取出来的 byte 数组变为字符串
        age = rdf.readInt();    // 读取数字
        System.out.println("第二个人的信息 --> 姓名：" + name + "；年龄：" + age);
        // 读取第一个人的信息
        rdf.seek(0);    // 指针回到文件的开头
        for (int i = 0; i < b.length; i++) {
            b[i] = rdf.readByte();    // 读取一个字节
        }
        name = new String(b);    // 将读取出来的 byte 数组变为字符串
        age = rdf.readInt();    // 读取数字
        System.out.println("第一个人的信息 --> 姓名：" + name + "；年龄：" + age);
        rdf.skipBytes(12);    // 空出第二个人的信息
        for (int i = 0; i < b.length; i++) {
            b[i] = rdf.readByte();    // 读取一个字节
        }
        name = new String(b);    // 将读取出来的 byte 数组变为字符串
        age = rdf.readInt();    // 读取数字
        System.out.println("第三个人的信息 --> 姓名：" + name + "；年龄：" + age);
        rdf.close();                // 关闭
    }
}

System

System 类中提供了大量的静态方法，可以获取系统相关的信息或系统级操作，其中提供了三个常用于 IO 的静态成员：

System.out - 一个 PrintStream 流。System.out 一般会把你写到其中的数据输出到控制台上。System.out 通常仅用在类似命令行工具的控制台程序上。System.out 也经常用于打印程序的调试信息（尽管它可能并不是获取程序调试信息的最佳方式）。
System.err - 一个 PrintStream 流。System.err 与 System.out 的运行方式类似，但它更多的是用于打印错误文本。一些类似 Eclipse 的程序，为了让错误信息更加显眼，会将错误信息以红色文本的形式通过 System.err 输出到控制台上。
System.in - 一个典型的连接控制台程序和键盘输入的 InputStream 流。通常当数据通过命令行参数或者配置文件传递给命令行 Java 程序的时候，System.in 并不是很常用。图形界面程序通过界面传递参数给程序，这是一块单独的 Java IO 输入机制。

【示例】重定向 System.out 输出流

import java.io.*;
public class SystemOutDemo {

    public static void main(String args[]) throws Exception {
        OutputStream out = new FileOutputStream("d:\\test.txt");
        PrintStream ps = new PrintStream(out);
        System.setOut(ps);
        System.out.println("人生若只如初见，何事秋风悲画扇");
        ps.close();
        out.close();
    }
}

【示例】重定向 System.err 输出流

public class SystemErrDemo {

    public static void main(String args[]) throws IOException {
        OutputStream bos = new ByteArrayOutputStream();        // 实例化
        PrintStream ps = new PrintStream(bos);        // 实例化
        System.setErr(ps);    // 输出重定向
        System.err.print("此处有误");
        System.out.println(bos);    // 输出内存中的数据
    }
}

【示例】System.in 接受控制台输入信息

import java.io.*;
public class SystemInDemo {

    public static void main(String args[]) throws IOException {
        InputStream input = System.in;
        StringBuffer buf = new StringBuffer();
        System.out.print("请输入内容：");
        int temp = 0;
        while ((temp = input.read()) != -1) {
            char c = (char) temp;
            if (c == '\n') {
                break;
            }
            buf.append(c);
        }
        System.out.println("输入的内容为：" + buf);
        input.close();
    }
}

Scanner

Scanner 可以获取用户的输入，并对数据进行校验。

【示例】校验输入数据是否格式正确

import java.io.*;
public class ScannerDemo {

    public static void main(String args[]) {
        Scanner scan = new Scanner(System.in);    // 从键盘接收数据
        int i = 0;
        float f = 0.0f;
        System.out.print("输入整数：");
        if (scan.hasNextInt()) {    // 判断输入的是否是整数
            i = scan.nextInt();    // 接收整数
            System.out.println("整数数据：" + i);
        } else {
            System.out.println("输入的不是整数！");
        }

        System.out.print("输入小数：");
        if (scan.hasNextFloat()) {    // 判断输入的是否是小数
            f = scan.nextFloat();    // 接收小数
            System.out.println("小数数据：" + f);
        } else {
            System.out.println("输入的不是小数！");
        }

        Date date = null;
        String str = null;
        System.out.print("输入日期（yyyy-MM-dd）：");
        if (scan.hasNext("^\\d{4}-\\d{2}-\\d{2}$")) {    // 判断
            str = scan.next("^\\d{4}-\\d{2}-\\d{2}$");    // 接收
            try {
                date = new SimpleDateFormat("yyyy-MM-dd").parse(str);
            } catch (Exception e) {}
        } else {
            System.out.println("输入的日期格式错误！");
        }
        System.out.println(date);
    }
}

输出：

输入整数：20
整数数据：20
输入小数：3.2
小数数据：3.2
输入日期（yyyy-MM-dd）：1988-13-1
输入的日期格式错误！
null

网络编程

关键词：Socket、ServerSocket、DatagramPacket、DatagramSocket

网络编程是指编写运行在多个设备（计算机）的程序，这些设备都通过网络连接起来。

java.net 包中提供了低层次的网络通信细节。你可以直接使用这些类和接口，来专注于解决问题，而不用关注通信细节。

java.net 包中提供了两种常见的网络协议的支持：

TCP - TCP 是传输控制协议的缩写，它保障了两个应用程序之间的可靠通信。通常用于互联网协议，被称 TCP/ IP。

UDP - UDP 是用户数据报协议的缩写，一个无连接的协议。提供了应用程序之间要发送的数据的数据包。

Socket 和 ServerSocket

套接字（Socket）使用 TCP 提供了两台计算机之间的通信机制。客户端程序创建一个套接字，并尝试连接服务器的套接字。

Java 通过 Socket 和 ServerSocket 实现对 TCP 的支持。Java 中的 Socket 通信可以简单理解为：**java.net.Socket 代表客户端，java.net.ServerSocket 代表服务端**，二者可以建立连接，然后通信。

以下为 Socket 通信中建立建立的基本流程：

服务器实例化一个 ServerSocket 对象，表示服务器绑定一个端口。
服务器调用 ServerSocket 的 accept() 方法，该方法将一直等待，直到客户端连接到服务器的绑定端口（即监听端口）。
服务器监听端口时，客户端实例化一个 Socket 对象，指定服务器名称和端口号来请求连接。
Socket 类的构造函数试图将客户端连接到指定的服务器和端口号。如果通信被建立，则在客户端创建一个 Socket 对象能够与服务器进行通信。
在服务器端，accept() 方法返回服务器上一个新的 Socket 引用，该引用连接到客户端的 Socket 。

连接建立后，可以通过使用 IO 流进行通信。每一个 Socket 都有一个输出流和一个输入流。客户端的输出流连接到服务器端的输入流，而客户端的输入流连接到服务器端的输出流。

TCP 是一个双向的通信协议，因此数据可以通过两个数据流在同一时间发送，以下是一些类提供的一套完整的有用的方法来实现 sockets。

ServerSocket

服务器程序通过使用 java.net.ServerSocket 类以获取一个端口，并且监听客户端请求连接此端口的请求。

ServerSocket 构造方法

ServerSocket 有多个构造方法：

方法	描述
`ServerSocket()`	创建非绑定服务器套接字。
`ServerSocket(int port)`	创建绑定到特定端口的服务器套接字。
`ServerSocket(int port, int backlog)`	利用指定的 `backlog` 创建服务器套接字并将其绑定到指定的本地端口号。
`ServerSocket(int port, int backlog, InetAddress address)`	使用指定的端口、监听 `backlog` 和要绑定到的本地 IP 地址创建服务器。

ServerSocket 常用方法

创建非绑定服务器套接字。如果 ServerSocket 构造方法没有抛出异常，就意味着你的应用程序已经成功绑定到指定的端口，并且侦听客户端请求。

这里有一些 ServerSocket 类的常用方法：

方法	描述
`int getLocalPort()`	返回此套接字在其上侦听的端口。
`Socket accept()`	监听并接受到此套接字的连接。
`void setSoTimeout(int timeout)`	通过指定超时值启用/禁用 `SO_TIMEOUT`，以毫秒为单位。
`void bind(SocketAddress host, int backlog)`	将 `ServerSocket` 绑定到特定地址（IP 地址和端口号）。

Socket

java.net.Socket 类代表客户端和服务器都用来互相沟通的套接字。客户端要获取一个 Socket 对象通过实例化，而服务器获得一个 Socket 对象则通过 accept() 方法 a 的返回值。

Socket 构造方法

Socket 类有 5 个构造方法：

方法	描述
`Socket()`	通过系统默认类型的 `SocketImpl` 创建未连接套接字
`Socket(String host, int port)`	创建一个流套接字并将其连接到指定主机上的指定端口号。
`Socket(InetAddress host, int port)`	创建一个流套接字并将其连接到指定 IP 地址的指定端口号。
`Socket(String host, int port, InetAddress localAddress, int localPort)`	创建一个套接字并将其连接到指定远程主机上的指定远程端口。
`Socket(InetAddress host, int port, InetAddress localAddress, int localPort)`	创建一个套接字并将其连接到指定远程地址上的指定远程端口。

当 Socket 构造方法返回，并没有简单的实例化了一个 Socket 对象，它实际上会尝试连接到指定的服务器和端口。

Socket 常用方法

下面列出了一些感兴趣的方法，注意客户端和服务器端都有一个 Socket 对象，所以无论客户端还是服务端都能够调用这些方法。

方法	描述
`void connect(SocketAddress host, int timeout)`	将此套接字连接到服务器，并指定一个超时值。
`InetAddress getInetAddress()`	返回套接字连接的地址。
`int getPort()`	返回此套接字连接到的远程端口。
`int getLocalPort()`	返回此套接字绑定到的本地端口。
`SocketAddress getRemoteSocketAddress()`	返回此套接字连接的端点的地址，如果未连接则返回 null。
`InputStream getInputStream()`	返回此套接字的输入流。
`OutputStream getOutputStream()`	返回此套接字的输出流。
`void close()`	关闭此套接字。

Socket 通信示例

服务端示例：

public class HelloServer {

    public static void main(String[] args) throws Exception {
        // Socket 服务端
        // 服务器在 8888 端口上监听
        ServerSocket server = new ServerSocket(8888);
        System.out.println("服务器运行中，等待客户端连接。");
        // 得到连接，程序进入到阻塞状态
        Socket client = server.accept();
        // 打印流输出最方便
        PrintStream out = new PrintStream(client.getOutputStream());
        // 向客户端输出信息
        out.println("hello world");
        client.close();
        server.close();
        System.out.println("服务器已向客户端发送消息，退出。");
    }

}

客户端示例：

public class HelloClient {

    public static void main(String[] args) throws Exception {
        // Socket 客户端
        Socket client = new Socket("localhost", 8888);
        InputStreamReader inputStreamReader = new InputStreamReader(client.getInputStream());
        // 一次性接收完成
        BufferedReader buf = new BufferedReader(inputStreamReader);
        String str = buf.readLine();
        buf.close();
        client.close();
        System.out.println("客户端接收到服务器消息：" + str + "，退出");
    }

}

DatagramSocket 和 DatagramPacket

Java 通过 DatagramSocket 和 DatagramPacket 实现对 UDP 协议的支持。

DatagramPacket：数据包类
DatagramSocket：通信类

UDP 服务端示例：

public class UDPServer {

    public static void main(String[] args) throws Exception { // 所有异常抛出
        String str = "hello World!!!";
        DatagramSocket ds = new DatagramSocket(3000); // 服务端在 3000 端口上等待服务器发送信息
        DatagramPacket dp =
            new DatagramPacket(str.getBytes(), str.length(), InetAddress.getByName("localhost"), 9000); // 所有的信息使用 buf 保存
        System.out.println("发送信息。");
        ds.send(dp); // 发送信息出去
        ds.close();
    }

}

UDP 客户端示例：

public class UDPClient {

    public static void main(String[] args) throws Exception { // 所有异常抛出
        byte[] buf = new byte[1024]; // 开辟空间，以接收数据
        DatagramSocket ds = new DatagramSocket(9000); // 客户端在 9000 端口上等待服务器发送信息
        DatagramPacket dp = new DatagramPacket(buf, 1024); // 所有的信息使用 buf 保存
        ds.receive(dp); // 接收数据
        String str = new String(dp.getData(), 0, dp.getLength()) + "from " + dp.getAddress().getHostAddress() + "："
            + dp.getPort();
        System.out.println(str); // 输出内容
    }

}

InetAddress

InetAddress 类表示互联网协议 (IP) 地址。

没有公有的构造函数，只能通过静态方法来创建实例。

1 2	InetAddress.getByName(String host); InetAddress.getByAddress(byte[] address);

URL

可以直接从 URL 中读取字节流数据。

public static void main(String[] args) throws IOException {

    URL url = new URL("http://www.baidu.com");

    /* 字节流 */
    InputStream is = url.openStream();

    /* 字符流 */
    InputStreamReader isr = new InputStreamReader(is, "utf-8");

    /* 提供缓存功能 */
    BufferedReader br = new BufferedReader(isr);

    String line;
    while ((line = br.readLine()) != null) {
        System.out.println(line);
    }

    br.close();
}

参考资料

Java 虚拟机之内存区域

发表于 2020-06-28 更新于 2025-09-13 分类于 Java ， JavaCore ， JVM 本文字数： 21k 阅读时长 ≈ 19 分钟

Java 虚拟机之内存区域

运行时数据区域

JVM 在执行 Java 程序的过程中会把它所管理的内存划分为若干个不同的数据区域。这些区域都有各自的用途，以及创建和销毁的时间，有的区域随着虚拟机进程的启动而存在，有些区域则依赖用户线程的启动和结束而建立和销毁。如下图所示：

程序计数器

程序计数器（Program Counter Register） 是一块较小的内存空间，它可以看做是当前线程所执行的字节码的行号指示器。字节码解释器工作时就是通过改变这个计数器的值来选取下一条需要执行的字节码指令，它是程序控制流的指示器，分支、循环、跳转、异常处理、线程恢复等基础功能都需要依赖这个计数器来完成。

由于 Java 虚拟机的多线程是通过线程轮流切换、分配处理器执行时间的方式来实现的，在任何一个确定的时刻，一个处理器都只会执行一条线程中的指令。因此，为了线程切换后能恢复到正确的执行位置，每条线程都需要有一个独立的程序计数器，各线程之间计数器互不影响，独立存储，我们称这类内存区域为“线程私有”的内存。

如果线程正在执行的是一个 Java 方法，这个计数器记录的是正在执行的虚拟机字节码指令的地址；如果正在执行的是本地（Native）方法，这个计数器值则应为空（Undefined）。

🔔 注意：程序计数器是 JVM 中没有规定任何 OutOfMemoryError 情况的唯一区域。

Java 虚拟机栈

Java 虚拟机栈（Java Virtual Machine Stacks） 也是线程私有的，它的生命周期与线程相同。Java 虚拟机栈以方法作为最基本的执行单元，描述的是 Java 方法执行的线程内存模型。每个方法被执行的时候，JVM 都会同步创建一个栈帧（Stack Frame），栈帧是用于支持虚拟机进行方法调用和方法执行背后的数据结构。栈帧存储了局部变量表、操作数栈、动态连接、方法返回地址等信息。每一个方法从调用开始至执行结束的过程，都对应着一个栈帧在虚拟机栈里面从入栈到出栈的过程。

一个线程中的方法调用链可能会很长，以 Java 程序的角度来看，同一时刻、同一条线程里面，在调用堆栈的所有方法都同时处于执行状态。而对于执行引擎来讲，在活动线程中，只有位于栈顶的方法才是在运行的，只有位于栈顶的栈帧才是生效的，其被称为“当前栈帧”（Current Stack Frame），与这个栈帧所关联的方法被称为“当前方法”（Current Method）。执行引擎所运行的所有字节码指令都只针对当前栈帧进行操作。

局部变量表 - 用于存放方法参数和方法内部定义的局部变量。
操作数栈 - 主要作为方法调用的中转站使用，用于存放方法执行过程中产生的中间计算结果。另外，计算过程中产生的临时变量也会放在操作数栈中。
动态连接 - 用于一个方法调用其他方法的场景。Class 文件的常量池中有大量的符号引用，字节码中的方法调用指令就以常量池中指向方法的符号引用为参数。这些符号引用一部分会在类加载阶段或第一次使用的时候转化为直接引用，这种转化称为静态解析；另一部分将在每一次的运行期间转化为直接应用，这部分称为动态连接。
方法返回地址 - 用于返回方法被调用的位置，恢复上层方法的局部变量和操作数栈。Java 方法有两种返回方式，一种是 return 语句正常返回，一种是抛出异常。无论采用何种退出方式，都会导致栈帧被弹出。也就是说，栈帧随着方法调用而创建，随着方法结束而销毁。无论方法正常完成还是异常完成都算作方法结束。

🔔 注意：

该区域可能抛出以下异常：

如果线程请求的栈深度大于虚拟机所允许的深度，将抛出 StackOverflowError 异常；

如果虚拟机栈进行动态扩展时，无法申请到足够内存，就会抛出 OutOfMemoryError 异常。

💡 提示：

可以通过 -Xss 这个虚拟机参数来指定一个程序的 Java 虚拟机栈内存大小：
1
java -Xss=512M HackTheJava

局部变量表

局部变量表（Local Variables Table）是一组变量值的存储空间，用于存放方法参数和方法内部定义的局部变量。在 Java 程序被编译为 Class 文件时，就在方法的 Code 属性的 max_locals 数据项中确定了该方法所需分配的局部变量表的最大容量。

局部变量表存放了编译期可知的各种 Java 虚拟机基本数据类型、对象引用（reference 类型，它不同于对象本身，可能是一个指向对象起始地址的引用指针，也可能是指向一个代表对象的句柄或其他与此对象相关的位置）和 returnAddress 类型（指向了一条字节码指令的地址）。这些数据类型在局部变量表中的存储空间以局部变量槽（Variable Slot）来表示，其中 64 位长度的 long 和 double 类型的数据会占用两个变量槽，其余的数据类型只占用一个。局部变量表所需的内存空间在编译期间完成分配，当进入一个方法时，这个方法需要在栈帧中分配多大的局部变量空间是完全确定的，在方法运行期间不会改变局部变量表的大小。

操作数栈

操作数栈（Operand Stack）也常被称为操作栈，它是一个后入先出（Last In First Out，LIFO）栈。操作数栈主要作为方法调用的中转站使用，用于存放方法执行过程中产生的中间计算结果。另外，计算过程中产生的临时变量也会放在操作数栈中。

当一个方法刚刚开始执行的时候，这个方法的操作数栈是空的，在方法的执行过程中，会有各种字节码指令往操作数栈中写入和提取内容，也就是出栈和入栈操作。譬如在做算术运算的时候是通过将运算涉及的操作数栈压入栈顶后调用运算指令来进行的，又譬如在调用其他方法的时候是通过操作数栈来进行方法参数的传递。

操作数栈中元素的数据类型必须与字节码指令的序列严格匹配，在编译程序代码的时候，编译器必须要严格保证这一点，在类校验阶段的数据流分析中还要再次验证这一点。

另外在概念模型中，两个不同栈帧作为不同方法的虚拟机栈的元素，是完全相互独立的。但是在大多虚拟机的实现里都会进行一些优化处理，令两个栈帧出现一部分重叠。让下面栈帧的部分操作数栈与上面栈帧的部分局部变量表重叠在一起，这样做不仅节约了一些空间，更重要的是在进行方法调用时就可以直接共用一部分数据，无须进行额外的参数复制传递了。

动态连接

用于一个方法调用其他方法的场景。Class 文件的常量池中有大量的符号引用，字节码中的方法调用指令就以常量池中指向方法的符号引用为参数。这些符号引用一部分会在类加载阶段或第一次使用的时候转化为直接引用，这种转化称为静态解析；另一部分将在每一次的运行期间转化为直接应用，这部分称为动态连接。

每个栈帧都包含一个指向运行时常量池中该栈帧所属方法的引用，持有这个引用是为了支持方法调用过程中的动态连接（Dynamic Linking）。通过第 6 章的讲解，我们知道 Class 文件的常量池中存有大量的符号引用，字节码中的方法调用指令就以常量池里指向方法的符号引用作为参数。这些符号引用一部分会在类加载阶段或者第一次使用的时候就被转化为直接引用，这种转化被称为静态解析。另外一部分将在每一次运行期间都转化为直接引用，这部分就称为动态连接。关于这两个转化过程的具体过程，将在 8.3 节中再详细讲解。

方法返回地址

方法返回地址用于返回方法被调用的位置，恢复上层方法的局部变量和操作数栈。

Java 方法有两种返回方式，一种是 return 语句正常返回，这时候可能会有返回值传递给上层的方法调用者，方法是否有返回值以及返回值的类型将根据遇到何种方法返回指令来决定；一种是遇到了异常，并且这个异常没有在方法体内得到妥善处理。无论采用何种退出方式，都会导致栈帧被弹出。也就是说，栈帧随着方法调用而创建，随着方法结束而销毁。无论方法正常完成还是异常完成都算作方法结束。

方法返回时可能需要在栈帧中保存一些信息，用来帮助恢复它的上层主调方法的执行状态。一般来说，方法正常退出时，主调方法的 PC 计数器的值就可以作为返回地址，栈帧中很可能会保存这个计数器值。而方法异常退出时，返回地址是要通过异常处理器表来确定的，栈帧中就一般不会保存这部分信息。方法退出的过程实际上等同于把当前栈帧出栈，因此退出时可能执行的操作有：恢复上层方法的局部变量表和操作数栈，把返回值（如果有的话）压入调用者栈帧的操作数栈中，调整 PC 计数器的值以指向方法调用指令后面的一条指令等。

本地方法栈

本地方法栈（Native Method Stack） 与虚拟机栈的作用非常相似，二者区别仅在于：虚拟机栈为 Java 方法服务；本地方法栈为 Native 方法服务。

🔔 注意：本地方法栈也会在栈深度溢出或者栈扩展失败时分别抛出 StackOverflowError 和 OutOfMemoryError 异常。

Java 堆

Java 堆（Java Heap）的作用就是存放对象实例，几乎所有的对象实例都是在这里分配内存。

注：由于即时编译技术的进步，尤其是逃逸分析技术的日渐强大，栈上分配、标量替换优化手段已经导致一些微妙的变化悄然发生，所以说 Java 对象实例都分配在堆上也渐渐变得不是那么绝对了。

Java 堆是垃圾收集器管理的内存区域（因此也被叫做”GC 堆”）。现代的垃圾收集器大部分都是采用分代收集理论设计的，该力量的思想是针对不同的对象采取不同的垃圾回收算法。

在 JDK 7 及之前版本，堆内存被通常分为下面三部分：

新生代（Young Generation）
老年代（Old Generation）
永久代（Permanent Generation）

JDK 8 版本之后 PermGen（永久代）已被 Metaspace（元空间）取代，元空间使用的是本地内存。

大部分情况，对象都会首先在 Eden 区域分配，在一次新生代垃圾回收后，如果对象还存活，则会进入 S0 或者 S1，并且对象的年龄还会加 1(Eden 区->Survivor 区后对象的初始年龄变为 1)，当它的年龄增加到一定程度（默认为 15 岁），就会被晋升到老年代中。对象晋升到老年代的年龄阈值，可以通过参数 -XX:MaxTenuringThreshold 来设置。不过，设置的值应该在 0-15，否则会爆出以下错误：

1	MaxTenuringThreshold of 20 is invalid; must be between 0 and 15

为什么年龄只能是 0-15?

因为记录年龄的区域在对象头中，这个区域的大小通常是 4 位。这 4 位可以表示的最大二进制数字是 1111，即十进制的 15。因此，对象的年龄被限制为 0 到 15。

这里我们简单结合对象布局来详细介绍一下。

在 HotSpot 虚拟机中，对象在内存中存储的布局可以分为 3 块区域：对象头（Header）、实例数据（Instance Data）和对齐填充（Padding）。其中，对象头包括两部分：标记字段（Mark Word）和类型指针（Klass Word）。关于对象内存布局的详细介绍，后文会介绍到，这里就不重复提了。

这个年龄信息就是在标记字段中存放的（标记字段还存放了对象自身的其他信息比如哈希码、锁状态信息等等）。

如果从分配内存的角度看，所有线程共享的 Java 堆中可以划分出多个线程私有的分配缓冲区（Thread Local Allocation Buffer，TLAB），以提升对象分配效率。不过无论从什么角度，无论如何划分，都不会改变 Java 堆中存储内容的共性，无论是哪个区域，存储的都只能是对象的实例，将 Java 堆细分的目的只是为了更好地回收内存，或者更快地分配内存。

🔔 注意：Java 堆既可以被实现成固定大小的，也可以是可扩展的，不过当前主流的 Java 虚拟机都是按照可扩展来实现的，扩展失败会抛出 OutOfMemoryError 异常。

可以通过 -Xms 和 -Xmx 两个虚拟机参数来指定一个程序的 Java 堆内存大小，第一个参数设置初始值，第二个参数设置最大值。
1
java -Xms=1M -Xmx=2M HackTheJava

方法区

方法区（Method Area）是各个线程共享的内存区域。方法区用于存放已被加载的类信息、常量、静态变量、即时编译器编译后的代码等数据。

在 JDK8 以前，方法区常被称为永久代，但这种说法是不准确的：仅仅是因为当时的 HotSpot 虚拟机使用永久代来实现方法区而已。对于其他虚拟机而言，是不存在永久代概念的。永久代这种设计，导致了 Java 应用更容易遇到内存溢出的问题（永久代有 -XX:MaxPermSize 的上限，即使不设置也有默认大小）。

JDK7 之前，HotSpot 虚拟机把它当成永久代来进行垃圾回收，可通过参数 -XX:PermSize 和 -XX:MaxPermSize 设置。
JDK8 之后，取消了永久代，用 **metaspace（元空间）**替代，可通过参数 -XX:MaxMetaspaceSize 设置。

方法区的内存回收目标主要是针对常量池的回收和对类型的卸载，一般来说这个区域的回收效果比较难令人满意，尤其是类型的卸载，条件相当苛刻。

🔔 注意：方法区和 Java 堆一样不需要连续的内存，并且可以动态扩展，动态扩展失败一样会抛出 OutOfMemoryError 异常。

运行时常量池

运行时常量池（Runtime Constant Pool） 是方法区的一部分，Class 文件中除了有类的版本、字段、方法、接口等描述信息，还有一项信息是常量池表（Constant Pool Table），用于存放编译器生成的各种字面量和符号引用，这部分内容将在类加载后写入。

字面量 - 文本字符串、声明为 final 的常量值等。
符号引用 - 类和接口的完全限定名（Fully Qualified Name）、字段的名称和描述符（Descriptor）、方法的名称和描述符。

运行时常量池相对于 Class 文件常量池的另外一个重要特征是具备动态性，Java 语言并不要求常量一定只有编译期才能产生，也就是说，并非预置入 Class 文件中常量池的内容才能进入方法区运行时常量池，运行期间也可以将新的常量放入池中，这种特性被开发人员利用得比较多的便是 String 类的 intern() 方法。

🔔 注意：当常量池无法再申请到内存时会抛出 OutOfMemoryError 异常。

直接内存

直接内存（Direct Memory）并不是虚拟机运行时数据区的一部分，也不是 JVM 规范中定义的内存区域。

JDK4 中新加入了 NIO，它可以使用 Native 函数库直接分配堆外内存，然后通过一个存储在 Java 堆里的 DirectByteBuffer 对象作为这块内存的引用进行操作。这样能在一些场景中显著提高性能，因为避免了在 Java 堆和 Native 堆中来回复制数据。

直接内存容量可通过 -XX:MaxDirectMemorySize 指定，如果不指定，则默认与 Java 堆最大值（-Xmx 指定）一样。

🔔 注意：直接内存这部分也被频繁的使用，且也可能导致 OutOfMemoryError 异常。

Java 内存区域对比

内存区域	内存作用范围	异常
程序计数器	线程私有	无
Java 虚拟机栈	线程私有	`StackOverflowError` 和 `OutOfMemoryError`
本地方法栈	线程私有	`StackOverflowError` 和 `OutOfMemoryError`
Java 堆	线程共享	`OutOfMemoryError`
方法区	线程共享	`OutOfMemoryError`
运行时常量池	线程共享	`OutOfMemoryError`
直接内存	非运行时数据区	`OutOfMemoryError`

虚拟机对象

对象的创建

当 Java 虚拟机遇到一条字节码 new 指令时，首先在常量池中尝试定位类的符号引用，并检查这个类是否已被类加载，如果没有，则必须先执行相应的类加载过程。

在类加载检查通过后，接下来虚拟机将为新生对象分配内存。对象所需内存的大小在类加载完成后便可完全确定，为对象分配空间的任务实际上便等同于把一块确定大小的内存块从 Java 堆中划分出来。分配对象内存有两种方式：

指针碰撞（Bump The Pointer） - 如果 Java 堆中内存是规整的，所有被使用过的内存都被放在一边，空闲的内存被放在另一边，中间放着一个指针作为分界点的指示器，那所分配内存就仅仅是把那个指针向空闲空间方向挪动一段与对象大小相等的距离。

空闲列表（Free List） - 如果 Java 堆中的内存是不规整的，已被使用的内存和空闲的内存相互交错在一起，那就没有办法简单地进行指针碰撞了，虚拟机就必须维护一个列表，记录上哪些内存块是可用的，在分配的时候从列表中找到一块足够大的空间划分给对象实例，并更新列表上的记录。

选择哪种分配方式由 Java 堆是否规整决定，而 Java 堆是否规整又由所采用的垃圾收集器是否采用标记-压缩算法决定。因此，当使用 Serial、ParNew 等带压缩整理过程的收集器时，系统采用的分配算法是指针碰撞，既简单又高效；而当使用 CMS 这种基于清除（Sweep）算法的收集器时，理论上就只能采用较为复杂的空闲列表来分配内存。

对象创建在虚拟机中是非常频繁的行为，因此还需要考虑分配内存空间的并发安全问题。一般有两种方案：

CAS 同步 - 对分配内存空间的动作进行同步处理——实际上虚拟机是采用 CAS 配上失败重试的方式保证更新操作的原子性；
TLAB - 另外一种是把内存分配的动作按照线程划分在不同的空间之中进行，即每个线程在 Java 堆中预先分配一小块内存，称为本地线程分配缓冲（Thread Local Allocation Buffer，TLAB），哪个线程要分配内存，就在哪个线程的本地缓冲区中分配，只有本地缓冲区用完了，分配新的缓存区时才需要同步锁定。

接下来，需要执行类的构造函数（即 <init>() 方法）对对象进行初始化。

对象的内存布局

在 HotSpot 虚拟机里，对象在堆内存中的存储布局可以划分为三个部分：

对象头（Header） - HotSpot 虚拟机对象的对象头部分包括两类信息。
- Mark Word - 用于存储对象自身的运行时数据。如哈希码（HashCode）、GC 分代年龄、锁状态标志、线程持有的锁、偏向线程 ID、偏向时间戳等，这部分数据的长度在 32 位和 64 位的虚拟机（未开启压缩指针）中分别为 32 个比特和 64 个比特。
- 类型指针 - 对象指向它的类型元数据的指针，Java 虚拟机通过这个指针来确定该对象是哪个类的实例。并不是所有的虚拟机实现都必须在对象数据上保留类型指针，换句话说，查找对象的元数据信息并不一定要经过对象本身。此外，如果对象是一个 Java 数组，那在对象头中还必须有一块用于记录数组长度的数据，因为虚拟机可以通过普通 Java 对象的元数据信息确定 Java 对象的大小，但是如果数组的长度是不确定的，将无法通过元数据中的信息推断出数组的大小。
实例数据（Instance Data） - 对象真正存储的有效信息，即我们在程序代码里面所定义的各种类型的字段内容，无论是从父类继承下来的，还是在子类中定义的字段都必须记录起来。
对齐填充（Padding） - 并不是必然存在的，也没有特别的含义，它仅仅起着占位符的作用。

对象的访问定位

Java 程序会通过栈上的 reference 数据来操作堆上的具体对象。主流的对象访问方式主要有使用句柄和直接指针两种：使用句柄访问和使用直接指针访问。

使用句柄访问

Java 堆中将可能会划分出一块内存来作为句柄池，reference 中存储的就是对象的句柄地址，而句柄中包含了对象实例数据与类型数据各自具体的地址信息。使用句柄来访问的最大好处就是 reference 中存储的是稳定句柄地址，在对象被移动（垃圾收集时移动对象是非常普遍的行为）时只会改变句柄中的实例数据指针，而 reference 本身不需要被修改。

使用直接指针访问

Java 堆中对象的内存布局就必须考虑如何放置访问类型数据的相关信息，reference 中存储的直接就是对象地址，如果只是访问对象本身的话，就不需要多一次间接访问的开销。使用直接指针来访问最大的好处就是速度更快，它节省了一次指针定位的时间开销。HotSpot 主要使用第二种方式进行对象访问。

内存分配

public class JVMCase {

	// 常量
	public final static String MAN_SEX_TYPE = "man";

	// 静态变量
	public static String WOMAN_SEX_TYPE = "woman";

	public static void main(String[] args) {

		Student stu = new Student();
		stu.setName("nick");
		stu.setSexType(MAN_SEX_TYPE);
		stu.setAge(20);

		JVMCase jvmcase = new JVMCase();

		// 调用静态方法
		print(stu);
		// 调用非静态方法
		jvmcase.sayHello(stu);
	}

	// 常规静态方法
	public static void print(Student stu) {
		System.out.println("name: " + stu.getName() + "; sex:" + stu.getSexType() + "; age:" + stu.getAge());
	}

	// 非静态方法
	public void sayHello(Student stu) {
		System.out.println(stu.getName() + "say: hello");
	}
}

class Student{
	String name;
	String sexType;
	int age;

	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}

	public String getSexType() {
		return sexType;
	}
	public void setSexType(String sexType) {
		this.sexType = sexType;
	}
	public int getAge() {
		return age;
	}
	public void setAge(int age) {
		this.age = age;
	}
}

运行以上代码时，JVM 处理过程如下：

（1）JVM 向操作系统申请内存，JVM 第一步就是通过配置参数或者默认配置参数向操作系统申请内存空间，根据内存大小找到具体的内存分配表，然后把内存段的起始地址和终止地址分配给 JVM，接下来 JVM 就进行内部分配。

（2）JVM 获得内存空间后，会根据配置参数分配堆、栈以及方法区的内存大小。

（3）class 文件加载、验证、准备以及解析，其中准备阶段会为类的静态变量分配内存，初始化为系统的初始值（这部分我在第 21 讲还会详细介绍）。

（4）完成上一个步骤后，将会进行最后一个初始化阶段。在这个阶段中，JVM 首先会执行构造器 <clinit> 方法，编译器会在 .java 文件被编译成 .class 文件时，收集所有类的初始化代码，包括静态变量赋值语句、静态代码块、静态方法，收集在一起成为 <clinit>() 方法。

（5）执行方法。启动 main 线程，执行 main 方法，开始执行第一行代码。此时堆内存中会创建一个 student 对象，对象引用 student 就存放在栈中。

（6）此时再次创建一个 JVMCase 对象，调用 sayHello 非静态方法，sayHello 方法属于对象 JVMCase，此时 sayHello 方法入栈，并通过栈中的 student 引用调用堆中的 Student 对象；之后，调用静态方法 print，print 静态方法属于 JVMCase 类，是从静态方法中获取，之后放入到栈中，也是通过 student 引用调用堆中的 student 对象。

内存溢出

OutOfMemoryError

OutOfMemoryError 简称为 OOM。Java 中对 OOM 的解释是，没有空闲内存，并且垃圾收集器也无法提供更多内存。通俗的解释是：JVM 内存不足了。

在 JVM 规范中，除了程序计数器区域外，其他运行时区域都可能发生 OutOfMemoryError 异常（简称 OOM）。

下面逐一介绍 OOM 发生场景。

堆空间溢出

java.lang.OutOfMemoryError: Java heap space 意味着：堆空间溢出。

更细致的说法是：Java 堆内存已经达到 -Xmx 设置的最大值。Java 堆用于存储对象实例，只要不断地创建对象，并且保证 GC Roots 到对象之间有可达路径来避免垃圾收集器回收这些对象，那么当堆空间到达最大容量限制后就会产生 OOM。

堆空间溢出有可能是内存泄漏（Memory Leak） 或 内存溢出（Memory Overflow） 。

Java heap space 分析步骤

使用 jmap 或 -XX:+HeapDumpOnOutOfMemoryError 获取堆快照。
使用内存分析工具（VisualVM、MAT、JProfile 等）对堆快照文件进行分析。
根据分析图，重点是确认内存中的对象是否是必要的，分清究竟是是内存泄漏还是内存溢出。

内存泄漏

内存泄漏（Memory Leak）是指由于疏忽或错误造成程序未能释放已经不再使用的内存的情况。

内存泄漏并非指内存在物理上的消失，而是应用程序分配某段内存后，由于设计错误，失去了对该段内存的控制，因而造成了内存的浪费。内存泄漏随着被执行的次数不断增加，最终会导致内存溢出。

内存泄漏常见场景：

静态容器
- 声明为静态（static）的 HashMap、Vector 等集合
- 通俗来讲 A 中有 B，当前只把 B 设置为空，A 没有设置为空，回收时 B 无法回收。因为被 A 引用。
监听器
- 监听器被注册后释放对象时没有删除监听器
物理连接
- 各种连接池建立了连接，必须通过 close() 关闭链接
内部类和外部模块等的引用
- 发现它的方式同内存溢出，可再加个实时观察
- jstat -gcutil 7362 2500 70

重点关注：

FGC — 从应用程序启动到采样时发生 Full GC 的次数。
FGCT — 从应用程序启动到采样时 Full GC 所用的时间（单位秒）。
FGC 次数越多，FGCT 所需时间越多，越有可能存在内存泄漏。

如果是内存泄漏，可以进一步查看泄漏对象到 GC Roots 的对象引用链。这样就能找到泄漏对象是怎样与 GC Roots 关联并导致 GC 无法回收它们的。掌握了这些原因，就可以较准确的定位出引起内存泄漏的代码。

导致内存泄漏的常见原因是使用容器，且不断向容器中添加元素，但没有清理，导致容器内存不断膨胀。

【示例】

/**
 * 内存泄漏示例
 * 错误现象：java.lang.OutOfMemoryError: Java heap space
 * VM Args：-verbose:gc -Xms10M -Xmx10M -XX:+HeapDumpOnOutOfMemoryError
 */
public class HeapMemoryLeakOOM {

    public static void main(String[] args) {
        List<OomObject> list = new ArrayList<>();
        while (true) {
            list.add(new OomObject());
        }
    }

    static class OomObject {}

}

内存溢出

如果不存在内存泄漏，即内存中的对象确实都必须存活着，则应当检查虚拟机的堆参数（-Xmx 和 -Xms），与机器物理内存进行对比，看看是否可以调大。并从代码上检查是否存在某些对象生命周期过长、持有时间过长的情况，尝试减少程序运行期的内存消耗。

【示例】

/**
 * 堆溢出示例
 * <p>
 * 错误现象：java.lang.OutOfMemoryError: Java heap space
 * <p>
 * VM Args：-verbose:gc -Xms10M -Xmx10M
 */
public class HeapOutOfMemoryOOM {

    public static void main(String[] args) {
        Double[] array = new Double[999999999];
        System.out.println("array length = [" + array.length + "]");
    }

}

上面的例子是一个极端的例子，试图创建一个维度很大的数组，堆内存无法分配这么大的内存，从而报错：Java heap space。

但如果在现实中，代码并没有问题，仅仅是因为堆内存不足，可以通过 -Xms 和 -Xmx 适当调整堆内存大小。

GC 开销超过限制

java.lang.OutOfMemoryError: GC overhead limit exceeded 这个错误，官方给出的定义是：超过 98% 的时间用来做 GC 并且回收了不到 2% 的堆内存时会抛出此异常。这意味着，发生在 GC 占用大量时间为释放很小空间的时候发生的，是一种保护机制。导致异常的原因：一般是因为堆太小，没有足够的内存。

【示例】

/**
 * GC overhead limit exceeded 示例
 * 错误现象：java.lang.OutOfMemoryError: GC overhead limit exceeded
 * 发生在 GC 占用大量时间为释放很小空间的时候发生的，是一种保护机制。导致异常的原因：一般是因为堆太小，没有足够的内存。
 * 官方对此的定义：超过 98%的时间用来做 GC 并且回收了不到 2%的堆内存时会抛出此异常。
 * VM Args: -Xms10M -Xmx10M
 */
public class GcOverheadLimitExceededOOM {

    public static void main(String[] args) {
        List<Double> list = new ArrayList<>();
        double d = 0.0;
        while (true) {
            list.add(d++);
        }
    }

}

【处理】

与 Java heap space 错误处理方法类似，先判断是否存在内存泄漏。如果有，则修正代码；如果没有，则通过 -Xms 和 -Xmx 适当调整堆内存大小。

永久代空间不足

【错误】

1	java.lang.OutOfMemoryError: PermGen space

【原因】

Perm （永久代）空间主要用于存放 Class 和 Meta 信息，包括类的名称和字段，带有方法字节码的方法，常量池信息，与类关联的对象数组和类型数组以及即时编译器优化。GC 在主程序运行期间不会对永久代空间进行清理，默认是 64M 大小。

根据上面的定义，可以得出 PermGen 大小要求取决于加载的类的数量以及此类声明的大小。因此，可以说造成该错误的主要原因是永久代中装入了太多的类或太大的类。

在 JDK8 之前的版本中，可以通过 -XX:PermSize 和 -XX:MaxPermSize 设置永久代空间大小，从而限制方法区大小，并间接限制其中常量池的容量。

初始化时永久代空间不足

【示例】

/**
 * 永久代内存空间不足示例
 * <p>
 * 错误现象：
 * <ul>
 * <li>java.lang.OutOfMemoryError: PermGen space (JDK8 以前版本）</li>
 * <li>java.lang.OutOfMemoryError: Metaspace (JDK8 及以后版本）</li>
 * </ul>
 * VM Args:
 * <ul>
 * <li>-Xmx100M -XX:MaxPermSize=16M (JDK8 以前版本）</li>
 * <li>-Xmx100M -XX:MaxMetaspaceSize=16M (JDK8 及以后版本）</li>
 * </ul>
 */
public class PermGenSpaceOOM {

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 100_000_000; i++) {
            generate("eu.plumbr.demo.Generated" + i);
        }
    }

    public static Class generate(String name) throws Exception {
        ClassPool pool = ClassPool.getDefault();
        return pool.makeClass(name).toClass();
    }

}

在此示例中，源代码遍历循环并在运行时生成类。javassist 库正在处理类生成的复杂性。

重部署时永久代空间不足

对于更复杂，更实际的示例，让我们逐步介绍一下在应用程序重新部署期间发生的 PermGen 空间错误。重新部署应用程序时，你希望垃圾回收会摆脱引用所有先前加载的类的加载器，并被加载新类的类加载器取代。

不幸的是，许多第三方库以及对线程，JDBC 驱动程序或文件系统句柄等资源的不良处理使得无法卸载以前使用的类加载器。反过来，这意味着在每次重新部署期间，所有先前版本的类仍将驻留在 PermGen 中，从而在每次重新部署期间生成数十兆的垃圾。

让我们想象一个使用 JDBC 驱动程序连接到关系数据库的示例应用程序。启动应用程序时，初始化代码将加载 JDBC 驱动程序以连接到数据库。对应于规范，JDBC 驱动程序向 java.sql.DriverManager 进行注册。该注册包括将对驱动程序实例的引用存储在 DriverManager 的静态字段中。

现在，当从应用程序服务器取消部署应用程序时，java.sql.DriverManager 仍将保留该引用。我们最终获得了对驱动程序类的实时引用，而驱动程序类又保留了用于加载应用程序的 java.lang.Classloader 实例的引用。反过来，这意味着垃圾回收算法无法回收空间。

而且该 java.lang.ClassLoader 实例仍引用应用程序的所有类，通常在 PermGen 中占据数十兆字节。这意味着只需少量重新部署即可填充通常大小的 PermGen。

PermGen space 解决方案

（1）解决初始化时的 OutOfMemoryError

在应用程序启动期间触发由于 PermGen 耗尽导致的 OutOfMemoryError 时，解决方案很简单。该应用程序仅需要更多空间才能将所有类加载到 PermGen 区域，因此我们只需要增加其大小即可。为此，更改你的应用程序启动配置并添加（或增加，如果存在）-XX:MaxPermSize 参数，类似于以下示例：

1	java -XX:MaxPermSize=512m com.yourcompany.YourClass

上面的配置将告诉 JVM，PermGen 可以增长到 512MB。

清理应用程序中 WEB-INF/lib 下的 jar，用不上的 jar 删除掉，多个应用公共的 jar 移动到 Tomcat 的 lib 目录，减少重复加载。

🔔 注意：-XX:PermSize 一般设为 64M

（2）解决重新部署时的 OutOfMemoryError

重新部署应用程序后立即发生 OutOfMemoryError 时，应用程序会遭受类加载器泄漏的困扰。在这种情况下，解决问题的最简单，继续进行堆转储分析–使用类似于以下命令的重新部署后进行堆转储：

1	jmap -dump:format=b,file=dump.hprof <process-id>

然后使用你最喜欢的堆转储分析器打开转储（Eclipse MAT 是一个很好的工具）。在分析器中可以查找重复的类，尤其是那些正在加载应用程序类的类。从那里，你需要进行所有类加载器的查找，以找到当前活动的类加载器。

对于非活动类加载器，你需要通过从非活动类加载器收集到 GC 根的最短路径来确定阻止它们被垃圾收集的引用。有了此信息，你将找到根本原因。如果根本原因是在第三方库中，则可以进入 Google/StackOverflow 查看是否是已知问题以获取补丁/解决方法。

（3）解决运行时 OutOfMemoryError

第一步是检查是否允许 GC 从 PermGen 卸载类。在这方面，标准的 JVM 相当保守-类是天生的。因此，一旦加载，即使没有代码在使用它们，类也会保留在内存中。当应用程序动态创建许多类并且长时间不需要生成的类时，这可能会成为问题。在这种情况下，允许 JVM 卸载类定义可能会有所帮助。这可以通过在启动脚本中仅添加一个配置参数来实现：

1	-XX:+CMSClassUnloadingEnabled

默认情况下，此选项设置为 false，因此要启用此功能，你需要在 Java 选项中显式设置。如果启用 CMSClassUnloadingEnabled，GC 也会扫描 PermGen 并删除不再使用的类。请记住，只有同时使用 UseConcMarkSweepGC 时此选项才起作用。

1	-XX:+UseConcMarkSweepGC

在确保可以卸载类并且问题仍然存在之后，你应该继续进行堆转储分析–使用类似于以下命令的方法进行堆转储：

1	jmap -dump:file=dump.hprof,format=b <process-id>

然后，使用你最喜欢的堆转储分析器（例如 Eclipse MAT）打开转储，然后根据已加载的类数查找最昂贵的类加载器。从此类加载器中，你可以继续提取已加载的类，并按实例对此类进行排序，以使可疑对象排在首位。

然后，对于每个可疑者，就需要你手动将根本原因追溯到生成此类的应用程序代码。

元数据区空间不足

【错误】

1	Exception in thread "main" java.lang.OutOfMemoryError: Metaspace

【原因】

Java8 以后，JVM 内存空间发生了很大的变化。取消了永久代，转而变为元数据区。

元数据区的内存不足，即方法区和运行时常量池的空间不足。

方法区用于存放 Class 的相关信息，如类名、访问修饰符、常量池、字段描述、方法描述等。

一个类要被垃圾收集器回收，判定条件是比较苛刻的。在经常动态生成大量 Class 的应用中，需要特别注意类的回收状况。这类常见除了 CGLib 字节码增强和动态语言以外，常见的还有：大量 JSP 或动态产生 JSP 文件的应用（JSP 第一次运行时需要编译为 Java 类）、基于 OSGi 的应用（即使是同一个类文件，被不同的加载器加载也会视为不同的类）等。

【示例】方法区出现 OutOfMemoryError

public class MethodAreaOutOfMemoryDemo {

    public static void main(String[] args) {
        while (true) {
            Enhancer enhancer = new Enhancer();
            enhancer.setSuperclass(Bean.class);
            enhancer.setUseCache(false);
            enhancer.setCallback(new MethodInterceptor() {
                @Override
                public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
                    return proxy.invokeSuper(obj, args);
                }
            });
            enhancer.create();
        }
    }

    static class Bean {}

}

【解决】

当由于元空间而面临 OutOfMemoryError 时，第一个解决方案应该是显而易见的。如果应用程序耗尽了内存中的 Metaspace 区域，则应增加 Metaspace 的大小。更改应用程序启动配置并增加以下内容：

1	-XX:MaxMetaspaceSize=512m

上面的配置示例告诉 JVM，允许 Metaspace 增长到 512 MB。

另一种解决方案甚至更简单。你可以通过删除此参数来完全解除对 Metaspace 大小的限制，JVM 默认对 Metaspace 的大小没有限制。但是请注意以下事实：这样做可能会导致大量交换或达到本机物理内存而分配失败。

无法新建本地线程

java.lang.OutOfMemoryError: Unable to create new native thread 这个错误意味着：Java 应用程序已达到其可以启动线程数的限制。

【原因】

当发起一个线程的创建时，虚拟机会在 JVM 内存创建一个 Thread 对象同时创建一个操作系统线程，而这个系统线程的内存用的不是 JVM 内存，而是系统中剩下的内存。

那么，究竟能创建多少线程呢？这里有一个公式：

1	线程数 = (MaxProcessMemory - JVMMemory - ReservedOsMemory) / (ThreadStackSize)

【参数】

MaxProcessMemory - 一个进程的最大内存
JVMMemory - JVM 内存
ReservedOsMemory - 保留的操作系统内存
ThreadStackSize - 线程栈的大小

**给 JVM 分配的内存越多，那么能用来创建系统线程的内存就会越少，越容易发生 unable to create new native thread**。所以，JVM 内存不是分配的越大越好。

但是，通常导致 java.lang.OutOfMemoryError 的情况：无法创建新的本机线程需要经历以下阶段：

JVM 内部运行的应用程序请求新的 Java 线程
JVM 本机代码代理为操作系统创建新本机线程的请求
操作系统尝试创建一个新的本机线程，该线程需要将内存分配给该线程
操作系统将拒绝本机内存分配，原因是 32 位 Java 进程大小已耗尽其内存地址空间（例如，已达到（2-4）GB 进程大小限制）或操作系统的虚拟内存已完全耗尽
引发 java.lang.OutOfMemoryError: Unable to create new native thread 错误。

【示例】

public class UnableCreateNativeThreadOOM {

    public static void main(String[] args) {
        while (true) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        TimeUnit.MINUTES.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

【处理】

可以通过增加操作系统级别的限制来绕过无法创建新的本机线程问题。例如，如果限制了 JVM 可在用户空间中产生的进程数，则应检查出并可能增加该限制：

[root@dev ~]# ulimit -a
core file size          (blocks, -c) 0
--- cut for brevity ---
max user processes              (-u) 1800

通常，OutOfMemoryError 对新的本机线程的限制表示编程错误。当应用程序产生数千个线程时，很可能出了一些问题—很少有应用程序可以从如此大量的线程中受益。

解决问题的一种方法是开始进行线程转储以了解情况。

直接内存溢出

直接内存（Direct Memory）的容量大小可通过 -XX:MaxDirectMemorySize 参数来指定，如果不指定，则默认与 Java 堆最大值（由 -Xmx 指定）一致。

由直接内存导致的内存溢出，一个明显的特征是在 Heap Dump 文件中不会看见有什么明显的异常情况，如果读者发现内存溢出之后产生的 Dump 文件很小，而程序中又直接或间接使用了 DirectMemory（典型的间接使用就是 NIO），那就可以考虑重点检查一下直接内存方面的原因了。

由直接内存导致的内存溢出，一个明显的特征是在 Heapdump 文件中不会看见明显的异常，如果发现 OOM 之后 Dump 文件很小，而程序中又直接或间接使用了 NIO，就可以考虑检查一下是不是这方面的原因。

【示例】直接内存 OutOfMemoryError

/**
 * 本机直接内存溢出示例
 * 错误现象：java.lang.OutOfMemoryError
 * VM Args：-Xmx20M -XX:MaxDirectMemorySize=10M
 */
public class DirectOutOfMemoryDemo {

    private static final int _1MB = 1024 * 1024;

    public static void main(String[] args) throws IllegalAccessException {
        Field unsafeField = Unsafe.class.getDeclaredFields()[0];
        unsafeField.setAccessible(true);
        Unsafe unsafe = (Unsafe) unsafeField.get(null);
        while (true) {
            unsafe.allocateMemory(_1MB);
        }
    }

}

StackOverflowError

HotSpot 虚拟机中并不区分虚拟机栈和本地方法栈。

对于 HotSpot 虚拟机来说，栈容量只由 -Xss 参数来决定。

栈溢出的常见原因：

递归函数调用层数太深 - 线程请求的栈深度大于虚拟机所允许的最大深度，将抛出 StackOverflowError 异常。
大量循环或死循环 - 虚拟机的栈内存允许动态扩展，当扩展栈容量无法申请到足够的内存时，将抛出

OutOfMemoryError 异常。

【示例】递归函数调用层数太深导致 StackOverflowError

/**
 * 以一个无限递归的示例方法来展示栈溢出
 * <p>
 * 栈溢出时，Java 会抛出 StackOverflowError ，出现此种情况是因为方法运行的时候栈的大小超过了虚拟机的上限所致。
 * <p>
 * Java 应用程序唤起一个方法调用时就会在调用栈上分配一个栈帧，这个栈帧包含引用方法的参数，本地参数，以及方法的返回地址。
 * <p>
 * 这个返回地址是被引用的方法返回后，程序能够继续执行的执行点。
 * <p>
 * 如果没有一个新的栈帧所需空间，Java 就会抛出 StackOverflowError。
 * <p>
 * VM 参数：
 * <ul>
 * <li>-Xss228k - 设置栈大小为 228k</li>
 * </ul>
 * <p>
 *
 */
public class StackOverflowErrorDemo {

    private int stackLength = 1;

    public static void main(String[] args) {
        StackOverflowErrorDemo obj = new StackOverflowErrorDemo();
        try {
            obj.recursion();
        } catch (Throwable e) {
            System.out.println("栈深度：" + obj.stackLength);
            e.printStackTrace();
        }
    }

    public void recursion() {
        stackLength++;
        recursion();
    }

}

【示例】大量循环或死循环导致 StackOverflowError

/**
 * 类成员循环依赖，导致 StackOverflowError
 *
 * VM 参数：
 *
 * -Xss228k - 设置栈大小为 228k
 *
 * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
 * @since 2019-06-25
 */
public class StackOverflowErrorDemo2 {

    public static void main(String[] args) {
        A obj = new A();
        System.out.println(obj.toString());
    }

    static class A {

        private int value;

        private B instance;

        public A() {
            value = 0;
            instance = new B();
        }

        @Override
        public String toString() {
            return "<" + value + ", " + instance + ">";
        }

    }

    static class B {

        private int value;

        private A instance;

        public B() {
            value = 10;
            instance = new A();
        }

        @Override
        public String toString() {
            return "<" + value + ", " + instance + ">";
        }

    }

}

参考资料

Redis 哨兵

发表于 2020-06-24 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 4.4k 阅读时长 ≈ 4 分钟

Redis 哨兵

Redis 2.8 版本，新增了哨兵模式，以支持“自动故障转移”，它是 Redis 的 HA 方案。

Redis 哨兵模式由一个或多个 Sentinel 实例组成 Sentinel 集群，可以监控任意多个主服务器，以及这些主服务器的所有从服务器；并在被监视的主服务器进入下线状态时，自动将下线主服务器的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求。

关键词：高可用、监控、选主、故障转移、Raft

哨兵简介

Redis 的主从复制模式，虽然提供了一定程度的 高可用性（High Availability）。但是，当主节点出现故障时，只能通过手动操作将从节点晋升为主节点，这显然是比较低效的。为了解决这个问题，Redis 2.8 版本提供了哨兵模式（Sentinel）来支持“自动故障转移”。

Redis 哨兵模式由一个或多个 Sentinel 实例组成 Sentinel 集群，可以监控任意多个主服务器，以及这些主服务器的所有从服务器；并在被监视的主服务器进入下线状态时，自动将下线主服务器的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求。

Sentinel 的主要功能如下：

监控（Monitoring） - Sentinel 不断检查主从服务器是否正常在工作。
通知（Notification） - Sentinel 可以通过一个 api 来通知系统管理员或者另外的应用程序，被监控的 Redis 实例有一些问题。
自动故障转移（Automatic Failover） - 如果一个主服务器下线，Sentinel 会开始自动故障转移：把一个从节点提升为主节点，并重新配置其他的从节点使用新的主节点，使用 Redis 服务的应用程序在连接的时候也被通知新的地址。
配置提供者（Configuration provider） - Sentinel 给客户端的服务发现提供来源：对于一个给定的服务，客户端连接到 Sentinels 来寻找当前主节点的地址。当故障转移发生的时候，Sentinel 将报告新的地址。

启动哨兵

启动一个 Sentinel 可以使用下面任意一条命令，两条命令效果完全相同。

1 2	redis-sentinel /path/to/sentinel.conf redis-server /path/to/sentinel.conf --sentinel

当一个 Sentinel 启动时，它需要执行以下步骤：

初始化服务器。
使用 Sentinel 专用代码。
初始化 Sentinel 状态。
初始化 Sentinel 的主服务器列表。
创建连向被监视的主服务器的网络连接。

Sentinel 本质上是一个运行在“特殊模式”下的 Redis 服务器。Sentinel 模式下 Redis 服务器只支持 PING、SENTINEL、INFO、SUBSCRIBE、UNSUBSCRIBE、PSUBSCRIBE、PUNSUBSCRIBE 七个命令。

创建连向被监视的主服务器的网络连接，Sentinel 将成为主服务器的客户端，它可以向主服务器发送命令，并从命令回复中获取相关的信息。Sentinel 会读入用户指定的配置文件，为每个要被监视的主服务器创建相应的实例结构，并创建连向主服务器的命令连接和订阅连接：

命令连接 - 专门用于向主服务器发送命令，并接受命令回复。
订阅连接 - 专门用于订阅主服务器的 __sentinel__:hello 频道。

监控

获取服务器信息

默认情况下， Sentinel 以“每十秒一次”的频率向被监视的主服务器和从服务器发送 INFO 命令，并通过分析 INFO 命令的回复来获取服务器的当前信息。

主服务器 - 可以获取主服务器自身信息，以及其所属从服务器的地址信息。
从服务器 - 从服务器自身信息，以及其主服务器的了解状态和地址。

Sentinel 通过向主服务器发送 INFO 命令来获得主服务器属下所有从服务器的地址信息，并为这些从服务器创建相应的实例结构，以及连向这些从服务器的“命令连接”和“订阅连接”。

对于监视同一个主服务器和从服务器的多个 Sentinel 来说，它们会以“每两秒一次”的频率，通过向被监视服务器的 __sentinel__:hello 频道发送消息来向其他 Sentinel 宣告自己的存在。Sentinel 只会与主服务器和从服务器创建命令连接和订阅连接， Sentinel 与 Sentinel 之间则只创建命令连接。

判断下线

主观下线

默认，每个 Sentinel 以“每秒一次”的频率，向它所知的“所有实例”发送一个 PING 命令。

“所知”是指，与 Sentinel 创建了命令连接的实例。
“所有实例”包括了主服务器、从服务器以及其他 Sentinel 实例。

如果，某实例在指定的时长（ down-after-milliseconds 设置的值，单位毫秒）中，未向 Sentinel 发送有效回复， Sentinel 会将该实例判定为“主观下线”。

一个有效的 PING 回复可以是：+PONG、-LOADING 或者 -MASTERDOWN。如果服务器返回除以上三种回复之外的其他回复，又或者在 指定时间 内没有回复 PING 命令，那么 Sentinel 认为服务器返回的回复无效。
“主观下线”适用于所有主节点和从节点。

客观下线

当一个“主服务器”被 Sentinel 标记为“主观下线”后，为了确认其是否真的下线，Sentinel 会向同样监听该主服务器的其他 Sentinel 发起询问。如果有“足够数量”的 Sentinel 在指定的时间范围内认为主服务器已下线，那么这个“主服务器”被标记为“客观下线”。

Sentinel 节点通过 sentinel is-master-down-by-addr 命令，向其它 Sentinel 节点询问对某主服务器的 状态判断。
“足够数量”是指 Sentinel 配置中 quorum 参数所设的值。
客观下线只适用于主节点。

注：默认情况下， Sentinel 以“每十秒一次”的频率向被监视的主服务器和从服务器发送 INFO 命令。当一个主服务器被 Sentinel 标记为“客观下线”时，Sentinel 向该主服务器的所有从服务器发送 INFO 命令的频率，会从“每十秒一次”改为“每秒一次”。

选主

Redis Sentinel 采用 Raft 协议实现了其 Sentinel 选主流程。Raft 是一种共识性算法，想了解其原理，可以参考深入剖析共识性算法 Raft。

当一个“主服务器”被判断为“客观下线”时，监视该主服务器的各个 Sentinel 会进行“协商”，选举出一个领头的 Sentinel（Leader），并由领头 Sentinel 对下线主服务器执行“故障转移”操作。

所有在线 Sentinel 都有资格被选为 Leader。

当一个 Sentinel 认定某主服务器是“客观下线”后，该 Sentinel 会先看看自己是否投过票。
- 如果已投票给其他 Sentinel 了，在 2 倍故障转移的超时时间内，都不能竞选 Leader——相当于它是一个 Follower。
- 如果未投票，那么该 Sentinel 可以竞选 Leader，转为 Candidate。
如 Raft 协议所描述的，Candidate 需要完成几件事情：
1. 更新故障转移状态为 start
2. 将当前纪元（epoch）加 1，表明开始新一轮的选举——这里的 epoch 相当于 Raft 协议中的 term。
3. 将自身的超时时间设为当前时间加上一个随机值，随机值为 1s 内的随机毫秒数。
4. 向其他节点发送 is-master-down-by-addr 命令，请求其他节点投票支持自己，命令会携带自己的 epoch。
5. Candidate 会投票给自己。在 Sentinel 中，投票的方式是把自己 master 结构体里的 leader 和 leader_epoch 改成投给的 Sentinel 和它的 epoch。
其他 Sentinel 收到 Candidate 的 is-master-down-by-addr 命令后，如果 Sentinel 当前 epoch 和 Candidate 传给他的 epoch 一样，说明他已经把自己 master 结构体里的 leader 和 leader_epoch 改成其他 Candidate，相当于把票投给了其他 Candidate。投票给其他 Sentinel 后，在当前 epoch 内，该 Sentinel 就只能成为 Follower。
Candidate 会不断的统计自己的票数，如果满足“当选投票条件”，则该 Candidate 当选 Leader：
1. 票数超过一半（监控主服务器的 Sentinel 的节点数的一半 + 1）
2. 票数超过 Sentinel 配置的 quorum 参数——注：Raft 协议中没有这个限制，这是 Redis Sentinel 所独有的
如果在一个选举周期内（epoch），Candidate 没有满足“当选投票条件”（第 4 点描述的），则竞选失败。
如果在一个选举周期内（epoch），没有一个 Candidate 满足“当选投票条件”，说明所有 Candidate 都竞选失败，本轮选举作废。在等待超过 2 倍故障转移的超时时间后，开始新一轮的选举。
与 Raft 协议不同的是，Leader 并不会把自己成为 Leader 的消息发给其他 Sentinel。当 Leader 完成故障转移后，其他 Sentinel 检测到新的主服务器正常工作后，就会去掉“客观下线”的标识，从而不需要再发起选举。

故障转移

在选举产生出 Sentinel Leader 后，Sentinel Leader 将对已下线的主服务器执行故障转移操作。操作含以下三个步骤：

（1）选出新的主服务器

故障转移第一步，是 Sentinel Leader 在已下线主服务属下的所有从服务器中，挑选一个状态良好、数据完整的从服务器。然后，向这个从服务器发送 SLAVEOF no one 命令，将其转换为主服务器。

Sentinel Leader 如何选出新的主服务器：

删除列表中所有处于下线或断线状态的从服务器。
删除列表中所有最近五秒没有回复过 Sentinel Leader 的 INFO 命令的从服务器。
删除所有与已下线主服务器连接断开超过 down-after-milliseconds * 10 毫秒的从服务器（down-after-milliseconds 指定了判断主服务器下线所需的时间）。
之后， Sentinel Leader 先选出优先级最高的从服务器；如果优先级一样高，再选择复制偏移量最大的从服务器；如果结果还不唯一，则选出运行 ID 最小的从服务器。

（2）修改从服务器的复制目标

选出新的主服务器后，Sentinel Leader 会向所有从服务器发送 SLAVEOF 命令，让它们去复制新的主服务器。

（3）将旧的主服务器变为从服务器

Sentinel Leader 将旧的主服务器标记为从服务器。当旧的主服务器重新上线，Sentinel 会向它发送 SLAVEOF 命令，让其成为从服务器。

参考资料

Redis 复制

发表于 2020-06-24 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 5.2k 阅读时长 ≈ 5 分钟

Redis 复制

在 Redis 中，可以通过执行 SLAVEOF 命令或设置 slaveof 选项，让一个服务器去复制（replicate）另一个服务器，其中，后者叫主服务器（master），前者叫从服务器（slave）。

Redis 2.8 以前的复制不能高效处理断线后重复制的情况，而 Redis 2.8 新添的部分重同步可以解决这个问题。

关键词：SLAVEOF、SYNC、PSYNC、命令传播、心跳

复制简介

Redis 通过 slaveof host port 命令来让一个服务器成为另一个服务器的从服务器。

一个主服务器可以有多个从服务器。不仅主服务器可以有从服务器，从服务器也可以有自己的从服务器，多个从服务器之间可以构成一个主从链。

一个从服务器只能有一个主服务器，并且不支持主主复制。

可以通过复制功能来让主服务器免于执行持久化操作：只要关闭主服务器的持久化功能，然后由从服务器去执行持久化操作即可。

在使用 Redis 复制功能时的设置中，强烈建议在 master 和在 slave 中启用持久化。当不启用时，例如由于非常慢的磁盘性能而导致的延迟问题，应该配置实例来避免重置后自动重启。

从 Redis 2.6 开始，从服务器支持只读模式，并且该模式为从服务器的默认模式。

只读模式由 redis.conf 文件中的 slave-read-only 选项控制，也可以通过 CONFIG SET parameter value 命令来开启或关闭这个模式。
只读从服务器会拒绝执行任何写命令，所以不会出现因为操作失误而将数据不小心写入到了从服务器的情况。

旧版复制

Redis 2.8 版本以前实现方式：SYNC 命令

Redis 的复制功能分为同步（sync）和命令传播（command propagate）两个操作：

同步（sync） - 用于将从服务器的数据库状态更新至主服务器当前的数据库状态。
命令传播（command propagate） - 当主服务器的数据库状态被修改，导致主从数据库状态不一致时，让主从服务器的数据库重新回到一致状态。

同步

SYNC 命令的执行步骤：

从服务器向主服务器发送 SYNC 命令。
收到 SYNC 命令的主服务器执行 BGSAVE 命令，在后台生成一个 RDB 文件，并使用一个缓冲区记录从现在开始执行的所有写命令。
主服务器执行 BGSAVE 完毕后，主服务器会将生成的 RDB 文件发送给从服务器。从服务器接收并载入 RDB 文件，更新自己的数据库状态。
主服务器将记录在缓冲区中的所有写命令发送给从服务器，从服务器执行这些写命令，更新自己的数据库状态。

命令传播

同步操作完成后，主从数据库的数据库状态将达到一致。每当主服务器执行客户端发送的写命令时，主从数据库状态不再一致。需要将写命令发送给从服务器执行，使得二者的数据库状态重新达到一致。

旧版复制的缺陷

从服务器对主服务器的复制存在两种情况：

初次复制 - 从服务器以前没有复制过将要复制的主服务器。
断线后重复制 - 处于命令传播阶段的主从服务器因为网络原因而中断了复制，当从服务器通过自动重连重新连上了主服务器后，继续复制主服务器。

对于初次复制，旧版复制功能可用很好完成任务；但是对于断线后重复制，由于每次任然需要生成 RDB 并传输，效率很低。

🔔 注意：SYNC 命令是一个非常耗费资源的操作。

主服务器执行 BGSAVE 命令生成 RDB 文件，这个操作会耗费主服务器大量的 CPU、内存和磁盘 I/O 资源。

主服务器传输 RDB 文件给从服务器，这个操作会耗费主从服务器大量的网络资源，并对主服务器响应时延产生影响。

从服务器载入 RDB 文件期间，会阻塞其他命令请求。

新版复制

Redis 2.8 版本以后的新实现方式：使用 PSYNC 命令替代 SYNC 命令。

PSYNC 命令具有完整重同步和部分重同步两种模式：

完整重同步（full resychronization） - 用于初次复制。执行步骤与 SYNC 命令基本一致。
部分重同步（partial resychronization） - 用于断线后重复制。如果条件允许，主服务器可以将主从服务器连接断开期间执行的写命令发送给从服务器，从服务器只需接收并执行这些写命令，即可将主从服务器的数据库状态保持一致。

部分重同步

部分重同步功能实现由三个部分构成：

主从服务器的复制偏移量（replication offset）
主服务器的复制积压缓冲区（replication backlog）
服务器的运行 ID

复制偏移量

主服务器和从服务器会分别维护一个复制偏移量。

如果主从服务器的复制偏移量相同，则说明二者的数据库状态一致；
反之，则说明二者的数据库状态不一致。

复制积压缓冲区

复制积压缓冲区是主服务器维护的一个固定长度的先进先出（FIFO）队列，默认大小为 1MB。

复制积压缓冲区会保存一部分最近传播的写命令，并且复制积压缓冲区会为队列中的每个字节记录相应的复制偏移量。

当从服务器断线重连主服务时，从服务器会通过 PSYNC 命令将自己的复制偏移量 offset 发送给主服务器，主服务器会根据这个复制偏移量来决定对从服务器执行何种同步操作。

如果 offset 之后的数据仍然在复制积压缓冲区，则主服务器对从服务器执行部分重同步操作。
反之，则主服务器对从服务器执行完整重同步操作。

🔔 注意：合理调整复制积压缓冲区的大小

Redis 复制积压缓冲区默认大小为 1MB。

复制积压缓冲区的最小大小可以根据公式 second * write_size_per_second 估算。

服务器的运行 ID

每个 Redis 服务器，都有运行 ID，用于唯一识别身份。
运行 ID 在服务器启动时自动生成，由 40 个随机的十六进制字符组成。例如：132e358005e29741f8d7b0a42d666aace286edda

从服务器对主服务器进行初次复制时，主服务器会将自己的运行 ID 传送给从服务器，从服务器会将这个运行 ID 保存下来。

当从服务器断线重连一个主服务器时，从服务器会发送之前保存的运行 ID：

如果保存的运行 ID 和当前主服务器的运行 ID 一致，则说明从服务器断线之前连接的就是这个主服务器，主服务器可以继续尝试执行部分重同步操作；
反之，若运行 ID 不一致，则说明从服务器断线之前连接的不是这个主服务器，主服务器将对从服务器执行完整重同步操作。

PSYNC 命令

了解了部分重同步的实现，PSYNC 的实现就很容易理解了，它的基本工作原理大致如下：

当从服务接收到 SLAVEOF 命令时，先判断从服务器以前是否执行过复制操作。

如果没有复制过任何主服务器，向要复制的主服务器发送 PSYNC ? -1 命令，主动请求进行完整重同步。
反之，向要复制的主服务器发送 PSYNC <runid> <offset> 命令。
- runid 是上一次复制的主服务器的运行 ID。
- offset 是复制偏移量。

接收到 PSYNC <runid> <offset> 命令的主服务会进行分析：

假如主从服务器的 master run id 相同，并且指定的偏移量（offset）在内存缓冲区中还有效，复制就会从上次中断的点开始继续。
如果其中一个条件不满足，就会进行完全重新同步（在 2.8 版本之前就是直接进行完全重新同步）。

心跳检测

在命令传播阶段，从服务器默认会以每秒一次的频率，向主服务器发送命令：

1	REPLCONF ACK <replication_offset>

其中，replication_offset 是从服务器当前的复制偏移量。

发送 REPLCONF ACK 命令对于主从服务器有三个作用：

检测主从服务器的网络连接状态。
辅助实现 min-slaves 选项。
检测命令丢失。

检测主从连接状态

可以通过发送和接收 REPLCONF ACK 命令来检查主从服务器之间的网络连接是否正常：如果主服务器超过一秒没有收到从服务器发来的 REPLCONF ACK 命令，那么主服务器就知道主从服务器之间的连接出现问题了。

可以通过向主服务器发送 INFO replication 命令，在列出的从服务器列表的 lag 一栏中，可以看到从服务器向主服务器发送 REPLCONF ACK 命令已经过去多少秒。

辅助实现 min-slaves 选项

Redis 的 min-slaves-to-write 和 min-slaves-max-lag 两个选项可以防止主服务器在不安全的情况下执行写命令。

【示例】min-slaves 配置项

1 2	min-slaves-to-write 3 min-slaves-max-lag 10

以上配置表示：从服务器小于 3 个，或三个从服务器的延迟（lag）都大于等于 10 秒时，主服务器将拒绝执行写命令。

检测命令丢失

如果因为网络故障，主服务传播给从服务器的写命令丢失，那么从服务器定时向主服务器发送 REPLCONF ACK 命令时，主服务器将发觉从服务器的复制偏移量少于自己的。然后，主服务器就会根据从服务器提交的复制偏移量，在复制积压缓冲区中找到从服务器缺少的数据，并将这些数据重新发送给从服务器。

复制的流程

通过向从服务器发送如下 SLAVEOF 命令，可以让一个从服务器去复制一个主服务器。

1	SLAVEOF <master_ip> <master_port>

步骤 1. 设置主从服务器

配置一个从服务器非常简单，只要在配置文件中增加以下的这一行就可以了：

1	slaveof 127.0.0.1 6379

当然，你需要将代码中的 127.0.0.1 和 6379 替换成你的主服务器的 IP 和端口号。

另外一种方法是调用 SLAVEOF host port 命令，输入主服务器的 IP 和端口，然后同步就会开始：

1 2	127.0.0.1:6379> SLAVEOF 127.0.0.1 10086 OK

步骤 2. 主从服务器建立 TCP 连接。

步骤 3. 发送 PING 检查通信状态。

步骤 4. 身份验证。

如果主服务器没有设置 requirepass ，从服务器没有设置 masterauth，则不进行身份验证；反之，则需要进行身份验证。如果身份验证失败，则放弃执行复制工作。

如果主服务器通过 requirepass 选项设置了密码，那么为了让从服务器的同步操作可以顺利进行，我们也必须为从服务器进行相应的身份验证设置。

对于一个正在运行的服务器，可以使用客户端输入以下命令：

1	config set masterauth <password>

要永久地设置这个密码，那么可以将它加入到配置文件中：

1	masterauth <password>

另外还有几个选项，它们和主服务器执行部分重同步时所使用的复制流缓冲区有关，详细的信息可以参考 Redis 源码中附带的 redis.conf 示例文件。

步骤 5. 发送端口信息。

从服务器执行 REPLCONF listening-port <port-number> ，向主服务器发送从服务器的监听端口号。

步骤 6. 同步。

前文已介绍，此处不赘述。

步骤 7. 命令传播。

在命令传播阶段，从服务器默认会以每秒一次的频率，向主服务发送命令：

1	REPLCONF ACK <replication_coffset>

命令的作用：

检测主从服务器的网络连接状态。
辅助实现 min-slave 选项。
检测命令丢失。

复制的配置项

从 Redis 2.8 开始，为了保证数据的安全性，可以通过配置，让主服务器只在有至少 N 个当前已连接从服务器的情况下，才执行写命令。

不过，因为 Redis 使用异步复制，所以主服务器发送的写数据并不一定会被从服务器接收到，因此，数据丢失的可能性仍然是存在的。

以下是这个特性的运作原理：

从服务器以每秒一次的频率 PING 主服务器一次，并报告复制流的处理情况。
主服务器会记录各个从服务器最后一次向它发送 PING 的时间。
用户可以通过配置，指定网络延迟的最大值 min-slaves-max-lag ，以及执行写操作所需的至少从服务器数量 min-slaves-to-write 。

如果至少有 min-slaves-to-write 个从服务器，并且这些服务器的延迟值都少于 min-slaves-max-lag秒，那么主服务器就会执行客户端请求的写操作。

你可以将这个特性看作 CAP 理论中的 C 的条件放宽版本：尽管不能保证写操作的持久性，但起码丢失数据的窗口会被严格限制在指定的秒数中。

另一方面，如果条件达不到 min-slaves-to-write 和 min-slaves-max-lag 所指定的条件，那么写操作就不会被执行，主服务器会向请求执行写操作的客户端返回一个错误。

以下是这个特性的两个选项和它们所需的参数：

min-slaves-to-write <number of slaves>
min-slaves-max-lag <number of seconds>

详细的信息可以参考 Redis 源码中附带的 redis.conf 示例文件。

参考资料

Redis 实战

发表于 2020-06-24 更新于 2025-09-13 分类于数据库， KV数据库， redis 本文字数： 10k 阅读时长 ≈ 9 分钟

Redis 实战

缓存

缓存是 Redis 最常见的应用场景。

Redis 有多种数据类型，以及丰富的操作命令，并且有着高性能、高可用的特性，非常适合用于分布式缓存。

缓存应用的基本原理，请参考 缓存基本原理 第四 ~ 第六节内容。

BitMap 和 BloomFilter

Redis 除了 5 种基本数据类型外，还支持 BitMap 和 BloomFilter（即布隆过滤器，可以通过 Redis Module 支持）。

BitMap 和 BloomFilter 都可以用于解决缓存穿透问题。要点在于：过滤一些不可能存在的数据。

什么是缓存穿透，可以参考：缓存基本原理

小数据量可以用 BitMap，大数据量可以用布隆过滤器。

分布式锁

使用 Redis 作为分布式锁，基本要点如下：

互斥性 - 使用 setnx 抢占锁。
避免永远不释放锁 - 使用 expire 加一个过期时间，避免一直不释放锁，导致阻塞。
原子性 - setnx 和 expire 必须合并为一个原子指令，避免 setnx 后，机器崩溃，没来得及设置 expire，从而导致锁永不释放。

更多分布式锁的实现方式及细节，请参考：分布式锁基本原理

根据 Redis 的特性，在实际应用中，存在一些应用小技巧。

keys 和 scan

使用 keys 指令可以扫出指定模式的 key 列表。

如果这个 redis 正在给线上的业务提供服务，那使用 keys 指令会有什么问题？

首先，Redis 是单线程的。keys 指令会导致线程阻塞一段时间，线上服务会停顿，直到指令执行完毕，服务才能恢复。

这个时候可以使用 scan 指令，scan 指令可以无阻塞的提取出指定模式的 key 列表，但是会有一定的重复概率，在客户端做一次去重就可以了，但是整体所花费的时间会比直接用 keys 指令长。

不过，增量式迭代命令也不是没有缺点的：举个例子，使用 SMEMBERS 命令可以返回集合键当前包含的所有元素，但是对于 SCAN 这类增量式迭代命令来说，因为在对键进行增量式迭代的过程中，键可能会被修改，所以增量式迭代命令只能对被返回的元素提供有限的保证。

大 Key 如何处理

什么是 Redis Big Key？

Big Key 并不是指 key 的值很大，而是 key 对应的 value 很大。

一般而言，下面这两种情况被称为Big Key：

String 类型的值大于 10 KB；
Hash、List、Set、ZSet 类型的元素的个数超过 5000 个，或总大小超过 10MB

Big Key 会造成什么问题？

Big Key 会带来以下四种影响：

内存分布不均：集群模型在 slot 分片均匀情况下，会出现数据和查询倾斜情况，部分有Big Key 的 Redis 节点占用内存多，QPS 也会比较大。
命令阻塞：Redis 单线程模型，操作大 Key 耗时，阻塞其他命令。
网络传输压力：每次获取Big Key 产生的网络流量较大，如果一个 key 的大小是 1 MB，每秒访问量为 1000，那么每秒会产生 1000MB 的流量，这对于普通千兆网卡的服务器来说是灾难性的。
客户端超时：由于 Redis 执行命令是单线程处理，然后在操作Big Key 时会比较耗时，那么就会阻塞 Redis，从客户端这一视角看，就是很久很久都没有响应。

如何找到Big Key ？

（1）使用 redis-cli --bigkeys

命令：

1	redis-cli -h 127.0.0.1 -p 6379 -a "password" --bigkeys

注意事项：

推荐在从节点执行（主节点执行可能阻塞业务）
低峰期执行 或 加 -i 参数控制扫描间隔（如 -i 0.1 表示每 100ms 扫描一次）

缺点：

只能返回每种数据类型最大的 1 个 Key（无法获取 Top N）
对集合类型只统计元素个数，而非实际内存占用

（2）使用 SCAN + 内存分析命令

遍历所有 Key（避免 KEYS * 阻塞 Redis）：

1	redis-cli --scan --pattern "*" \| while read key; do ...; done

分析 Key 大小：

String：STRLEN $key（字节数）
集合类型（List/Hash/Set/ZSet）：
- 方法 1：LLEN/HLEN/SCARD/ZCARD（元素个数 × 预估元素大小）
- 方法 2（Redis 4.0+）：MEMORY USAGE $key（精确内存占用）

优点：

可自定义筛选条件（如大小 Top 10）
精确计算内存占用

（3）使用 RdbTools 分析 RDB 文件

命令：

1	rdb dump.rdb -c memory --bytes 10240 -f redis.csv # 导出 >10KB 的 Key 到 CSV

适用场景：

离线分析，不影响线上 Redis
精准统计所有 Key 的内存分布

缺点：需要 Redis 生成 RDB 快照

如何删除Big Key？

删除操作的本质是要释放键值对占用的内存空间，不要小瞧内存的释放过程。

释放内存只是第一步，为了更加高效地管理内存空间，在应用程序释放内存时，操作系统需要把释放掉的内存块插入一个空闲内存块的链表，以便后续进行管理和再分配。这个过程本身需要一定时间，而且会阻塞当前释放内存的应用程序。

所以，如果一下子释放了大量内存，空闲内存块链表操作时间就会增加，相应地就会造成 Redis 主线程的阻塞，如果主线程发生了阻塞，其他所有请求可能都会超时，超时越来越多，会造成 Redis 连接耗尽，产生各种异常。

因此，删除Big Key 这一个动作，我们要小心。具体要怎么做呢？这里给出两种方法：

分批次删除
异步删除（Redis 4.0 版本以上）

1、分批次删除

对于删除大 Hash，使用 hscan 命令，每次获取 100 个字段，再用 hdel 命令，每次删除 1 个字段。

Python 代码：

def del_large_hash():
  r = redis.StrictRedis(host='redis-host1', port=6379)
    large_hash_key ="xxx" #要删除的大hash键名
    cursor = '0'
    while cursor != 0:
        # 使用 hscan 命令，每次获取 100 个字段
        cursor, data = r.hscan(large_hash_key, cursor=cursor, count=100)
        for item in data.items():
                # 再用 hdel 命令，每次删除1个字段
                r.hdel(large_hash_key, item[0])

对于删除大 List，通过 ltrim 命令，每次删除少量元素。

Python 代码：

def del_large_list():
  r = redis.StrictRedis(host='redis-host1', port=6379)
  large_list_key = 'xxx'  #要删除的大list的键名
  while r.llen(large_list_key)>0:
      #每次只删除最右100个元素
      r.ltrim(large_list_key, 0, -101)

对于删除大 Set，使用 sscan 命令，每次扫描集合中 100 个元素，再用 srem 命令每次删除一个键。

Python 代码：

def del_large_set():
  r = redis.StrictRedis(host='redis-host1', port=6379)
  large_set_key = 'xxx'   # 要删除的大set的键名
  cursor = '0'
  while cursor != 0:
    # 使用 sscan 命令，每次扫描集合中 100 个元素
    cursor, data = r.sscan(large_set_key, cursor=cursor, count=100)
    for item in data:
      # 再用 srem 命令每次删除一个键
      r.srem(large_size_key, item)

对于删除大 ZSet，使用 zremrangebyrank 命令，每次删除 top 100 个元素。

Python 代码：

def del_large_sortedset():
  r = redis.StrictRedis(host='large_sortedset_key', port=6379)
  large_sortedset_key='xxx'
  while r.zcard(large_sortedset_key)>0:
    # 使用 zremrangebyrank 命令，每次删除 top 100个元素
    r.zremrangebyrank(large_sortedset_key,0,99)

2、异步删除

从 Redis 4.0 版本开始，可以采用异步删除法，用 unlink 命令代替 del 来删除。

这样 Redis 会将这个 key 放入到一个异步线程中进行删除，这样不会阻塞主线程。

除了主动调用 unlink 命令实现异步删除之外，我们还可以通过配置参数，达到某些条件的时候自动进行异步删除。

主要有 4 种场景，默认都是关闭的：

lazyfree-lazy-eviction no
lazyfree-lazy-expire no
lazyfree-lazy-server-del
noslave-lazy-flush no

它们代表的含义如下：

lazyfree-lazy-eviction：表示当 Redis 运行内存超过 maxmeory 时，是否开启 lazy free 机制删除；
lazyfree-lazy-expire：表示设置了过期时间的键值，当过期之后是否开启 lazy free 机制删除；
lazyfree-lazy-server-del：有些指令在处理已存在的键时，会带有一个隐式的 del 键的操作，比如 rename 命令，当目标键已存在，Redis 会先删除目标键，如果这些目标键是一个 big key，就会造成阻塞删除的问题，此配置表示在这种场景中是否开启 lazy free 机制删除；
slave-lazy-flush：针对 slave (从节点) 进行全量数据同步，slave 在加载 master 的 RDB 文件前，会运行 flushall 来清理自己的数据，它表示此时是否开启 lazy free 机制删除。

建议开启其中的 lazyfree-lazy-eviction、lazyfree-lazy-expire、lazyfree-lazy-server-del 等配置，这样就可以有效的提高主线程的执行效率。

最受欢迎文章

选出最受欢迎文章，需要支持对文章进行评分。

对文章进行投票

（1）使用 HASH 存储文章

使用 HASH 类型存储文章信息。其中：key 是文章 ID；field 是文章的属性 key；value 是属性对应值。

操作：

存储文章信息 - 使用 HSET 或 HMGET 命令
查询文章信息 - 使用 HGETALL 命令
添加投票 - 使用 HINCRBY 命令

（2）使用 ZSET 针对不同维度集合排序

使用 ZSET 类型分别存储按照时间排序和按照评分排序的文章 ID 集合。

操作：

添加记录 - 使用 ZADD 命令
添加分数 - 使用 ZINCRBY 命令
取出多篇文章 - 使用 ZREVRANGE 命令

（3）为了防止重复投票，使用 SET 类型记录每篇文章 ID 对应的投票集合。

操作：

添加投票者 - 使用 SADD 命令
设置有效期 - 使用 EXPIRE 命令

（4）假设 user:115423 给 article:100408 投票，分别需要高更新评分排序集合以及投票集合。

当需要对一篇文章投票时，程序需要用 ZSCORE 命令检查记录文章发布时间的有序集合，判断文章的发布时间是否超过投票有效期（比如：一星期）。

public void articleVote(Jedis conn, String user, String article) {
    // 计算文章的投票截止时间。
    long cutoff = (System.currentTimeMillis() / 1000) - ONE_WEEK_IN_SECONDS;

    // 检查是否还可以对文章进行投票
    // （虽然使用散列也可以获取文章的发布时间，
    // 但有序集合返回的文章发布时间为浮点数，
    // 可以不进行转换直接使用）。
    if (conn.zscore("time:", article) < cutoff) {
        return;
    }

    // 从article:id标识符（identifier）里面取出文章的ID。
    String articleId = article.substring(article.indexOf(':') + 1);

    // 如果用户是第一次为这篇文章投票，那么增加这篇文章的投票数量和评分。
    if (conn.sadd("voted:" + articleId, user) == 1) {
        conn.zincrby("score:", VOTE_SCORE, article);
        conn.hincrBy(article, "votes", 1);
    }
}

发布并获取文章

发布文章：

添加文章 - 使用 INCR 命令计算新的文章 ID，填充文章信息，然后用 HSET 命令或 HMSET 命令写入到 HASH 结构中。
将文章作者 ID 添加到投票名单 - 使用 SADD 命令添加到代表投票名单的 SET 结构中。
设置投票有效期 - 使用 EXPIRE 命令设置投票有效期。

public String postArticle(Jedis conn, String user, String title, String link) {
    // 生成一个新的文章ID。
    String articleId = String.valueOf(conn.incr("article:"));

    String voted = "voted:" + articleId;
    // 将发布文章的用户添加到文章的已投票用户名单里面，
    conn.sadd(voted, user);
    // 然后将这个名单的过期时间设置为一周（第3章将对过期时间作更详细的介绍）。
    conn.expire(voted, ONE_WEEK_IN_SECONDS);

    long now = System.currentTimeMillis() / 1000;
    String article = "article:" + articleId;
    // 将文章信息存储到一个散列里面。
    HashMap<String, String> articleData = new HashMap<String, String>();
    articleData.put("title", title);
    articleData.put("link", link);
    articleData.put("user", user);
    articleData.put("now", String.valueOf(now));
    articleData.put("votes", "1");
    conn.hmset(article, articleData);

    // 将文章添加到根据发布时间排序的有序集合和根据评分排序的有序集合里面。
    conn.zadd("score:", now + VOTE_SCORE, article);
    conn.zadd("time:", now, article);

    return articleId;
}

分页查询最受欢迎文章：

使用 ZINTERSTORE 命令根据页码、每页记录数、排序号，根据评分值从大到小分页查出文章 ID 列表。

public List<Map<String, String>> getArticles(Jedis conn, int page, String order) {
    // 设置获取文章的起始索引和结束索引。
    int start = (page - 1) * ARTICLES_PER_PAGE;
    int end = start + ARTICLES_PER_PAGE - 1;

    // 获取多个文章ID。
    Set<String> ids = conn.zrevrange(order, start, end);
    List<Map<String, String>> articles = new ArrayList<>();
    // 根据文章ID获取文章的详细信息。
    for (String id : ids) {
        Map<String, String> articleData = conn.hgetAll(id);
        articleData.put("id", id);
        articles.add(articleData);
    }

    return articles;
}

对文章进行分组

如果文章需要分组，功能需要分为两块：

记录文章属于哪个群组
负责取出群组里的文章

将文章添加、删除群组：

public void addRemoveGroups(Jedis conn, String articleId, String[] toAdd, String[] toRemove) {
    // 构建存储文章信息的键名。
    String article = "article:" + articleId;
    // 将文章添加到它所属的群组里面。
    for (String group : toAdd) {
        conn.sadd("group:" + group, article);
    }
    // 从群组里面移除文章。
    for (String group : toRemove) {
        conn.srem("group:" + group, article);
    }
}

取出群组里的文章：

通过对存储群组文章的集合和存储文章评分的有序集合执行 ZINTERSTORE 命令，可以得到按照文章评分排序的群组文章。
通过对存储群组文章的集合和存储文章发布时间的有序集合执行 ZINTERSTORE 命令，可以得到按照文章发布时间排序的群组文章。

public List<Map<String, String>> getGroupArticles(Jedis conn, String group, int page, String order) {
    // 为每个群组的每种排列顺序都创建一个键。
    String key = order + group;
    // 检查是否有已缓存的排序结果，如果没有的话就现在进行排序。
    if (!conn.exists(key)) {
        // 根据评分或者发布时间，对群组文章进行排序。
        ZParams params = new ZParams().aggregate(ZParams.Aggregate.MAX);
        conn.zinterstore(key, params, "group:" + group, order);
        // 让Redis在60秒钟之后自动删除这个有序集合。
        conn.expire(key, 60);
    }
    // 调用之前定义的getArticles函数来进行分页并获取文章数据。
    return getArticles(conn, page, key);
}

管理令牌

网站一般会以 Cookie、Session、令牌这类信息存储用户身份信息。

可以将 Cookie/Session/令牌和用户的映射关系存储在 HASH 结构。

下面以令牌来举例。

查询令牌

public String checkToken(Jedis conn, String token) {
    // 尝试获取并返回令牌对应的用户。
    return conn.hget("login:", token);
}

更新令牌

用户每次访问页面，可以记录下令牌和当前时间戳的映射关系，存入一个 ZSET 结构中，以便分析用户是否活跃，继而可以周期性清理最老的令牌，统计当前在线用户数等行为。
用户如果正在浏览商品，可以记录到用户最近浏览过的商品有序集合中（集合可以限定数量，超过数量进行裁剪），存入到一个 ZSET 结构中，以便分析用户最近可能感兴趣的商品，以便推荐商品。

public void updateToken(Jedis conn, String token, String user, String item) {
    // 获取当前时间戳。
    long timestamp = System.currentTimeMillis() / 1000;
    // 维持令牌与已登录用户之间的映射。
    conn.hset("login:", token, user);
    // 记录令牌最后一次出现的时间。
    conn.zadd("recent:", timestamp, token);
    if (item != null) {
        // 记录用户浏览过的商品。
        conn.zadd("viewed:" + token, timestamp, item);
        // 移除旧的记录，只保留用户最近浏览过的25个商品。
        conn.zremrangeByRank("viewed:" + token, 0, -26);
        conn.zincrby("viewed:", -1, item);
    }
}

清理令牌

上一节提到，更新令牌时，将令牌和当前时间戳的映射关系，存入一个 ZSET 结构中。所以可以通过排序得知哪些令牌最老。如果没有清理操作，更新令牌占用的内存会不断膨胀，直到导致机器宕机。

比如：最多允许存储 1000 万条令牌信息，周期性检查，一旦发现记录数超出 1000 万条，将 ZSET 从新到老排序，将超出 1000 万条的记录清除。

public static class CleanSessionsThread extends Thread {

    private Jedis conn;

    private int limit;

    private volatile boolean quit;

    public CleanSessionsThread(int limit) {
        this.conn = new Jedis("localhost");
        this.conn.select(15);
        this.limit = limit;
    }

    public void quit() {
        quit = true;
    }

    @Override
    public void run() {
        while (!quit) {
            // 找出目前已有令牌的数量。
            long size = conn.zcard("recent:");
            // 令牌数量未超过限制，休眠并在之后重新检查。
            if (size <= limit) {
                try {
                    sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
                continue;
            }

            // 获取需要移除的令牌ID。
            long endIndex = Math.min(size - limit, 100);
            Set<String> tokenSet = conn.zrange("recent:", 0, endIndex - 1);
            String[] tokens = tokenSet.toArray(new String[tokenSet.size()]);

            // 为那些将要被删除的令牌构建键名。
            ArrayList<String> sessionKeys = new ArrayList<String>();
            for (String token : tokens) {
                sessionKeys.add("viewed:" + token);
            }

            // 移除最旧的那些令牌。
            conn.del(sessionKeys.toArray(new String[sessionKeys.size()]));
            conn.hdel("login:", tokens);
            conn.zrem("recent:", tokens);
        }
    }

}