/**
 * 拒絕策略
 */
@FunctionalInterface
interface RejectPolicy<T>{
	//queue就是我們自己實(shí)現(xiàn)的阻塞隊(duì)列，task是任務(wù)
    void reject(BlockingQueue<T> queue,T task);
}

/**
 * 阻塞隊(duì)列
 */
class BlockingQueue<T>{
    //阻塞隊(duì)列
    private Deque<T> queue = new ArrayDeque<>();

    //鎖
    private ReentrantLock lock = new ReentrantLock();

    //生產(chǎn)者條件變量
    private Condition fullWaitSet = lock.newCondition();

    //消費(fèi)者條件變量
    private Condition emptyWaitSet = lock.newCondition();

    //容量
    private int capacity;

    public BlockingQueue(int capacity){
        this.capacity = capacity;
    }

    //帶有超時(shí)阻塞獲取
    public T poll(long timeout, TimeUnit timeUnit){
        lock.lock();
        try {
            //將timeout統(tǒng)一轉(zhuǎn)換為納秒
            long nanos = timeUnit.toNanos(timeout);
            while(queue.isEmpty()){
                try {
                    if(nanos <= 0){
                        //小于0，說(shuō)明上次沒(méi)有獲取到，代表已經(jīng)超時(shí)了
                        return null;
                    }
                    //返回值是剩余的時(shí)間
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            //通知生產(chǎn)者
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞獲取
    public T take(){
        lock.lock();
        try{
            while(queue.isEmpty()){ //如果任務(wù)隊(duì)列為空，代表線程池沒(méi)有可以執(zhí)行的內(nèi)容
                try {
                     /*
                    也就說(shuō)此時(shí)進(jìn)來(lái)的線程是執(zhí)行不了任務(wù)的，所以此時(shí)emptyWaitSet消費(fèi)者要進(jìn)行阻塞狀態(tài)
                    等待下一次喚醒，然后繼續(xù)判斷隊(duì)列是否為空
                     */
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            /*
            代碼執(zhí)行到這里。說(shuō)明任務(wù)隊(duì)列不為空，線程池就從任務(wù)隊(duì)列拿出一個(gè)任務(wù)出來(lái)執(zhí)行
            也就是說(shuō)把阻塞隊(duì)列的一個(gè)任務(wù)出隊(duì)
             */
            T t = queue.removeFirst();
            /*
            然后喚醒之前存放在生成者Condition休息室，因?yàn)橛捎谥白枞?duì)列已滿(mǎn)，fullWaitSet才會(huì)進(jìn)入阻塞狀態(tài)
            所以當(dāng)阻塞隊(duì)列刪除了任務(wù)，就要喚醒之前進(jìn)入阻塞狀態(tài)的fullWaitSet
             */
            fullWaitSet.signal();
            //返回任務(wù)
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T task){
        lock.lock();
        try {
            while(queue.size() == capacity){    //任務(wù)隊(duì)列滿(mǎn)了
                try {
                    System.out.println("等待加入任務(wù)隊(duì)列"+task);
                    /*
                    也就說(shuō)此時(shí)進(jìn)來(lái)的任務(wù)是進(jìn)不了阻塞隊(duì)列的，已經(jīng)滿(mǎn)了，所以此時(shí)生產(chǎn)者Condition要進(jìn)入阻塞狀態(tài)
                    等待下一次喚醒，然后繼續(xù)判斷隊(duì)列是否為空
                     */
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //任務(wù)隊(duì)列還未滿(mǎn)
            System.out.println("加入任務(wù)隊(duì)列"+task);
            //把任務(wù)加入阻塞隊(duì)列
            queue.addLast(task);
            /*
            然后喚醒之前存放在消費(fèi)者Condition休息室，因?yàn)橛捎谥白枞?duì)列為空，emptyWaitSet才會(huì)進(jìn)入阻塞狀態(tài)
            所以當(dāng)阻塞隊(duì)列加入了任務(wù)，就要喚醒之前進(jìn)入阻塞狀態(tài)的emptyWaitSet
             */
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //帶超時(shí)阻塞時(shí)間添加
    public boolean offer(T task,long timeout,TimeUnit timeUnit){
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while(queue.size() == capacity){
                try {
                    if(nanos < 0){
                        return false;
                    }
                    System.out.println("等待加入任務(wù)隊(duì)列"+task);
                    //不會(huì)一直阻塞，超時(shí)就會(huì)繼續(xù)向下執(zhí)行
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("加入任務(wù)隊(duì)列"+task);
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    //獲取任務(wù)數(shù)量
    public int size(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

    //嘗試添加任務(wù)，如果阻塞隊(duì)列已經(jīng)滿(mǎn)了，就使用拒絕策略
    public void tryPut(RejectPolicy<T> rejectPolicy, T task){
        lock.lock();
        try {
            //判斷隊(duì)列是否已滿(mǎn)
            if(queue.size() == capacity){
                rejectPolicy.reject(this,task);
            }else{  //有空閑
                System.out.println("加入任務(wù)隊(duì)列"+task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}

/**
 * 線程池
 */
class ThreadPool{
    //阻塞隊(duì)列
    private BlockingQueue<Runnable> taskQueue;

    //線程集合
    private HashSet<Worker> workers = new HashSet<>();

    //核心線程數(shù)
    private int coreSize;

    //獲取任務(wù)的超時(shí)時(shí)間
    private long timeout;

    private TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }

    //執(zhí)行任務(wù)
    public void execute(Runnable task){
        synchronized (workers){
            if(workers.size() <= coreSize){  //當(dāng)前的線程數(shù)小于核心線程數(shù)
                Worker worker = new Worker(task);
                workers.add(worker);
                //讓線程開(kāi)始工作，執(zhí)行它的run方法
                worker.start();
            }else{
                // 1) 死等
                // 2) 帶超時(shí)等待
                // 3) 讓調(diào)用者放棄任務(wù)執(zhí)行
                // 4) 讓調(diào)用者拋出異常
                // 5) 讓調(diào)用者自己執(zhí)行任務(wù)
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }

    /**
     * 工作線程，也就是線程池里面的線程
     */
    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            //執(zhí)行任務(wù)
            // 1) 當(dāng) task 不為空，執(zhí)行任務(wù)
            // 2) 當(dāng) task 執(zhí)行完畢，再接著從任務(wù)隊(duì)列獲取任務(wù)并執(zhí)行
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    System.out.println("正在執(zhí)行的任務(wù)" + task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //代表這個(gè)任務(wù)已經(jīng)執(zhí)行完了
                    task = null;
                }
            }
            synchronized (workers) {
                System.out.println("worker 被移除" + this);
                workers.remove(this);
            }
        }
    }
}