构建高效的串行任务执行器：SerialExecutor深度解析

news2024/11/14 12:13:39

本文主要介绍怎么去实现一个支持串行执行任务的SerialExecutor执行器

摘要

在复杂的异步编程中，有时我们需要确保任务以串行的方式执行，以维护任务间的依赖关系或顺序。SerialExecutor 是一个自定义的执行器，它封装了 Java 的 Executor 接口，确保任务按照 FIFO（先进先出）的顺序执行。
本文将深入探讨 SerialExecutor的工作原理，并分析其实现细节。

引言

在 Java 并发编程中，Executor 框架提供了一种灵活的机制来异步执行任务。然而，某些场景要求任务必须串行执行。
例如，在处理具有顺序依赖性的数据库操作或需要按特定顺序发送的网络请求时。SerialExecutor正是为了满足这类需求而设计的。

执行器接口

Executor 接口只提供了一个函数：execute()。
ExecutorService 接口是对Executor接口进行了扩展，额外提供了很多其他函数，包括submit()、shutdown()、shutdownNow()、awaitTermination()等。

我们最常见的ThreadPoolExecutor就是执行器的一种，它实现了ExecutorService接口。
而 Executors 工具类则是用来创建各种执行器。

自定义执行器

SerialExecutor是一个实现了 Executor 接口的类，它使用一个任务队列和一个单线程执行器来保证任务的串行执行。

核心组件

tasks：一个线程安全的队列，用于存储待执行的任务。
executor：一个 Executor 实例，用于实际执行任务。
active：一个 Runnable 任务引用，表示当前正在执行的任务。

构造方法

SerialExecutor提供了两个构造方法：

一个使用默认的单线程执行器。
一个允许用户自定义执行器。

执行方法

execute 方法是 SerialExecutor 的核心，它接受一个 Runnable 任务并将其封装在一个 lambda 表达式中，然后添加到任务队列中。如果当前没有活动任务，execute 方法会立即调度下一个任务。

调度方法

scheduleNext 方法负责从任务队列中取出下一个任务，并使用 executor 执行它。此方法在当前任务执行完毕后被调用，确保了任务的连续执行。

代码实现

public class SerialExecutor implements Executor {

    final Queue<Runnable> tasks = new ArrayDeque<>();

    final Executor executor;

    Runnable active;

    SerialExecutorJDK() {
        this.executor = Executors.newSingleThreadExecutor();
    }
    
    SerialExecutorJDK(Executor executor) {
        this.executor = executor;
    }

    @Override
    public synchronized void execute(final Runnable r) {
        tasks.offer(() -> {
            try {
                r.run();
            } finally {
                scheduleNext();
            }
        });

        if (active == null) {
            scheduleNext();
        }
    }

    private synchronized void scheduleNext() {
        if ((active = tasks.poll()) != null) {
            executor.execute(active);
        }
    }
}