码迷,mamicode.com
首页 > 其他好文 > 详细

ThreadPoolExecutor的分析(二)

时间:2017-11-19 14:18:44      阅读:174      评论:0      收藏:0      [点我收藏+]

标签:pool   obj   线程安全   解锁   www   date   无锁   this   exe   

说明:本作者是文章的原创作者,转载请注明出处:本文地址:http://www.cnblogs.com/qm-article/p/7859620.html

内部类Worker的分析

从源码可知。该内部类继承了AQS,且实现了runnable接口,说明,此类拥有锁的功能,且能充当线程使用,在前面的博文<线程的使用及ThreadPoolExecutor的分析(一)>中的addworker方法中当满足条件时会new Worker(firsttask),其内部发生了先是设置状态值为-1,其作用是为了在运行runWorker方法之前禁止中断,同时下一步构建了新的线程,并传入this参数

 Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

该类的其他方法作用如下

isHeldExclusively(),检查是否排他,即检查是否上了锁,若true则上锁状态,否则为不上锁状态

tryAcquire(int) ,该方法为AbstractQueuedSynchronizer,里的,该类重写了该方法,其作用是检查当前线程是尝试获取锁,通过cas操作来保证原子性,若返回true,则说明处于无锁状态,随机设置当前线程为此锁的主人

setExclusiveOwnerThread(Thread.currentThread()),设置当前线程为此锁的拥有者

tryRelease(int) ,尝试释放锁,其内部将此锁的拥有者设为null,即释放了锁

lock() ,加锁

tryLock(),尝试加锁

unlock()  解锁

isLocked(),返回是否已经被其他线程获取了锁

interruptIfStarted() ,若开始了,则中断线程

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

  其中getState()>=0,在开始的构造方法中设置state的值为-1,就是防止线程被中断,

在这里分析下tryAcquire(int)中的cas操作

点进去看其源码

protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

  其属于AQS中的方法,其中this,代表当前调用该方法的类的实例,在这里代表worker的实例,其中stateOffset代表要被更新的值,expect,代表内存中的旧值,update代表被更新后的值,

cas大致原理就是若内存中的值,与即将被更新的值一样,则用update取代该值,否则不取代,举个例子,如数据库表中,有一个version,和一个name字段,在开始时,你查询出了这个表的所有字段

之后,version的值已经存在内存中了,你突然要改变name值,此时,你也怕在这个期间有其他线程也更改了该name字段的值,此时你只要查询一次该表,若此时查询出来的version值,与已经在内存

中的version值一样,则更新name字段的值,否则不更新,cas操作也常用于乐观锁的实现。

其中stateOffset的值

       stateOffset = unsafe.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("state"));

  该段代码的作用获取AbstractQueuedSynchronizer类中state字段的值,再赋值给stateOffset,该操作时线程安全的,private volatile int state;这是state的定义,其默认值为0,

protected final void setState(int newState) {
        state = newState;
    }

也可以通过以上set方法来进行重新赋值

 

线程启动的入口

在addWorker的t.start()中

   if (workerAdded) {
        t.start();
        workerStarted = true;
   }

 

表明一启动该thread,则会调用该复写的run方法

 public void run() {
            runWorker(this);
  }

 

 其中runworker()源码如下

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 {
                        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);
        }
    }

该方法大致作用是遍历队列,执行线程任务

先介绍该方法内的其他方法作用

getTask(),其作用是从阻塞队列中取出线程任务

runStateAtLeast(int,int),判断线程池至少处于什么状态,前面博文中,有写线程池的几个状态及其大小
beforeExecute(Thread t, Runnable r),提供给ThreadPoolExecutor子类去实现
afterExecute(Runnable r, Throwable t),也是提供给子类去实现
processWorkerExit(Worker w, boolean completedAbruptly) 主要是为了清除和记录以及死亡的worker线程

在runWorker方法中,通过while循环,去取当前worker中的线程任务firstWorker,和阻塞队列中的线程任务
第一个if中的判断是为了检测线程池的状态及线程是否中断了,如果是,则中断当前线程
try里面的run方法是执行获取到的线程的任务,将完成的线程任务进行累加

ThreadPoolExecutor的分析(二)

标签:pool   obj   线程安全   解锁   www   date   无锁   this   exe   

原文地址:http://www.cnblogs.com/qm-article/p/7859620.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!