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

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

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

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

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

    //容量
    private int capacity;

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

    //帶有超時阻塞獲取
    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，說明上次沒有獲取到，代表已經(jīng)超時了
                        return null;
                    }
                    //返回值是剩余的時間
                    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ù)隊列為空，代表線程池沒有可以執(zhí)行的內(nèi)容
                try {
                     /*
                    也就說此時進來的線程是執(zhí)行不了任務(wù)的，所以此時emptyWaitSet消費者要進行阻塞狀態(tài)
                    等待下一次喚醒，然后繼續(xù)判斷隊列是否為空
                     */
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            /*
            代碼執(zhí)行到這里。說明任務(wù)隊列不為空，線程池就從任務(wù)隊列拿出一個任務(wù)出來執(zhí)行
            也就是說把阻塞隊列的一個任務(wù)出隊
             */
            T t = queue.removeFirst();
            /*
            然后喚醒之前存放在生成者Condition休息室，因為由于之前阻塞隊列已滿，fullWaitSet才會進入阻塞狀態(tài)
            所以當阻塞隊列刪除了任務(wù)，就要喚醒之前進入阻塞狀態(tài)的fullWaitSet
             */
            fullWaitSet.signal();
            //返回任務(wù)
            return t;
        }finally {
            lock.unlock();
        }
    }

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

    //帶超時阻塞時間添加
    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ù)隊列"+task);
                    //不會一直阻塞，超時就會繼續(xù)向下執(zhí)行
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("加入任務(wù)隊列"+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ù)，如果阻塞隊列已經(jīng)滿了，就使用拒絕策略
    public void tryPut(RejectPolicy<T> rejectPolicy, T task){
        lock.lock();
        try {
            //判斷隊列是否已滿
            if(queue.size() == capacity){
                rejectPolicy.reject(this,task);
            }else{  //有空閑
                System.out.println("加入任務(wù)隊列"+task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}

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

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

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

    //獲取任務(wù)的超時時間
    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){  //當前的線程數(shù)小于核心線程數(shù)
                Worker worker = new Worker(task);
                workers.add(worker);
                //讓線程開始工作，執(zhí)行它的run方法
                worker.start();
            }else{
                // 1) 死等
                // 2) 帶超時等待
                // 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) 當 task 不為空，執(zhí)行任務(wù)
            // 2) 當 task 執(zhí)行完畢，再接著從任務(wù)隊列獲取任務(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 {
                    //代表這個任務(wù)已經(jīng)執(zhí)行完了
                    task = null;
                }
            }
            synchronized (workers) {
                System.out.println("worker 被移除" + this);
                workers.remove(this);
            }
        }
    }
}