Java Executors 框架源码深度分析(下)

我真的是大笨蛋

411人浏览 · 2025-09-22 21:49:46

我真的是大笨蛋 · 2025-09-22 21:49:46 发布

Callable 适配器机制

Runnable 到 Callable 的适配模式

适配器设计思想

/**
 * A callable that runs given task and returns given result.
 */
private static final class RunnableAdapter<T> implements Callable<T> {
    private final Runnable task;
    private final T result;
    
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    
    public T call() {
        task.run();
        return result;
    }
    
    public String toString() {
        return super.toString() + "[Wrapped task = " + task + "]";
    }
}

设计模式分析

适配器模式：将 Runnable 接口适配为 Callable 接口
对象组合：持有被适配对象的引用
接口转换：将 void run() 转换为 T call()

工厂方法封装

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

public static Callable<Object> callable(Runnable task) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<Object>(task, null);
}

类型系统设计

泛型参数 T 提供类型安全
重载方法支持有返回值和无返回值两种场景
空值检查确保健壮性

匿名内部类实现方式

PrivilegedAction 适配

public static Callable<Object> callable(final PrivilegedAction<?> action) {
    if (action == null)
        throw new NullPointerException();
    return new Callable<Object>() {
        public Object call() { 
            return action.run(); 
        }
    };
}

匿名类优势

简洁性：无需定义单独的命名类
封装性：实现细节完全隐藏
上下文捕获：自动捕获 final 变量

PrivilegedExceptionAction 适配

public static Callable<Object> callable(final PrivilegedExceptionAction<?> action) {
    if (action == null)
        throw new NullPointerException();
    return new Callable<Object>() {
        public Object call() throws Exception { 
            return action.run(); 
        }
    };
}

异常处理

异常签名传播：保留 throws Exception 声明
类型安全：编译期异常检查
透明代理：调用者感知原始异常类型

特权安全调用机制

特权调用设计原理

PrivilegedCallable 核心实现

private static final class PrivilegedCallable<T> implements Callable<T> {
    final Callable<T> task;
    @SuppressWarnings("removal")
    final AccessControlContext acc;

    @SuppressWarnings("removal")
    PrivilegedCallable(Callable<T> task) {
        this.task = task;
        this.acc = AccessController.getContext();
    }

    @SuppressWarnings("removal")
    public T call() throws Exception {
        try {
            return AccessController.doPrivileged(
                new PrivilegedExceptionAction<T>() {
                    public T run() throws Exception {
                        return task.call();
                    }
                }, acc);
        } catch (PrivilegedActionException e) {
            throw e.getException();
        }
    }
}

安全机制分析

访问控制上下文捕获

// 构造时捕获当前安全上下文
this.acc = AccessController.getContext();

// 执行时在捕获的上下文中运行
AccessController.doPrivileged(privilegedAction, acc);

异常处理链

catch (PrivilegedActionException e) {
    throw e.getException();  // 解包真实异常
}

类加载器感知的特权调用

private static final class PrivilegedCallableUsingCurrentClassLoader<T>
        implements Callable<T> {
    final Callable<T> task;
    @SuppressWarnings("removal")
    final AccessControlContext acc;
    final ClassLoader ccl;

    @SuppressWarnings("removal")
    PrivilegedCallableUsingCurrentClassLoader(Callable<T> task) {
        SecurityManager sm = System.getSecurityManager();
        if (sm != null) {
            // 权限预检查
            sm.checkPermission(SecurityConstants.GET_CLASSLOADER_PERMISSION);
            sm.checkPermission(new RuntimePermission("setContextClassLoader"));
        }
        this.task = task;
        this.acc = AccessController.getContext();
        this.ccl = Thread.currentThread().getContextClassLoader();
    }
}

安全检查策略

Fail-fast：构造时立即检查权限，避免运行时失败
最小权限：只请求必要的权限
防御性编程：空安全检查和权限验证

类加载器上下文管理

类加载器恢复机制

@SuppressWarnings("removal")
public T call() throws Exception {
    try {
        return AccessController.doPrivileged(
            new PrivilegedExceptionAction<T>() {
                public T run() throws Exception {
                    Thread t = Thread.currentThread();
                    ClassLoader cl = t.getContextClassLoader();
                    if (ccl == cl) {
                        return task.call();
                    } else {
                        t.setContextClassLoader(ccl);
                        try {
                            return task.call();
                        } finally {
                            t.setContextClassLoader(cl);
                        }
                    }
                }
            }, acc);
    } catch (PrivilegedActionException e) {
        throw e.getException();
    }
}

资源管理模式

// try-finally 保证资源恢复
ClassLoader original = Thread.currentThread().getContextClassLoader();
try {
    Thread.currentThread().setContextClassLoader(targetClassLoader);
    // 执行任务
    return task.call();
} finally {
    Thread.currentThread().setContextClassLoader(original);
}

性能优化策略

if (ccl == cl) {
    return task.call();  // 避免不必要的类加载器设置
} else {
    // 需要时才设置和恢复类加载器
}

优化思想

短路优化：类加载器相同时跳过设置操作
惰性操作：只在必要时修改线程状态
最小化影响：减少上下文切换开销

线程工厂模式

线程命名策略

private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        @SuppressWarnings("removal")
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
    }
}

命名体系

池级别标识：poolNumber 静态计数器，全局唯一
线程级别标识：threadNumber 实例计数器，池内唯一
可读性格式：pool-1-thread-1 便于监控和调试

线程标准化配置

public Thread newThread(Runnable r) {
    Thread t = new Thread(group, r,
                          namePrefix + threadNumber.getAndIncrement(),
                          0);
    if (t.isDaemon())
        t.setDaemon(false);
    if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
    return t;
}

特权线程工厂安全机制

安全上下文传递

private static class PrivilegedThreadFactory extends DefaultThreadFactory {
    @SuppressWarnings("removal")
    final AccessControlContext acc;
    final ClassLoader ccl;

    public Thread newThread(final Runnable r) {
        return super.newThread(new Runnable() {
            @SuppressWarnings("removal")
            public void run() {
                AccessController.doPrivileged(new PrivilegedAction<>() {
                    public Void run() {
                        Thread.currentThread().setContextClassLoader(ccl);
                        r.run();
                        return null;
                    }
                }, acc);
            }
        });
    }
}

执行上下文封装

// 原始任务被包装在特权上下文中执行
Runnable privilegedRunnable = new Runnable() {
    public void run() {
        AccessController.doPrivileged(privilegedAction, acc);
    }
};

// 线程工厂创建执行特权任务的线程
Thread thread = new Thread(privilegedRunnable);

嵌套装饰模式

设计层次

原始任务 (r) -> 特权装饰器 (设置类加载器 + 安全上下文)  -> 线程执行包装器 -> 线程对象

委托执行器服务

完整源码

    /**
     * A wrapper class that exposes only the ExecutorService methods
     * of an ExecutorService implementation.
     */
    private static class DelegatedExecutorService
            implements ExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) {
            try {
                e.execute(command);
            } finally { reachabilityFence(this); }
        }
        public void shutdown() {
            try {
                e.shutdown();
            } finally { reachabilityFence(this); }
        }
        public List<Runnable> shutdownNow() {
            try {
                return e.shutdownNow();
            } finally { reachabilityFence(this); }
        }
        public boolean isShutdown() {
            try {
                return e.isShutdown();
            } finally { reachabilityFence(this); }
        }
        public boolean isTerminated() {
            try {
                return e.isTerminated();
            } finally { reachabilityFence(this); }
        }
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            try {
                return e.awaitTermination(timeout, unit);
            } finally { reachabilityFence(this); }
        }
        public Future<?> submit(Runnable task) {
            try {
                return e.submit(task);
            } finally { reachabilityFence(this); }
        }
        public <T> Future<T> submit(Callable<T> task) {
            try {
                return e.submit(task);
            } finally { reachabilityFence(this); }
        }
        public <T> Future<T> submit(Runnable task, T result) {
            try {
                return e.submit(task, result);
            } finally { reachabilityFence(this); }
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
            try {
                return e.invokeAll(tasks);
            } finally { reachabilityFence(this); }
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                             long timeout, TimeUnit unit)
            throws InterruptedException {
            try {
                return e.invokeAll(tasks, timeout, unit);
            } finally { reachabilityFence(this); }
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
            try {
                return e.invokeAny(tasks);
            } finally { reachabilityFence(this); }
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                               long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            try {
                return e.invokeAny(tasks, timeout, unit);
            } finally { reachabilityFence(this); }
        }
    }

可达性栅栏机制

内存模型保证

public void execute(Runnable command) {
    try {
        e.execute(command);
    } finally { 
        reachabilityFence(this); 
    }
}

技术原理

import static java.lang.ref.Reference.reachabilityFence;

// 防止 this 在方法执行期间被垃圾回收
// 确保委托对象在方法调用期间保持强可达

使用场景

异步回调：防止回调执行前对象被回收
内存安全：避免 use-after-free 类问题

异常安全实现

public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
    throws InterruptedException, ExecutionException {
    try {
        return e.invokeAny(tasks);
    } finally { 
        reachabilityFence(this); 
    }
}