terminated 线程_Java进阶专题(十三) 从电商系统角度研究多线程(上)

//shutdown后不接受新任务，但是task1，仍然可以执行完成
ExecutorService poolExecutor = Executors.newFixedThreadPool(5);
poolExecutor.execute(new Runnable() {
   public void run() {
       try {
           Thread.sleep(1000);
           System.out.println("finish task 1");
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }
});
poolExecutor.shutdown();
poolExecutor.execute(new Runnable() {
   public void run() {
       try {
           Thread.sleep(1000);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }
});
System.out.println("ok");
 
 
 

//改为shutdownNow后，任务立马终止，sleep被打断，新任务无法提交，task1停止
poolExecutor.shutdownNow();
 
 
 

//自定义类，重写terminated方法
public class MyExecutorService extends ThreadPoolExecutor {
   public MyExecutorService(int corePoolSize, int maximumPoolSize, long
keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
       super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
workQueue);
   }
   @Override
   protected void terminated() {
       super.terminated();
       System.out.println("treminated");
   }
   
   //调用 shutdownNow， ternimated方法被调用打印
   public static void main(String[] args) throws InterruptedException {
       MyExecutorService service = new
MyExecutorService(1,2,10000,TimeUnit.SECONDS,new
LinkedBlockingQueue<Runnable>(5));
       service.shutdownNow();
   }
}

//任务提交阶段
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    //判断当前workers中的线程数量有没有超过核心线程数
    if (workerCountOf(c) < corePoolSize) {
        //如果没有则创建核心线程数(参数true指的就是核心线程)
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //如果超过核心线程数了 先校验线程池是否正常运行后向阻塞队列workQueue末尾添加任务
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //再次检查线程池运行状态,若不在运行则移除该任务并且执行拒绝策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        //若果没有线程在执行
        else if (workerCountOf(recheck) == 0)
            //则创建一个空的worker 该worker从队列中获取任务执行
            addWorker(null, false);
    }
    //否则直接添加非核心线程执行任务 若非核心线程也添加失败 则执行拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}

//addWorker通过cas保证了并发安全性
private boolean addWorker(Runnable firstTask, boolean core) {
    	//第一部分 计数判断,不符合返回false
        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);
                //判断线程数,最大29位(CAPACITY=29位二进制),所以设置线程池的线程数不是任意大的
                if (wc >= CAPACITY ||
                    //判断工作中的核心线程是否大于设置的核心线程或者设置的最大线程数
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //通过cas新增 若添加失败会一直重试 若成功则跳过结束retry
                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
            }
        }
		//第二部分:创建新的work放入works(一个hashSet)
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            //将task任务封装在新建的work中
            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();
                        //将work加入到workers中
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    //上述work添加成功了,就开始执行任务操作了
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //如果上述添加任务失败了,会执行移除该任务操作
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

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;
		//检查当前worker中线程数量是否超过max 并且上次循环poll等待超时了,则将队列数量进行原子性减
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
 
            try {
                //线程可以被释放,那就是poll,释放时间就是keepAliveTime
                //否则,线程不会被释放,take一直阻塞在这里,直至新任务继续工作
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                //到这里说明可被释放的线程等待超时，已经销毁，设置该标记，下次循环将线程数减少
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

public class PrivateLock {
   Lock lock = new ReentrantLock();
   long start = System.currentTimeMillis();
   void read() {
       lock.lock();
       try {
           Thread.sleep(100);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }finally {
           lock.unlock();
       }
       System.out.println("read time = "+(System.currentTimeMillis() - start));
   }
   public static void main(String[] args) {
       final PrivateLock lock = new PrivateLock();
       for (int i = 0; i < 10; i++) {
           new Thread(new Runnable() {
               public void run() {
                   lock.read();
               }
           }).start();
       }
   }
}
 

public class SharedLock {
   ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
   Lock lock = readWriteLock.readLock();
   long start = System.currentTimeMillis();
   void read() {
       lock.lock();
       try {
           Thread.sleep(100);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }finally {
           lock.unlock();
       }
       System.out.println("end time = "+(System.currentTimeMillis() - start));
   }
   public static void main(String[] args) {
       final SharedLock lock = new SharedLock();
       for (int i = 0; i < 10; i++) {
           new Thread(new Runnable() {
               public void run() {
                   lock.read();
               }
           }).start();
       }
   }
}

Lock lock = readWriteLock.writeLock();

public class ParentLock {
   byte[] lock = new byte[0];
   public void f1(){
       synchronized (lock){
           System.out.println("f1 from parent");
       }
   }
}

public class SonLock extends ParentLock {
   public void f1() {
       synchronized (super.lock){
           super.f1();
           System.out.println("f1 from son");
       }
   }
   public static void main(String[] args) {
       SonLock lock = new SonLock();
       lock.f1();
   }
}

public class NestedLock {
   public synchronized void f1(){
       System.out.println("f1");
   }
   public synchronized void f2(){
       f1();
       System.out.println("f2");
   }
   public static void main(String[] args) {
       NestedLock lock = new NestedLock();
       //可以正常打印 f1,f2
       lock.f2();
   }
}

举个生活的例子，假设公司只有一个会议室（共享资源）
偏向锁：
前期公司只有1个团队，那么什么时候开会都能满足，就不需要询问和查看会议室的占用情况，直接进入使用
状态。会议室门口挂了个牌子写着A使用，A默认不需要预约（ThreadID=A）
轻量级锁：
随着业务发展，扩充为2个团队，B团队肯定不会同意A无法无天，于是当AB同时需要开会时，两者竞争，谁抢
到谁算谁的。偏向锁升级为轻量级锁，但是未抢到者在门口会不停敲门询问（自旋，循环），开完没有？开完
没有？
重量级锁：
后来发现，这种不停敲门的方式很烦，A可能不理不睬，但是B要不停的闹腾。于是锁再次升级。
如果会议室被A占用，那么B团队直接闭嘴，在门口安静的等待（wait进入阻塞），直到A用完后会通知
B（notify）。
 

public class ObjectLock {
   public static Integer i=0;
   public void inc(){
       synchronized (this){
           int j=i;
           try {
               Thread.sleep(100);
               j++;
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
           i=j;
       }
   }
   public static void main(String[] args) throws InterruptedException {
       for (int i = 0; i < 10; i++) {
           new Thread(new Runnable() {
               public void run() {
                 //重点！
                   new ObjectLock().inc();
               }
           }).start();
       }
       Thread.sleep(3000);
       //理论上10才对。可是....
       System.out.println(ObjectLock.i);
   }
}

public class AtomicCounter {
   private static AtomicInteger i = new AtomicInteger(0);
   public int get(){
       return i.get();
   }
   public void inc(){
       try {
           Thread.sleep(100);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       i.incrementAndGet();
   }
   public static void main(String[] args) throws InterruptedException {
       final AtomicCounter counter = new AtomicCounter();
       for (int i = 0; i < 10; i++) {
           new Thread(new Runnable() {
               public void run() {
                   counter.inc();
               }
           }).start();
       }
       Thread.sleep(3000);
       //同样可以正确输出10
       System.out.println(counter.i.get());
   }
}

public class BadAtomic {
   AtomicInteger i = new AtomicInteger(0);
   static int j=0;
   public void badInc(){
       int k = i.incrementAndGet();
       try {
           Thread.currentThread().sleep(new Random().nextInt(100));
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       j=k;
   }
   public static void main(String[] args) throws InterruptedException {
       BadAtomic atomic = new BadAtomic();
       for (int i = 0; i < 10; i++) {
           new Thread(()->{
               atomic.badInc();
           }).start();
       }
       Thread.sleep(3000);
       System.out.println(atomic.j);
   }
}