EmbeddedChannel
LocalServerChannel
NioDatagramChannel
NioSctpChannel
NioSocketChannel
...

ChannelHandlerAdapter
ChannelInboundHandlerAdapter
ChannelOutboundHandlerAdapter
ChannelDuplexHandler

SimpleChannelInboundHandler<I> extends ChannelInboundHandlerAdapter

// 底层源码:
unsafe.allocateMemory(size)

-XX:MaxDirectMemorySize
-Dsun.nio.MaxDirectMemorySize
directMemory = Runtime.getRuntime().maxMemory()

private static class Deallocator implements Runnable  {
    private static Unsafe unsafe = Unsafe.getUnsafe();
    private long address;
    private long size;
    private int capacity;
    private Deallocator(long address, long size, int capacity) {
        assert (address != 0);
        this.address = address;
        this.size = size;
        this.capacity = capacity;
    }
 
    public void run() {
        if (address == 0) {
            // Paranoia
            return;
        }
        unsafe.freeMemory(address);
        address = 0;
        Bits.unreserveMemory(size, capacity);
    }
}

public class CountTask extends RecursiveTask<Integer> {
    private final static int THREDSHOLD = 2;
    private final int start;
    private final int end;

    public CountTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum = 0;
        boolean canCompute = (end - start) <= THREDSHOLD;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 任务太大, 分治
            int mid = start + (end - start) / 2;
            CountTask left = new CountTask(start, mid);
            CountTask right = new CountTask(mid + 1, end);
            left.fork();
            right.fork();
            int leftRes = left.join();
            int rightRes = right.join();
            sum = leftRes + rightRes;
        }
        return sum;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();// 可以传入并行度参数, 不传则默认从RunningTime取核数作为并行度
        CountTask task = new CountTask(1, 10);// RecursiveTask=> ForkJoinTask
        Future<Integer> res = forkJoinPool.submit(task);// ForkJoinTask
        System.out.print(res.get());


    }
}

// 外部接口：
// 1. 需要返回值：
RecursiveTask => ForkJoinTask implements Future<V>, Serializable

// 2. 不需要返回值：
RecursiveAction => ForkJoinTask implements Future<V>, Serializable

// 3. 管理线程池和工作任务：
ForkJoinPool => extends AbstractExecutorService => implements ExecutorService

// 内部组件 
@sun.misc.Contended
    static final class WorkQueue {
        // 核心部分: 
        volatile int scanState;    // versioned, <0: inactive; odd:scanning
        int stackPred;             // pool stack (ctl) predecessor
        volatile int qlock;        // 1: locked, < 0: terminate; else 0
        volatile int base;         // 队尾,poll用,会被窃取线程更改。
        int top;                   // 队首,push用,只会被当前线程更改,
                                   // 因此没有volatile
        ForkJoinTask<?>[] array;   // 双端队列
        volatile Thread parker;    // == owner during call to park; else null
        volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
        volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer
        // 其他成员省略:
        ... 
    ... 
}

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public interface Callable<V> {
    V call() throws Exception;
}

Future<String> future = executorService.submit(new MyCallable());

Callable<Integer> callable = () -> new Random().nextInt(100);
FutureTask<Integer> future = new FutureTask<>(callable);
// 方法1: 可以由当前线程执行:
// future.run();
// 方法2: 可以由别的线程来执行:
new Thread(future).start();
System.out.println(future.get());

public class FutureTask<V> implements RunnableFuture<V> 
    private volatile int state;// 可能的取值如下: 
    private static final int NEW          = 0; // 初始状态
    private static final int COMPLETING   = 1; // 计算中
    private static final int NORMAL       = 2; // 正常结束 （终结状态）
    private static final int EXCEPTIONAL  = 3; // 异常结束 （终结状态）
    private static final int CANCELLED    = 4; // 已取消   （终结状态）
    private static final int INTERRUPTING = 5; // 打断中
    private static final int INTERRUPTED  = 6; // 已打断   （终结状态）

    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    private volatile WaitNode waiters;
// 省略很多方法: ...

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}
 
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

LinkedTransferQueue->TransferQueue->BlockingQueue

public class LinkedTransferQueue<E> extends AbstractQueue<E>
    implements TransferQueue<E>, java.io.Serializable {

void transfer(E e) throws InterruptedException;// 阻塞
boolean tryTransfer(E e);
boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;// 一段时间阻塞
boolean hasWaitingConsumer();
int getWaitingConsumerCount();

/**一般来说，不抛`InterruptedException`异常的是不阻塞的方法;
抛`InterruptedException`异常的是阻塞,或者定时阻塞的方法.
*/

private static final int NOW   = 0; // 即使没有可以用的空间或者可用的数据,也立即返回。
// for 不计时的poll和tryTransfer

private static final int ASYNC = 1; // 异步/非阻塞/立即返回. 
// for offer, put, add

private static final int SYNC  = 2; // 同步/阻塞/无限制等待.
// for transfer, take

private static final int TIMED = 3; // 可以等待一定时间.
// for 计时的poll和 tryTransfer

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class DynamicProxyTest {
    private static class AHandler implements InvocationHandler {

        // 目标对象
        private Object a1Impl;

        public AHandler(Object a1Impl) {
            super();
            this.a1Impl = a1Impl;
        }

        /**
         * 创建代理实例
         *
         * @return proxy Object
         * @throws Throwable e
         */
        public Object getProxy() throws Throwable {
            // 实现1: 进行代理:
            return Proxy.newProxyInstance(Thread.currentThread()
                    .getContextClassLoader(), this.a1Impl.getClass()
                    .getInterfaces(), this);
            // 实现2: 不进行代理,依然使用默认实现:
            // 这样写只返回了目标对象，没有生成代理对象。
            // return a1Impl;
        }

        /**
         * 实现InvocationHandler接口方法
         * 执行目标对象的方法，并进行增强
         */
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            // 使用代理proxy,调用方法method,参数为args
            Object result;
            System.out.println("代理类方法，进行了增强。。。");
            System.out.println("事务开始。。。");
            // 执行目标方法对象
            // 调用a1Impl的method方法,参数为args
            result = method.invoke(a1Impl, args);
            System.out.println("事务结束。。。");
            return result;
        }
    }

    public static void main(String[] args) throws Throwable {
        A a1 = new A1();
        AHandler handler = new AHandler(a1);
        // 根据目标生成代理对象
        A a2 = (A) handler.getProxy();
        a2.doSomething();
    }

}

public class AImpl {// 并不继承接口,但不是final的(可以有子类)

    public void doSomething() {
        System.out.println("AImpl doSomething");
    }
}

import java.lang.reflect.Method;
import net.sf.cglib.proxy.Enhancer;
import net.sf.cglib.proxy.MethodInterceptor;
import net.sf.cglib.proxy.MethodProxy;

/**
 * @author xiaoyue26
 */
public class AInterceptor implements MethodInterceptor {
    private Object target;

    /**
     * 创建代理实例
     *
     * @param target
     * @return
     */
    public Object getInstance(Object target) {
        this.target = target;
        Enhancer enhancer = new Enhancer();
        enhancer.setSuperclass(this.target.getClass());// 继承传入的对象
        // 设置回调方法
        enhancer.setCallback(this);
        // 创建代理对象
        return enhancer.create();
    }

    /**
     * 实现MethodInterceptor接口要重写的方法。
     * 回调方法
     */
    @Override
    public Object intercept(Object obj, Method method
            , Object[] args, MethodProxy proxy) throws Throwable {
        System.out.println("事务开始。。。");
        Object result = proxy.invokeSuper(obj, args);// 调用父类对象(调用原实现)
        System.out.println("事务结束。。。");
        return result;
    }

    public static void main(String[] args) {
        AInterceptor aInterceptor = new AInterceptor();
        AImpl a1 = (AImpl) aInterceptor.getInstance(new AImpl());
        a1.doSomething();
    }


}

<T>Future<T>submit(Callable<T>task)

ThreadPoolExecutor->AbstractExecutorService->ExecutorService
ThreadPoolExecutor extends AbstractExecutorService
abstract class AbstractExecutorService implements ExecutorService

// ThreadPoolExecutor实现类中的通用方法:
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
//  Executors工具类中的FixedThreadPool:
public static ExecutorService newFixedThreadPool(int nThread){
    return new ThreadPoolExecutor(nThread,nThread,0L,TimeUnit.MILLISECENDS
    ,new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE
            ,60L, TimeUnit.SECONDS
            ,new SynchronousQueue<Runnable>());
    }