`

从Java源码看java线程池-三句话说清楚

 
阅读更多
简单的说就是,我预计最少需要3个(corePoolSize)正事员工(线程),如果事情做不过来,我可以扩招到10个(maximumPoolSize),再多我财务承受不起,只能让在职的人加班,如果还是做不过来,作为老板的我就得定个策略:是帮他们排期呢?还是直接拒绝呢。。。员工呢,签定合同的周期为keepAliveTime ,单位可以是TimeUnit;更有甚着,员工必须是

ThreadFactory 这个地方出来的。





以上是线程池的核心思想,接下来就要深入代码了:

//首先是员工:

//这是员工列表
 private final HashSet<Worker> workers = new HashSet<Worker>();




//这是员工的基本定义





 private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;

        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        /** Delegates main run loop to outer runWorker  */
        public void run() {

                  //执行任务
            runWorker(this);
        }

        // Lock methods
        //
        // The value 0 represents the unlocked state.
        // The value 1 represents the locked state.

        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    } 




//这是添加员工的方法


private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();

                       //
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {

                //这点是在容易漏掉 启动一个线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }





//接下来是任务:


//其实任务队列是个阻塞队列


  private final BlockingQueue<Runnable> workQueue;



//然后是获取任务的方法:





 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 {

           //获取任务
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }


 //执行任务:


 final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            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 {

                       //容易搞混的点,thread中start()是开启一个线程,而这里仅仅只是调用run方法。
                        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);
        }
    }

//最后可能少一个大家经常看到的入口
 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

 

 
分享到:
评论

相关推荐

    基于java的开发源码-github-Java-api.zip

    基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于...

    基于java的开发源码-P2P-多用户在线聊天室(Java源码).zip

    基于java的开发源码-P2P--多用户在线聊天室(Java源码).zip 基于java的开发源码-P2P--多用户在线聊天室(Java源码).zip 基于java的开发源码-P2P--多用户在线聊天室(Java源码).zip 基于java的开发源码-P2P--多...

    java源码包---java 源码 大量 实例

     Java 3DMenu 界面源码,有人说用到游戏中不错,其实平时我信编写Java应用程序时候也能用到吧,不一定非要局限于游戏吧,RES、SRC资源都有,都在压缩包内。 Java zip压缩包查看程序源码 1个目标文件 摘要:Java源码...

    Java课程设计----单词簿----源码和报告-----高分课设

    Java课程设计----课题:单词簿 内附源码和课设报告,报告一共27页。 课题3 单词簿 设计GUI界面的单词簿。具体要求如下: 1)使用内置Derby数据库。在数据库中使用表存储单词和该单词的翻译解释,例如“sun,太阳”...

    java 游戏源码-----模拟钢琴

    "Java 游戏源码-----模拟钢琴"是一个项目,它提供了一种独特的学习体验,让你能够深入理解如何使用Java语言来创建互动性的游戏应用。这个源码实现了一个模拟钢琴,允许用户在虚拟键盘上弹奏音乐,体验接近真实钢琴的...

    线程池管理源码 java 源码

    线程池是Java多线程编程中的重要概念,它的出现是为了高效、有序地管理线程资源,避免频繁创建和销毁线程导致的性能开销。本文将深入解析线程池的管理源码,帮助读者理解其工作原理和优化策略。 在Java中,`java....

    taobao-sdk-java-auto源码

    taobao-sdk-java-auto源码, taobao-sdk-java-auto源码, taobao-sdk-java-auto源码, taobao-sdk-java-auto源码

    java实战项目-基于JavaSSM实现在线教育平台系统源码(含教程)-java毕设

    java实战项目-基于JavaSSM实现在线教育平台系统源码(含教程)-java毕设 技术栈: Java Spring SpringMVC MyBatis 基于Java+Spring+SpringMVC+MyBatis实现的在线教育平台系统源码,包含项目的开发教程,可以作为计算机...

    java线程池完整代码

    描述 "Java 线程池完整源码" 说明了这篇文章的内容是关于 Java 线程池的完整实现代码,包括线程池的源代码、配置文件的解析和线程池的管理等方面。 标签解析 标签 "Java 线程池完整源码" 说明了这篇文章的主题是...

    Java 多线程与并发(17-26)-JUC线程池- FutureTask详解.pdf

    ### Java多线程与并发(17-26)-JUC线程池-FutureTask详解 #### 一、概述 本文将围绕Java多线程与并发中的重要概念——`FutureTask`进行深入探讨。`FutureTask`是Java并发库中的一个关键组件,它实现了`...

    基于java的开发源码-Message-Driven Bean EJB实例源代码.zip

    基于java的开发源码-Message-Driven Bean EJB实例源代码.zip 基于java的开发源码-Message-Driven Bean EJB实例源代码.zip 基于java的开发源码-Message-Driven Bean EJB实例源代码.zip 基于java的开发源码-Message-...

    mysql-connector-java-5.1.37jar包和源码

    在本案例中,"mysql-connector-java-5.1.37.jar" 是一个特定版本的MySQL JDBC驱动程序,发布于2016年,属于较旧但仍然广泛使用的版本。这个jar包是Java开发者用来连接到MySQL服务器的关键组件。 首先,我们需要了解...

    线程池 -- 链表

    NULL 博文链接:https://shaojiashuai123456.iteye.com/blog/1415886

    成百上千个Java 源码DEMO 4(1-4是独立压缩包)

    Java 3DMenu 界面源码 5个目标文件 内容索引:Java源码,窗体界面,3DMenu Java 3DMenu 界面源码,有人说用到游戏中不错,其实平时我信编写Java应用程序时候也能用到吧,不一定非要局限于游戏吧,RES、SRC资源都有,都...

    JAVA源码密钥管理工具Keytool-IUI

    JAVA源码密钥管理工具Keytool-IUI

    基于Java的源码-Javashop-eop Java开发框架.zip

    总的来说,Javashop-eop Java开发框架是一个强大的工具,它为Java开发者提供了便捷的开发环境和丰富的功能支持。通过深入学习和实践,开发者可以借助Javashop-eop快速构建出稳定、高效的Web应用,满足各种业务需求。...

    纯java写的课程设计-java通讯录管理系统源码.zip

    纯java写的课程设计----java通讯录管理系统源码.zip纯java写的课程设计----java通讯录管理系统源码.zip纯java写的课程设计----java通讯录管理系统源码.zip纯java写的课程设计----java通讯录管理系统源码.zip纯java写...

    基于Java的源码-github-Java-api.zip

    基于Java的源码-github-Java-api.zip

    C++实现线程池详解(基于boost源码以及封装等线程池)

    五、看看人家线程池怎么写的 - 要理解精髓 六、线程池应用场景与实践 6.1 服务器应用 6.2 数据处理与计算密集型任务 6.3 图形界面与事件驱动程序 七、C++线程池高级应用与实际案例 7.1 基于负载均衡的任务分配策略 ...

    Java实现的线程池、消息队列功能

    标题中的“Java实现的...总的来说,这个主题涵盖了Java并发编程中的核心概念,线程池和消息队列的实现原理及应用场景,以及可能用到的相关开发工具。理解并熟练运用这些知识对于提升Java应用的性能和稳定性至关重要。

Global site tag (gtag.js) - Google Analytics