java線程池趣味事：這不是線程池

// 創(chuàng)建固定數(shù)量的線程池Executors.newFixedThreadPool(8);// 創(chuàng)建無限動態(tài)創(chuàng)建的線程池Executors.newCachedThreadPool();// 創(chuàng)建定時調(diào)度線程池Executors.newScheduledThreadPool(2);// 還有個創(chuàng)建單線程的就不說了，都一樣

// 自定義各線程參數(shù)ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 20, 20, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

    // 初始化線程池    private ExecutorService executor            = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),                Runtime.getRuntime().availableProcessors(),                0L, TimeUnit.SECONDS,                new ArrayBlockingQueue<>(50),                new NamedThreadFactory("test-pool"),                new ThreadPoolExecutor.CallerRunsPolicy());    // 使用線程池處理任務    public Integer doTask(String updateIntervalDesc) throws Exception {        long startTime = System.currentTimeMillis();        List<TestDto> testList;        AtomicInteger affectNum = new AtomicInteger(0);        int pageSize = 1000;        AtomicInteger pageNo = new AtomicInteger(1);        Map<String, Object> condGroupLabel = new HashMap<>();        log.info("start do sth:{}", updateIntervalDesc);        List<Future<?>> futureList = new ArrayList<>();        do {            PageHelper.startPage(pageNo.getAndIncrement(), pageSize);            List<TestDto> list                    = testDao.getLabelListNew(condGroupLabel);            testList = list;            // 循環(huán)向線程池中提交任務            for (TestDto s : list) {                Future<?> future = executor.submit(() -> {                    try {                        // do sth...                        affectNum.incrementAndGet();                    }                    catch (Throwable e) {                        log.error("error:{}", pageNo.get(), e);                    }                });                futureList.add(future);            }        } while (testList.size() >= pageSize);        // 等待任務完成        int i = 0;        for (Future<?> future : futureList) {            future.get();            log.info("done:+{} ", i++);        }        log.info("doTask done:{}, num:{}, cost:{}ms",                updateIntervalDesc, affectNum.get(), System.currentTimeMillis() - startTime);        return affectNum.get();    }

    // java.util.concurrent.ThreadPoolExecutor#runWorker    /**     * Main worker run loop.  Repeatedly gets tasks from queue and     * executes them, while coping with a number of issues:     *     * 1. We may start out with an initial task, in which case we     * don't need to get the first one. Otherwise, as long as pool is     * running, we get tasks from getTask. If it returns null then the     * worker exits due to changed pool state or configuration     * parameters.  Other exits result from exception throws in     * external code, in which case completedAbruptly holds, which     * usually leads processWorkerExit to replace this thread.     *     * 2. Before running any task, the lock is acquired to prevent     * other pool interrupts while the task is executing, and then we     * ensure that unless pool is stopping, this thread does not have     * its interrupt set.     *     * 3. Each task run is preceded by a call to beforeExecute, which     * might throw an exception, in which case we cause thread to die     * (breaking loop with completedAbruptly true) without processing     * the task.     *     * 4. Assuming beforeExecute completes normally, we run the task,     * gathering any of its thrown exceptions to send to afterExecute.     * We separately handle RuntimeException, Error (both of which the     * specs guarantee that we trap) and arbitrary Throwables.     * Because we cannot rethrow Throwables within Runnable.run, we     * wrap them within Errors on the way out (to the thread's     * UncaughtExceptionHandler).  Any thrown exception also     * conservatively causes thread to die.     *     * 5. After task.run completes, we call afterExecute, which may     * also throw an exception, which will also cause thread to     * die. According to JLS Sec 14.20, this exception is the one that     * will be in effect even if task.run throws.     *     * The net effect of the exception mechanics is that afterExecute     * and the thread's UncaughtExceptionHandler have as accurate     * information as we can provide about any problems encountered by     * user code.     *     * @param w the worker     */    final void runWorker(Worker w) {        Thread wt = Thread.currentThread();        Runnable task = w.firstTask;        w.firstTask = null;        w.unlock(); // allow interrupts        boolean completedAbruptly = true;        try {            // worker 不停地向隊列中獲取任務，然后執(zhí)行            // 其中獲取任務的過程，可能被中斷，也可能不會，受到線程池伸縮配置的影響            while (task != null || (task = getTask()) != null) {                w.lock();                // If pool is stopping, ensure thread is interrupted;                // if not, ensure thread is not interrupted.  This                // requires a recheck in second case to deal with                // shutdownNow race while clearing interrupt                if ((runStateAtLeast(ctl.get(), STOP) ||                     (Thread.interrupted() &&                      runStateAtLeast(ctl.get(), STOP))) &&                    !wt.isInterrupted())                    wt.interrupt();                try {                    beforeExecute(wt, task);                    Throwable thrown = null;                    try {                        task.run();                    } catch (RuntimeException x) {                        thrown = x; throw x;                    } catch (Error x) {                        thrown = x; throw x;                    } catch (Throwable x) {                        thrown = x; throw new Error(x);                    } finally {                        afterExecute(task, thrown);                    }                } finally {                    task = null;                    w.completedTasks++;                    w.unlock();                }            }            completedAbruptly = false;        } finally {            processWorkerExit(w, completedAbruptly);        }    }    /**     * Performs blocking or timed wait for a task, depending on     * current configuration settings, or returns null if this worker     * must exit because of any of:     * 1. There are more than maximumPoolSize workers (due to     *    a call to setMaximumPoolSize).     * 2. The pool is stopped.     * 3. The pool is shutdown and the queue is empty.     * 4. This worker timed out waiting for a task, and timed-out     *    workers are subject to termination (that is,     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})     *    both before and after the timed wait, and if the queue is     *    non-empty, this worker is not the last thread in the pool.     *     * @return task, or null if the worker must exit, in which case     *         workerCount is decremented     */    private Runnable getTask() {        boolean timedOut = false; // Did the last poll() time out?
        for (;;) {            int c = ctl.get();            int rs = runStateOf(c);
            // Check if queue empty only if necessary.            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {                decrementWorkerCount();                return null;            }
            int wc = workerCountOf(c);
            // Are workers subject to culling?            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            if ((wc > maximumPoolSize || (timed && timedOut))                && (wc > 1 || workQueue.isEmpty())) {                if (compareAndDecrementWorkerCount(c))                    return null;                continue;            }
            try {                // 可能調(diào)用超時方法，也可能調(diào)用阻塞方法                // 固定線程池的情況下，調(diào)用阻塞 take() 方法                Runnable r = timed ?                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                    workQueue.take();                if (r != null)                    return r;                timedOut = true;            } catch (InterruptedException retry) {                timedOut = false;            }        }    }

    // java.util.concurrent.ArrayBlockingQueue#take    public E take() throws InterruptedException {        final ReentrantLock lock = this.lock;        // 此處鎖，保證執(zhí)行線程安全性        lock.lockInterruptibly();        try {            while (count == 0)                // 此處釋放鎖等待，再次喚醒時，要求必須重新持有鎖                notEmpty.await();            return dequeue();        } finally {            lock.unlock();        }    }    //     /**     * Inserts the specified element at the tail of this queue, waiting     * for space to become available if the queue is full.     *     * @throws InterruptedException {@inheritDoc}     * @throws NullPointerException {@inheritDoc}     */    public void put(E e) throws InterruptedException {        checkNotNull(e);        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        try {            while (count == items.length)                notFull.await();            enqueue(e);        } finally {            lock.unlock();        }    }    /**     * Inserts element at current put position, advances, and signals.     * Call only when holding lock.     */    private void enqueue(E x) {        // assert lock.getHoldCount() == 1;        // assert items[putIndex] == null;        final Object[] items = this.items;        items[putIndex] = x;        if (++putIndex == items.length)            putIndex = 0;        count++;        // 通知取等線程，喚醒        notEmpty.signal();    }

    // 初始化線程池    private ExecutorService executor            = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),                Runtime.getRuntime().availableProcessors(),                0L, TimeUnit.SECONDS,                new ArrayBlockingQueue<>(50),                new NamedThreadFactory("test-pool"),                new ThreadPoolExecutor.CallerRunsPolicy());    // 使用線程池處理任務    public Integer doTask(String updateIntervalDesc) throws Exception {        long startTime = System.currentTimeMillis();        List<TestDto> testList;        AtomicInteger affectNum = new AtomicInteger(0);        int pageSize = 1000;        AtomicInteger pageNo = new AtomicInteger(1);        Map<String, Object> condGroupLabel = new HashMap<>();        log.info("start do sth:{}", updateIntervalDesc);        List<Future<?>> futureList = new ArrayList<>();        do {            PageHelper.startPage(pageNo.getAndIncrement(), pageSize);            List<TestDto> list                    = testDao.getLabelListNew(condGroupLabel);            testList = list;            // 一批任務只向線程池中提交任務            Future<?> future = executor.submit(() -> {                for (TestDto s : list) {                    try {                        // do sth...                        affectNum.incrementAndGet();                    }                    catch (Throwable e) {                        log.error("error:{}", pageNo.get(), e);                    }                }            });            futureList.add(future);        } while (testList.size() >= pageSize);        // 等待任務完成        int i = 0;        for (Future<?> future : futureList) {            future.get();            log.info("done:+{} ", i++);        }        log.info("doTask done:{}, num:{}, cost:{}ms",                updateIntervalDesc, affectNum.get(), System.currentTimeMillis() - startTime);        return affectNum.get();    }