package com.fly.singleton;

/**
 * 单例模式写法1:饿汉式单例模式
 */
public class SingletonPattern01 {
    /**
     * static修饰SingletonPattern01单例类,该实例在类加载时就被初始化
     */
    private static final SingletonPattern01 instance = new SingletonPattern01();

    /**
     * 获取单例实例
     * @return 返回单例实例
     */
    private static SingletonPattern01 getInstance(){
        return instance;
    }
}

package com.fly.singleton;

/**
 * 单例模式写法2:懒汉单例模式,可以实现延迟初始化实例,但存在线程安全问题
 */
public class SingletonPattern02 {
    private static SingletonPattern02 instance;

    // 私有构造器
    private SingletonPattern02() {
    }

    /**
     * 获取单例实例。在懒汉单例模式下获取实例首先会判断单例实例是否为null,
     * 如果为null则进行初始化,否则直接返回单例实例。
     * @return 单例实例
     */
    static SingletonPattern02 getInstance() {
        if (instance == null) { // 注意点1
            instance = new SingletonPattern02(); // 注意点2
        }
        return instance;
    }
}

package com.fly.singleton;

/**
 * 单例模式写法3:同步懒汉单例模式,通过synchronized同步机制解决
 * 饿汉单例模式产生的线程安全问题。在线程竞争激烈的情况下,内置锁(synchronized)会
 * 被升级为重量锁,重量级锁开销大、性能低,在高并发场景不推荐使用。
 */
public class SingletonPattern03 {
    private static SingletonPattern03 instance;

    // 私有构造器
    private SingletonPattern03() {}

    /**
     * 获取单例实例。通过synchronized同步机制使线程同步访问
     * @return 单例实例
     */
    static synchronized SingletonPattern03 getInstance() {
        if (instance == null) {
            instance = new SingletonPattern03();
        }
        return instance;
    }
}

package com.fly.singleton;

/**
 * 单例模式写法4:双重检查锁。
 */
public class SingletonPattern04 {
    private static SingletonPattern04 instance;

    private SingletonPattern04() {
    }

    public static SingletonPattern04 getInstance() {
        if (instance == null) { // 检查1
            /**
             * 使用SingletonPattern04类做为锁资源,
             * 同步代码块粒度比同步方法更细,性能相对更好
             */
            synchronized (SingletonPattern04.class) {
                if (instance == null) { // 检查2
                    instance = new SingletonPattern04();
                }
            }
        }
        return instance;
    }
}

package com.fly.singleton;

/**
 * 单例模式写法5:双重检查锁 + volatile。
 */
public class SingletonPattern05 {
    /**
     * 使用volatile修饰单例对象,volatile可以保证线程之间的可见性,
     * 在线程操作时禁止指令重排序
     */
    private static volatile SingletonPattern05 instance;

    private SingletonPattern05() {}

    public static SingletonPattern05 getInstance() {
        if (instance == null) { // 检查1
            synchronized (SingletonPattern04.class) {
                if (instance == null) { // 检查2
                    instance = new SingletonPattern05();
                }
            }
        }
        return instance;
    }
}

package com.fly.singleton;

/**
 * 单例模式写法6:静态内部类模式,其核心是通过静态内部类获取实例,
 * 对比饿汉单例模式,都是采用了类装载的机制来保证初始化实例时只有一个线程,
 * 与饿汉模式不同是:在饿汉模式中是类加载后就会初始化单例对象,无法延迟初始化对象,
 * 而静态内部类模式在类加载后并不会立即初始化,而是使用getInstance()时才会装载
 * 该类,从而实现单例对象的初始化。
 */
public class SingletonPattern06 {

    static class SingletonPatternHolder {
        private static final SingletonPattern06 instance = new SingletonPattern06();

        private SingletonPatternHolder() {
        }
    }

    private SingletonPattern06() {}

    public static SingletonPattern06 getInstance() {
        return SingletonPatternHolder.instance;
    }
}

package com.fly.singleton;

/**
* 单例模式写法7:基于枚举实现单例模式,具有线程安全、延迟加载、防止反序列化重新创建新的对象等优点。
*/
public enum SingletonPattern07 {
    INSTANCE;
}

package com.fly.masterworker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Master负责接收客户端提交的任务,然后通过阻塞队列进行存储。Master所拥有的线程
 * 作为阻塞队列的消费者,不断从阻塞队列中获取任务并轮流分给Worker执行
 * @param <T> 任务泛型
 * @param <R> 任务执行结果泛型
 */
public class Master<T extends Task, R> {
    // 所有Worker集合
    private HashMap<String, Worker<T, R>> workers = new HashMap<>();
    // 任务阻塞队列,用于存储任务
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    // 任务处理结果集合
    protected Map<String, R> resultMap = new ConcurrentHashMap<>();
    // Master的任务调度线程
    private Thread thread = null;
    // 保存最终的求和
    private AtomicLong sum = new AtomicLong(0);
    public Long getSum(){
        return sum.get();
    }

    public Master(int workerCount) {

        // 每个Worker对象都需要持有queue的引用,用于处理任务和提交结果
        for (int i = 0; i < workerCount; i++) {
            // 初始化workerCount个Worker实例
            Worker<T, R> worker = new Worker<>();
            workers.put("worker-" + i, worker);
        }
        thread = new Thread(() -> this.execute());
        thread.start();
    }

    // 提交任务,向任务队列添加任务
    public void submit(T task) {
        taskQueue.add(task);
    }

    // 获取Worker结果处理的回调函数
    private void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker-" + task.getId();
        // 获取执行结果
        R result = task.getResult();
        // 将执行结果存储在Map中
        resultMap.put(taskName, result);
        // 获取累加和
        sum.getAndAdd((Long) result);
    }

    // 执行所有子任务
    public void execute() {
        for (; ; ) {
            for (Map.Entry<String, Worker<T, R>> entry : workers.entrySet()) {
                T task = null;
                try {
                    // 从任务队列中获取任务,take()会阻塞后续代码执行
                    task = this.taskQueue.take();
                    // 获取Worker节点
                    Worker worker = entry.getValue();
                    // 分配任务
                    worker.submit(task, this::resultCallBack);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 打印最终结果
     */
    public void printResult() {
        for (Map.Entry<String, R> entry : resultMap.entrySet()) {
            String taskName = entry.getKey();
            System.out.println(taskName + ":" + entry.getValue());
        }
        System.out.println("sum is:" + sum.get());
    }
}

package com.fly.masterworker;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * Worker类用于接收Master分配的任务,同样也通过阻塞队列对局部任务进行缓存。Worker
 * 所拥有的线程作为局部任务的阻塞队列的消费者,不断的从阻塞队列中获取任务并执行,
 * 执行完成后执行Master传递过来的回调函数。
 * @param <T> 任务泛型
 * @param <R> 任务执行结果泛型
 */
public class Worker<T extends Task, R> {
    // 接收任务的阻塞队列
    private LinkedBlockingDeque<T> taskQueue = new LinkedBlockingDeque();
    // 通过AtomicInteger作为Worker编号计数器
    private AtomicInteger index = new AtomicInteger(1);
    // worker id
    private int workerId;
    // 执行任务的线程
    private Thread thread = null;

    public Worker() {
        this.workerId = index.getAndIncrement();
        thread = new Thread(() -> this.run());
        thread.start();
    }

    public void run() {
        // 不停向任务队列中获取任务并执行
        for (; ; ) {
            try {
                // 从阻塞队列中提取任务
                T task = this.taskQueue.take();
                // 设置worker id
                task.setWorkerId(workerId);
                // 执行任务
                task.execute();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void submit(T task, Consumer<R> action) {
        // 设置任务的回调函数
        task.resultAction = action;
        try {
            // 将任务数添加到任务队列中
            taskQueue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

package com.fly.masterworker;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * 异步任务类在子类任务的doExecute()方法之后,回调执行Master传递过来的回调函数。
 * @param <R> 任务执行结果泛型
 */
public class Task<R> {
    static AtomicInteger index = new AtomicInteger(1);
    // 任务的回调函数
    public Consumer<Task<R>> resultAction;
    // 任务id
    private int id;
    // Worker id,用于表示当前异步子任务属于那个Worker
    private int workerId;
    // 执行结果
    R result = null;

    public Task() {
        this.id = index.getAndIncrement();
    }

    public int getId() {
        return this.id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public int getWorkerId() {
        return workerId;
    }

    public void setWorkerId(int workerId) {
        this.workerId = workerId;
    }


    public R getResult() {
        return result;
    }

    public void setResult(R result) {
        this.result = result;
    }

    // 执行任务
    public void execute() {
        this.result = this.doExecute();
        // 执行回调函数
        resultAction.accept(this);
    }

    // 由子类实现
    protected R doExecute() {
        return null;
    }
}

package com.fly.masterworker;

import java.util.Random;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Master-Worker模式测试类
 */
public class MasterWorkerTest {
    private static Random random = new Random();

    static class SimpleTask extends Task<Long> {
        @Override
        protected Long doExecute() {
            // 生成两个0到10范围的整数
            Long i1 = (long) random.nextInt(10);
            Long i2 = (long) random.nextInt(10);
            System.out.println("求和表达式:i1+i2="+i1+"+"+i2);
            return i1 + i2;
        }
    }

    public static void main(String[] args) {
        // 创建Master实例并指定4个Worker处理任务
        Master<SimpleTask, Integer> master = new Master<>(4);
        // 创建可调度线程池
        ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(4);
        // 定期向master提交任务,每隔2s延迟0s向master提交任务
        pool.scheduleAtFixedRate(() -> master.submit(new SimpleTask()),
                0, 2000, TimeUnit.MILLISECONDS);

        // 定期从master获取结果,初始化延迟0s每隔5s从master获取执行结果
        pool.scheduleAtFixedRate(() -> {
            master.printResult();
        }, 0, 5000, TimeUnit.MILLISECONDS);
    }
}

package com.fly.forkjoin;

import java.util.concurrent.RecursiveTask;

/**
 * 自定义累加计算任务类
 */
public class AccumulateTask extends RecursiveTask<Integer> {
    // 任务拆分阈值
    private static final int THRESHOLD = 2;
    // 累加的起始编号
    private int start;
    // 累加的结束编号
    private int end;

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

    @Override
    protected Integer compute() {
        int sum = 0;
        // 判断任务的规模,对于小任务则直接计算返回
        boolean canCompute = (end - start) <= THRESHOLD;
        // 如果任务足够小则直接累加返回结果
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 如果任务足够大,则将任务一分为2
            int middle = (start + end) / 2;
            AccumulateTask lTask = new AccumulateTask(start,middle);
            AccumulateTask rTask = new AccumulateTask(middle+1,end);
            // 调用fork()执行子任务
            lTask.fork();
            rTask.fork();
            // 等待子任务执行完成,依次调用子任务的join()合并执行结果
            int leftResult = lTask.join();
            int rightResult = rTask.join();
            sum = leftResult + rightResult;
        }
        return sum;
    }
}

package com.fly.forkjoin;

import java.util.concurrent.*;

public class AccumulateTaskTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        // 创建ForkJoinPool线程池
        ForkJoinPool pool = new ForkJoinPool();
        AccumulateTask task =new AccumulateTask(1,100);
        // 向ForkJoin线程池提交任务
        Future<Integer> future = pool.submit(task);
        // 获取Future结果,如果1s内未获取结果则抛出TimeoutException
        Integer sum = future.get(2, TimeUnit.SECONDS);
        System.out.println("sum:"+sum); // sum:5050
    }
}

public ForkJoinPool(int parallelism,
                    ForkJoinWorkerThreadFactory factory,
                    UncaughtExceptionHandler handler,
                    boolean asyncMode) {
    this(checkParallelism(parallelism),
         checkFactory(factory),
         handler,
         asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
         "ForkJoinPool-" + nextPoolId() + "-worker-");
    checkPermission();
}

static final int MAX_CAP      = 0x7fff;        // 并行度常量 32767
public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}

private static ForkJoinPool makeCommonPool() {

    final ForkJoinWorkerThreadFactory commonPoolForkJoinWorkerThreadFactory =
            new CommonPoolForkJoinWorkerThreadFactory();
    int parallelism = -1;
    ForkJoinWorkerThreadFactory factory = null;
    UncaughtExceptionHandler handler = null;
    try {
        // 从系统变量中获取并行度
        String pp = System.getProperty
            ("java.util.concurrent.ForkJoinPool.common.parallelism");
        // 从系统变量中获取线程工厂
        String fp = System.getProperty
            ("java.util.concurrent.ForkJoinPool.common.threadFactory");
        // 从系统变量中获取异常处理程序
        String hp = System.getProperty
            ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
        if (pp != null)
            parallelism = Integer.parseInt(pp);
        if (fp != null)
            factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
                       getSystemClassLoader().loadClass(fp).newInstance());
        if (hp != null)
            handler = ((UncaughtExceptionHandler)ClassLoader.
                       getSystemClassLoader().loadClass(hp).newInstance());
    } catch (Exception ignore) {
    }
    if (factory == null) {
        if (System.getSecurityManager() == null)
            factory = commonPoolForkJoinWorkerThreadFactory;
        else // use security-managed default
            factory = new InnocuousForkJoinWorkerThreadFactory();
    }
    // parallelism默认为 cores - 1
    if (parallelism < 0 && // default 1 less than #cores
        (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
        parallelism = 1;
    if (parallelism > MAX_CAP)
        parallelism = MAX_CAP;
    return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
                            "ForkJoinPool.commonPool-worker-");
}

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","8")

-Djava.util.concurrent.ForkJoinPool.common.parallelism=8