高并发设计模式-07

public class Singleton1 {

    // 私有构造区
    private Singleton1() {
    }

    // 静态成员
    private static final Singleton1 single = new Singleton1();

    public static Singleton1 getInstance() {
        return single;
    }
}

/**
 * 简单的懒汉单例模式
 *
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class ASingleton {

    // 静态成员
    static ASingleton instance;

    // 私有构造区
    private ASingleton() {
    }

    // 获取单例的方法
    static ASingleton getInstance() {
        if (instance == null) {
            instance = new ASingleton();
        }
        return instance;
    }
}

/**
 * 简单的懒汉单例模式
 * 使用内置锁 synchronized 进行单例获取同步
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class BSingleton {

    // 静态成员
    static BSingleton instance;

    // 私有构造区
    private BSingleton() {
    }

    // 获取单例的方法
    static synchronized BSingleton getInstance() {
        if (instance == null) {
            instance = new BSingleton();
        }
        return instance;
    }
}

/**
 * 简单的懒汉单例模式
 * 使用内置锁 synchronized 进行单例获取同步
 *
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class ESingleton {

    // 静态成员
    static ESingleton instance;

    // 私有构造区
    private ESingleton() {
    }

    // 获取单例的方法
    static synchronized ESingleton getInstance() {
        if (instance == null) { // 1。检查
            synchronized (ESingleton.class) {    // 加锁
                if (instance == null) {      // 2。检查
                    instance = new ESingleton();
                }
            }

        }
        return instance;
    }
}

//初始化单例
instance = new Singleton();

/**
 * 双重检查锁 + volatile相结合的单例模式的实现
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class FSingleton {

    // 静态成员
    // 保持单例的静态成员具有内存可见性
    static volatile FSingleton instance;

    // 私有构造区
    private FSingleton() {
    }

    // 获取单例的方法
    static synchronized FSingleton getInstance() {
        if (instance == null) { // 1。检查
            synchronized (FSingleton.class) {    // 加锁
                if (instance == null) {      // 2。检查
                    instance = new FSingleton();
                }
            }

        }
        return instance;
    }
}

/**
 * 使用静态内部类实现懒汉式单例模式
 *
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class Singleton {

    // 静态内部类
    private static class LazyHolder {
        // 通过final 保障初始化时的线程安全
        public static final Singleton INSTANCE = new Singleton();
        ;
    }

    // 私有构造区
    private Singleton() {
    }

    //获取单例的方法
    public static final Singleton getInstance() {
        // 返回内部类的静态，最终成员
        return LazyHolder.INSTANCE;
    }
}

/**
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class Master<T extends Task, R> {
    // 所有work的集合
    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 AtomicInteger sum = new AtomicInteger(0);

    public Master(int workerCount) {
        // 每个worker 对象都需要持有queue的引用，用于领任务与提交任务
        for (int i = 0; i < workerCount; i++) {
            Worker<T, R> worker = new Worker<>();
            workers.put("子节点：" + i, worker);
        }
        thread = new Thread(() -> this.execute());
        thread.start();
    }

    // 提交任务
    public void submit(T task) {
        taskQueue.add(task);
    }

    // 启动所有的子任务
    public void execute() {
        for (; ; ) {
            // 从任务队列中获取任务，然后worker节点轮询，轮流分配任务
            for (Map.Entry<String, Worker<T, R>> entry : workers.entrySet()) {
                T task = null;
                try {
                    task = this.taskQueue.take();
                    Worker worker = entry.getValue();
                    worker.submit(task, this::resultCallBack);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // 结果处理的回调函数
    private void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker:" + task.getWorkId() + "-" + "Task:" + task.getId();
        //        Print.tco(taskName + ":" + task.getResult());
        R result = task.getResult();
        resultMap.put(taskName, result);

        sum.getAndAdd((Integer) result);
    }

    // 获取最终的结果
    public void printResult() {
        Print.tco("----------sum is :" + sum.get());
        for (Map.Entry<String, R> entry : resultMap.entrySet()) {
            String taskName = entry.getKey();
            Print.fo(taskName + ":" + entry.getValue());
        }
    }
}

/**
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class Worker<T extends Task, R> {

    // 接收任务的阻塞队列
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    // worker 的编号
    static AtomicInteger index = new AtomicInteger(1);
    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();
                task.setWorkId(workerId);
                task.execute();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    // 接收任务到异步队列
    public void submit(T task, Consumer<R> action) {
        task.resultAction = action;
        try {
            this.taskQueue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

/**
 * 异步任务类在执行子类任务的doExecute()方法之后，回调一下Master传递过来的回调函数，将执行完成后的任务进行回填。
 *
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
@Data
public class Task<R> {

    static AtomicInteger index = new AtomicInteger(1);
    // 任务的回调函数
    public Consumer<Task<R>> resultAction;
    // 任务的id
    private int id;

    // worker ID
    private int workId;

    // 计算结果
    R result = null;

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

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

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

/**
 * @author: wangyj
 * @create: 2022-04-21
 * @version: 1.0.0
 **/
public class MasterWorkerTest {

    // 简单任务
    static class SimpleTask extends Task<Integer> {

        @Override
        protected Integer doExecute() {
            Print.tcfo("task " + getId() + " is done ");
            return getId();
        }
    }

    public static void main(String[] args) {
        //创建Master ，包含四个worker，并启动master的执行线程
        Master<SimpleTask, Integer> master = new Master<>(4);

        //定期向master提交任务
        ThreadUtil.scheduleAtFixedRate(() -> master.submit(
                new SimpleTask()),
                2, TimeUnit.SECONDS);

        //定期从master提取结果
        ThreadUtil.scheduleAtFixedRate(
                () -> master.printResult(),
                5, TimeUnit.SECONDS);
    }
}

public class AccumulateTask extends RecursiveTask<Integer> {

    private static final int THRESHOLD = 10;
    //累加的起始编号
    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) {
            //直接计算并返回结果,Recursive结束
            for (int i = start; i <= end; i++) {
                sum += i;
            }
            Print.tco("执行任务，计算" + start + "到" + end + "的和，结果是：" + sum);
        } else {
            //任务过大，需要切割,Recursive 递归计算
            Print.tco("切割任务：将" + start + "到" + end + "的和一分为二");
            int middle = (start + end) / 2;
            //切割成两个子任务
            AccumulateTask lTask = new AccumulateTask(start, middle);
            AccumulateTask rTask = new AccumulateTask(middle + 1, end);
            // 依次调用每个子任务的fork方法执行子任务
            lTask.fork();
            rTask.fork();
            //等待子任务的完成，依次调用每个子任务的join方法合并执行结果
            Integer leftResult = lTask.join();
            Integer rightResult = rTask.join();
            //合并子任务执行结果
            sum = leftResult + rightResult;
        }
        return sum;
    }
}

public class ForkJoinTest {

    @Test
    public void testAccumulateTask() throws ExecutionException, InterruptedException, TimeoutException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //创建一个累加任务，计算 由1加到10
        AccumulateTask countTask = new AccumulateTask(1, 100);
        ForkJoinTask<Integer> future = forkJoinPool.submit(countTask);
        Integer sum = future.get(1, TimeUnit.SECONDS);
        Print.tcfo("最终的计算结果：" + sum);
        //预期的结果为5050
        Assert.assertTrue(sum == 5050);
    }
}