❤ 爆肝JUC 包中的 API 解读

落日映苍穹つ 2022-09-10 00:09 141阅读 0赞

### 文章目录 ###

*   *  一、JUC包中的锁应用？
     *   *  1.1 Lock接口及ReentrantLock对象分析及应用？
         *  1.2 Condition接口对象分析与应用？
         *  1.3 ReadWriteLock接口及实现类分析与应用？
         *  1.4 StampedLock对象分析及应用？
     *  二、JUC包中的原子(Atomic)类应用？
     *   *  2.1 Java中的无锁对象应用？
         *  2.2 Java中的无锁对象原理分析？
         *  2.3 Java中的无锁对象问题分析？
     *  三、JUC包中的并发工具类应用？
     *   *  3.1 CountDownLatch对象分析及应用
         *  3.2 CyclicBarrier对象分析及应用
         *  3.3 Semaphore对象分析及应用
     *  四、JUC包中的线程池应用？
     *   *  4.1 Java线程池简述？
         *  4.2 ThreadPoolExecutor对象应用
         *  4.3 ScheduledThreadPoolExecutor对象应用
     *  五、总结
     *   *  5.1 重难点分析
         *  5.2 常见FAQ
         *  5.3 BUG分析

## 一、JUC包中的锁应用？ ##

### 1.1 Lock接口及ReentrantLock对象分析及应用？ ###

并发编程领域，有两大核心问题：  
① 一个是互斥，即同一时刻只允许一个线程访问共享资源；  
② 另一个是同步，即线程之间如何通信、协作。  
这两大问题，在Java SDK 并发包可通过 Lock 和 Condition 两个接口来实现，其中Lock 用于解决互斥问题，Condition 用于解决同步问题。Java SDK 并发包里的 Lock 接口中，不仅有支持类似 synchronized 的隐式加锁方法，还支持超时、非阻塞、可中断的方式获取锁， 这三种方式为我们编写更加安全、健壮的并发程序提供了很大的便利。

我们来一起看看Lock接口常用方法，关键方法如下：

*  `void lock()`：获取锁对象，优先考虑是锁的获取，而非中断；
 *  `void lockInterruptibly()`：获取锁，但优先响应中断而非锁的获取；
 *  `boolean tryLock()`：试图获取锁；
 *  `boolean tryLock(long timeout, TimeUnit timeUnit)`：试图获取锁，并设置等待时长；
 *  `void unlock()`：释放锁对象。

ReentrantLock实现了Lock接口，是一个可重入的互斥锁(“独占锁”), 同时提供了”公平锁”和”非公平锁”的支持。所谓公平锁和非公平锁其含义如下：

*  公平锁：在多个线z程争用锁的情况下，公平策略倾向于将访问权授予等待时间最长的线程。也就是说，相当于有一个线程等待队列，先进入等待队列的线程后续会先获得锁。
 *  非公平锁：在多个线程争用锁的情况下，能够最终获得锁的线程是随机的（由底层OS调度）。

ReetrantLock简易应用如下（默认为非公平策略）

class Counter{ 
    	ReentrantLock lock = new ReentrantLock();
    	int count = 0;
    	void increment() { 
    	lock.lock();
    	try { 
    		count++;
    	} finally { 
    		lock.unlock();
    	}
    }

其中，这里的锁通过 `lock()` 方法获取锁,通过 `unlock()` 方法释放锁。重要的是将代码包装成try/finally块，以确保在出现异常时解锁。这个方法和synchronized关键字修饰的方法一样是线程安全的。在任何给定的时间，只有一个线程可以持有锁。

ReetrantLock对象在构建时，可以基于ReentrantLock(boolean fair)构造方法参数，设置对象的”公平锁”和”非公平锁”特性。其中fair的值true表示“公平锁”。这种公平锁，会影响其性能，但是在一些公平比效率更加重要的场合中公平锁特性就会显得尤为重要。关键代码示例如下：

public void performFairLock(){ 
    	//...
    	ReentrantLock lock = new ReentrantLock(true);
    	try { 
    		//Critical section here
    	} finally { 
    		lock.unlock();
    	}
    	//...
    }

ReetrantLock对象在基于业务获取锁时，假如希望有等待时间，可以借助tryLock实现，关键代码示例如下：

public void performTryLock(){ 
    	//...
    	boolean isLockAcquired = lock.tryLock(1, TimeUnit.SECONDS);
    	if(isLockAcquired) { 	
    		try { 
    			//Critical section here
    		} finally { 
    			lock.unlock();
    		}
    	}
    	//...
    }

“锁”是为了保护竞争资源，防止多个线程同时操作线程而出错，ReentrantLock在同一个时间点只能被一个线程获取(当某线程获取到“锁”时，其它线程就必须等待)；ReentrantLock是通过一个FIFO（先进先出）的等待队列来管理获取该锁所有线程的。在“公平锁”的应用机制下，线程依次排队获取锁；而“非公平锁”，在锁是可获取状态时，不管自己是不是在队列的开头都会获取锁。

**参考代码**

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * @author archibald * @date 2021/9/2 * @apiNote */
    public class ReentrantLockTests { 
    	
    	public static void main(String[] args) throws InterruptedException { 
    		//doTestGetLock();
    		Counter c=new Counter();
    		for(int i=0;i<1000;i++){ 
    			new Thread(()->c.increment()).start();
    		}
    		TimeUnit.SECONDS.sleep(1);
    		System.out.println(c.count);
    	}
    	static class Counter{ 
    		int count;
    		ReentrantLock lock=new ReentrantLock();
    		void increment() { 
    			lock.lock();
    			//lock.lockInterruptibly();
    			try { 
    				count++;
    			}finally { 
    				lock.unlock();
    			}
    		}
    	}
    
    	private static void doTestGetLock() throws InterruptedException { 
    		ReentrantLock lock=new ReentrantLock();
    		lock.lock();
    		Thread t1=new Thread() { 
    			@Override
    			public void run() { 
    				//lock.lock();//获取锁，拿不到锁会一直阻塞,不响应中断
    				//lock.lockInterruptibly();//获取锁，但会优先响应中断
    				//if(lock.tryLock()) {//尝试获取锁,当拿不到锁，不阻塞
    				try { 
    					if(lock.tryLock(5,TimeUnit.SECONDS)) { 
    						System.out.println("thread.execute");
    					}else { 
    						System.out.println("do not get lock");
    					}
    				}catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    			};
    		};
    		t1.start();
    		TimeUnit.SECONDS.sleep(1);//休眠一秒钟
    		t1.interrupt();
    		//lock.unlock();
    	}
    }

### 1.2 Condition接口对象分析与应用？ ###

JUC包中提供的Condition接口对象，实现了对Object对象中的wait()、notify()、notifyAll()方法功能的增强。我们可以借助Condition接口对象实现线程之间更加高效的通讯。

Condition相关方法介绍：  
（1）`await()`方法相当于Object的wait()方法。  
（2）`signal()`方法相当于Object的notify()方法。  
（3）`signalAll()`方法相当于Object的notifyAll()方法。

案例分析：基于Condition实现阻塞式栈(Stack)

关键代码分析：

class BlockingStack{ 
    	LinkedList<String> stack = new LinkedList<>();
    	int CAPACITY = 3;
    	ReentrantLock lock = new ReentrantLock();
    	Condition pushCondition = lock.newCondition();
    	Condition popCondition = lock.newCondition();
    	public void pushToStack(String item) throws InterruptedException{ 
    		try { 
    			lock.lock();
    			while(stack.size() == CAPACITY) { 
    				pushCondition.await();
    			}
    			stack.push(item);
    			popCondition.signalAll();
    		} finally { 
    			lock.unlock();
    		}
    	}
    	public String popFromStack() throws InterruptedException{ 
    		try { 
    			lock.lock();
    			while(stack.size() == 0) { 
    				popCondition.await();
    			}
    			String e=stack.pop();
    			pushCondition.signalAll();
    			return e;
    		} finally { 
    			lock.unlock();
    		}
    	}
    }

Condition的强大之处在于它可以为多个线程间建立不同的Condition。我们知道对于栈而言：  
① 假设栈中数据已经满了，那么阻塞的肯定是写线程，唤醒的肯定是读线程；  
② 相反，阻塞的肯定是读线程，唤醒的肯定是写线程；

那么假设只有一个Condition会有什么效果呢，缓存队列中已经存满，这个Lock不知道唤醒的是读线程还是写线程了，如果唤醒的是读线程，皆大欢喜，如果唤醒的是写线程，那么线程刚被唤醒，又被阻塞了，这时又去唤醒，这样就浪费了很多时间。

**参考代码**

package com.cy.pj.juc.lock;
    
    import com.sun.org.apache.bcel.internal.generic.POP;
    
    import java.util.LinkedList;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * @author archibald * @date 2021/9/3 * @apiNote */
    
    //阻塞式栈结构实现
    class BlockingStack { 
        //stack结构
        private LinkedList<String> stack = new LinkedList<>();
    
        //capacity,默认容量
        private int capacity = 3;
    
        //lock object
        private ReentrantLock lock = new ReentrantLock();
    
        //Condition
        Condition pushCondition = lock.newCondition();
        Condition popCondition = lock.newCondition();
    
        public BlockingStack() { }
        public BlockingStack(int capacity) { 
            this.capacity = capacity;
        }
    
        //入栈
        public void push(String element) throws InterruptedException { 
            //1.获取锁
            lock.lockInterruptibly();
            try { 
                //2.检测stack是否已满，满了则阻塞生产者线程
                while (stack.size() == capacity) { 
                    pushCondition.await();
                }
                //3.放数据
                stack.push(element);
                //4.通知消费者线程取数据
                popCondition.signalAll();
            } finally { 
                lock.unlock();
            }
        }
    
        //出栈
        public String pop() throws InterruptedException { 
            //1.获取锁对象
            lock.lockInterruptibly();
            try { 
                //2.检测stack是否为空，为空则等待
                while (stack.size() == 0) { 
                    popCondition.await();
                }
                //3.取数据
                String temp = stack.pop();
                //4.通知生产者线程继续放数据
                pushCondition.signalAll();
                return temp;
            } finally { 
                lock.unlock();
            }
        }
    }
    
    public class ConditionTests { 
        public static void main(String[] args) throws InterruptedException { 
            BlockingStack stack = new BlockingStack();
            stack.push("A");
            stack.push("B");
            stack.push("C");
            //stack.push("D");
    
            stack.pop();
            stack.pop();
            stack.pop();
            //stack.pop();
        }
    }

### 1.3 ReadWriteLock接口及实现类分析与应用？ ###

ReadWriteLock是一个是读写锁接口。其中“读锁”又称“**共享锁**”，能同时被多个线程获取。“写锁”又称**独占锁**，只能被一个线程获取。  
读写锁与互斥锁的一个重要区别就是读写锁允许多个线程同时读共享变量，而互斥锁是不允许的，这是读写锁在读多写少的场景下性能优于互斥锁的关键。  
但读写锁的写操作是互斥的，当一个线程在写共享变量的时候，是不允许其他线程执行写操作和读操作的。

案例分析：ReentrantReadWriteLock 对象应用

构建一个简易的线程安全的MapCache对象，并允许多个线程同时从cache读数据。

class MapCache{ 
    	private Map<String,Object> map=new HashMap<>();
    	private ReadWriteLock readWriteLock=new ReentrantReadWriteLock();
    	public void writeObject(String key,Object value){ 
    		readWriteLock.writeLock().lock();
    		try { 
    			map.put(key, value);
    		}finally { 
    			readWriteLock.writeLock().unlock();
    		}
    	}
    	public Object readObject(String key) { 
    		readWriteLock.readLock().lock();
    		try { 
    			return map.get(key);
    		}finally { 
    			readWriteLock.readLock().unlock();
    		}
    	}
    }

ReentrantReadWriteLock可以让多个读线程可以同时持有读锁（只要写锁未被占用），而写锁是独占的。但是，假如读写锁使用不当，很容易产生“饥饿”问题：比如在读线程非常多，写线程很少的情况下，很容易导致写线程“饥饿”，虽然使用“公平”策略可以一定程度上缓解这个问题，但是“公平”策略是以牺牲系统吞吐量为代价的。

ReadWriteLock 锁应用的升降级应用

ReadWriteLock中的锁不支持升级操作，比方说我们在读锁内获取写锁，这个过程我们通常理解为锁的升级。代码分析如下：

static void doMethod01() { 
    	ReadWriteLock rtLock = new ReentrantReadWriteLock();
    	rtLock.readLock().lock();
    	System.out.println("get readLock.");
    	rtLock.writeLock().lock();
    	System.out.println("get writeLock");
    }

ReadWriteLock中的锁虽不支持升级操作，但支持降级操作，比方说我们在写锁内获取读锁，这个过程我们通常理解为锁的降级。代码分析如下

static void doMethod02() { 
    	ReadWriteLock rtLock = new ReentrantReadWriteLock();
    	rtLock.writeLock().lock();
    	System.out.println("get writeLock.");
    	rtLock.readLock().lock();
    	System.out.println("get readLock");
    }

练习：分析如下代码检查是否存在问题？

r.lock();        
    	try { 
    		v = m.getObject(key);
    		if (v == null) { 
    			w.lock();
    			try { 
    				//更新缓存(代码省略)
    			} finally{ 
    				w.unlock();
    			}
    		}
    	} finally{ 
    		r.unlock();
    	}

对于如上代码，看上去好像是没有问题的，先是获取读锁，然后再升级为写锁。可惜 ReadWriteLock 并不支持这种升级。在上面的代码示例中，读锁还没有释放，此时获取写锁，会导致写锁永久等待，最终导致相关线程都被阻塞，永远也没有机会被唤醒。那如何修改呢？参考代码如下：

r.lock();
    	try { 
    		v = m.getObject(key);
    		if (v == null) { 
    			r.unlock();
    			w.lock();
    			try { 
    				//假如缓存没有从数据库或一级缓存查询
    				//然后更新缓存(代码省略)
    				r.lock();
    			} finally{ 
    				w.unlock();
    			}
    		}
    	} finally{ 
    		r.unlock();
    	}

**参考代码**

package com.cy.pj.juc.lock;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.locks.ReadWriteLock;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    /** * @author archibald * @date 2021/9/3 * @apiNote */
    
    class MapCache{ 
        //HashMap作为缓存数据的对象
        private Map<String,Object> cache = new HashMap<>();
        //写锁ReadWriteLock
        private ReadWriteLock readWriteLock  = new ReentrantReadWriteLock();
        //writeObject() 向缓存写数据，写锁，互斥锁
        public void writeObject(String key,Object value) { 
            //加锁
            readWriteLock.writeLock().lock();
            //测试
            System.out.println(Thread.currentThread().getName() +"->writeObject");
            try { 
                cache.put(key,value);
                try { 
                    Thread.sleep(3000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            } finally { 
                System.out.println(Thread.currentThread().getName() +"->writeObjectFinally");
                //释放锁
                readWriteLock.writeLock().unlock();
            }
        }
    
        //读操作，读锁，共享锁
        public Object readObject(String key) { 
            readWriteLock.readLock().lock();
            System.out.println(Thread.currentThread().getName() +"->readObject");
            try { 
                Object temp = cache.get(key);
                return temp;
            } finally { 
                System.out.println(Thread.currentThread().getName() +"->readObjectFinally");
                readWriteLock.readLock().unlock();
            }
        }
    }
    
    public class ReadWriteLockTests { 
        public static void main(String[] args) { 
            //doTestReadWriteLock01();
            //doMethod01();
            doMethod02();
        }
    
        static void doMethod01() { //锁升级，不允许
            ReadWriteLock rtLock = new ReentrantReadWriteLock();
            rtLock.readLock().lock();
            System.out.println("get readLock.");
            rtLock.writeLock().lock();
            System.out.println("get writeLock");
        }
    
        static void doMethod02() { 
            ReadWriteLock rtLock = new ReentrantReadWriteLock();
            rtLock.writeLock().lock();
            System.out.println("get writeLock.");
            rtLock.readLock().lock();
            System.out.println("get readLock");
        }
    
        private static void doTestReadWriteLock01() { 
            //测试写操作
            MapCache cache = new MapCache();
            new Thread(()->cache.writeObject("A",100)).start();
            new Thread(()->cache.writeObject("B",100)).start();
    
            //测试读操作
            new Thread(()->cache.readObject("A")).start();
            new Thread(()->cache.readObject("B")).start();
            new Thread(()->cache.readObject("C")).start();
        }
    }

### 1.4 StampedLock对象分析及应用？ ###

StampedLock类，在JDK1.8时引入，是对读写锁ReadWriteLock的增强，该类优化了读锁、写锁的应用，同时使读写锁之间可以互相转换，实现了更加细粒度的并发控制。StampedLock 支持三种模式，分别是写锁、悲观读锁和乐观读。其中，写锁、悲观读锁的语义和 ReadWriteLock 的写锁、读锁的语义非常类似，允许多个线程同时获取悲观读锁，但是只允许一个线程获取写锁，写锁和悲观读锁是互斥的。不同的是，StampedLock 里的写锁和悲观读锁加锁成功之后，都会返回一个 stamp；然后解锁的时候，需要传入这个 stamp。

StampedLock 应用案例分析：

基于StampedLock 实现一个线程安全的cache对象。

class StampedMapCache{ 
    	private Map<String,Object> map=new HashMap<>();
    	private StampedLock lock=new StampedLock ();
    	public void writeObject(String key,Object value){ 
    		long stamp=lock.writeLock();
    		try { 
    			map.put(key, value);
    		}finally { 
    			lock.unlock(stamp);
    		}
    	}
    	public Object readObject(String key) { 
    		long stamp=lock.readLock();
    		try { 
    			return map.get(key);
    		}finally { 
    			Lock.unlock(stamp);
    		}
    	}
    }

StampedLock 的性能之所以比 ReadWriteLock 还要好，其关键是 StampedLock 支持乐观读的方式。ReadWriteLock 支持多个线程同时读，但是当多个线程同时读的时候，所有的写操作会被阻塞；而 StampedLock 提供的乐观读，是允许一个线程获取写锁的，也就是说不是所有的写操作都被阻塞。 注意这里，我们用的是“乐观读”这个词，而不是“乐观读锁”，是要提醒你，乐观读这个操是无锁的，所以相比ReadWriteLock 的读锁，乐观读的性能更好一些。

StampedLock 对象基于乐观读的方式从缓存对象中获取数据。

public String readWithOptimisticLock(String key) { 
    	long stamp = lock.tryOptimisticRead();
    	String value = map.get(key);
    	if(!lock.validate(stamp)) { 
    		stamp = lock.readLock();
    		try { 
    			return map.get(key);
    		} finally { 
    			lock.unlock(stamp);              
    		}
    	}
    	return value;
    }

其中：在代码中，首先通过调用 lock对象的tryOptimisticRead()方法 获取了一个 stamp，这里的 tryOptimisticRead() 就是我前面提到的乐观读。需要注意的是，由于 tryOptimisticRead() 是无锁的，因此最后读完之后，还需要再次验证一下是否存在写操作(这个验证操作是通过调 用 validate(stamp) 来实现的)，来保证数据的一致性。

StampedLock对象在应用时需要注意如下几个点：

*  获取锁时返回一个Stamp值，值为0表示获取失败，其余都表示成功；
 *  释放锁时需要一个Stamp值，这个值必须是和成功获取锁时得到的Stamp值是一致的；
 *  StampedLock是不可重入的。（如果一个线程已经持有了写锁，再去获取写锁的话就会造成死锁）
 *  StampedLock有三种访问模式：
    
     *  第一：Reading（读模式）：功能和ReentrantReadWriteLock的读锁类似
     *  第二：Writing（写模式）：功能和ReentrantReadWriteLock的写锁类似
     *  第三：Optimistic reading（乐观读模式）：这是一种优化的读模式。
 *  StampedLock支持读锁和写锁的相互转换。
 *  无论写锁还是读锁，都不支持Conditon等待。

> 总之： 相比ReadWriteLock读写锁，StampedLock通过提供乐观读在多线程多写线程少的情况下可以提供更好的性能，因为乐观读不需要进行CAS设置锁的状态。

package com.cy.pj.juc.lock;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.locks.StampedLock;
    
    /** * @author archibald * @date 2021/9/3 * @apiNote */
    
    class StampedMapCache { 
        private Map<String, Object> cache = new HashMap<>();
        private StampedLock lock = new StampedLock();
    
        //写动作
        public void writeObject(String key, Object value) { 
            //1.获取写锁
            //独占锁
            long stamp = lock.writeLock();
            try { 
                //2.执行写操作
                cache.put(key,value);
            } finally { 
                //3.释放锁
                lock.unlock(stamp);
            }
        }
    
        public Object readObject(String key) { 
            //1.获取读锁
            // 悲观读（共享锁）
            long stamp = lock.readLock();
            try { 
                //2.读数据
                Object object = cache.get(key);
                return object;
            } finally { 
                //3.释放资源
                lock.unlock(stamp);
            }
        }
    
        public Object readWithOptimisticRead(String key) { 
            //1.尝试获取乐观读
            long stamp = lock.tryOptimisticRead();
            if (stamp == 0) { 
                System.out.println("readWithOptimisticRead.stamp:"+stamp);
            }
            //2.从cache取数据
            Object value = cache.get(key);
            //3.验证乐观读时是否有现成执行了写操作
            if (!lock.validate(stamp)){ 
                //4.假如有写操作，则重新读取数据
                stamp = lock.readLock();
                try { 
                    return cache.get(key);
                } finally { 
                    lock.unlock(stamp);
                }
            }
            //5.返回读取的数据
            return value;
        }
    }
    
    public class StampedLockTests { 
        public static void main(String[] args) { 
            StampedMapCache cache = new StampedMapCache();
            cache.writeObject("A",100);
            cache.writeObject("B",200);
    
            //创建一个线程不断地去写
            new Thread() { 
                @Override
                public void run() { 
                    int value = 500;
                    while (true) { 
                        cache.writeObject("A",value);
                        value++;
                    }
                };
            }.start();
            //创建一个线程不断地去读
            new Thread() { 
                @Override
                public void run() { 
                    Object value = cache.readWithOptimisticRead("A");
                    //System.out.println(value);
                }
            }.start();
        }
    }

## 二、JUC包中的原子(Atomic)类应用？ ##

### 2.1 Java中的无锁对象应用？ ###

我们先看个案例，关键代码如下：

int count=1;
    int count() { 
    	return count++;
    }

其中：count() 这个方法不是线程安全的，问题就出在变量count的可见性和count++的原子性上。可见性问题可以用volatile来解决，而原子性问题我们前面一直都是采用的互斥锁方案。但这种方案，性能上会有一定的损失。其实对于简单的原子性问题，还有一种无锁方案。JUC并发包将这种无锁方案封装提炼之后，实现了一系列的原子类。

AtomicLong 类型对象案例应用：

AtomicLong al=new AtomicLong(1);
    long atomicCount(){ 
    	return al.getAndIncrement();
    }

AtomicLong atomicLong=new AtomicLong(0);
    LongStream.range(0, 1000)
    					.parallel()
    					.forEach((t)->atomicLong.incrementAndGet());
    					System.out.println(atomicLong.get());

无锁方案相对互斥锁方案，最大的好处就是性能。 互斥锁方案为了保证互斥性，需要执行加锁、 解锁操作，而加锁、解锁操作本身就消耗性能。同时拿不到锁的线程还会进入阻塞状态，进而触发线程切换，线程切换对性能的消耗也很大。 相比之下，无锁方案则完全没有加锁、解锁的性能消耗，可谓绝佳方案。

### 2.2 Java中的无锁对象原理分析？ ###

无锁化实现的原理其实也很简单，硬件支持而已。CPU 为了解决并发问题，提供了 CAS 指令（CAS，全称是 Compare And Swap，即“比较并交换”）。CAS 指令包含 3 个参数：共享变量的内存地址 A、用于比较的值 B 和共享变量的新值 C；并且只有当内存中地址 A 处的值等于 B时，才能将内存中地址 A 处的值更新为新值 C。作为一条 CPU 指令，CAS 指令本身是能够保证原子性的。

简易的CAS算法的实现：

class SimulatedCAS{ 
    	int count;
    	synchronized int cas(int expect, int newValue){ 
    		// 读⽬前 count 的值
    		int curValue = count;
    		// ⽐较⽬前 count 值是否 == 期望值
    		if(curValue == expect){ 
    			// 如果是，则更新 count 的值
    			count = newValue;
    		}
    		// 返回写⼊前的值
    		return curValue;
    	}
    }

> 说明：原子类的方法都是针对”一个共享变量“的，如果你需要解决多个变量的原子性问题，建议还是使用互斥锁方案。

### 2.3 Java中的无锁对象问题分析？ ###

Java中的无锁方案相对于互斥锁方案，优点非常多，首先性能好，其次是基本不会出现死锁问题，但也会有一些问题，例如：

**ABA问题：**

对于一个旧的变量值A,线程2将A的值改成B又改成可A,此时线程1通过CAS看到A并没有变化，但实际A已经发生了变化，这就是ABA问题。解决这个问题的方法很简单，记录一下变量的版本就可以了，在变量的值发生变化时,对应的版本也做出相应的变化，然后CAS操作时比较一下版本就知道变量有没有发生变化。此时可借助在java的atomic包下的AtomicStampedReference类进行实现。

**自旋问题：**

无锁对象会多次尝试CAS操作直至成功或失败，这个过程叫做自旋。通过自旋的过程我们可以看出自旋操作不会将线程挂起，从而避免了内核线程切换，但是自旋的过程也可以看做CPU死循环，会一直占用CPU资源。这种情形在单CPU的机器上是不能容忍的，因此自旋一般都会有个次数限制，即超过这个次数后线程就会放弃时间片，等待下次机会。因此自旋操作在资源竞争不激烈的情况下确实能提高效率，但是在资源竞争特别激烈的场景中，CAS操作会的失败率就会大大提高，这时使用中重量级锁的效率可能会更高。当前，也可以使用 `LongAdder` 类来替换，它则采用了分段锁的思想来解决并发竞争的问题。

LongStream.range(0, 1000)
    				.parallel()
    				.forEach((t)->longAdder.increment());
    				System.out.println(longAdder.sumThenReset());

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.atomic.AtomicLong;
    import java.util.concurrent.atomic.LongAdder;
    import java.util.stream.LongStream;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class AtomicTests { 
        public static void main(String[] args) { 
            //java.util.concurrent.atomic.AtomicLong
            //CAS算法 --> 资源竞争不够激烈的情况下使用
            AtomicLong atomicLong = new AtomicLong(0);
            LongStream.range(0,1000)
                    .parallel()
                    .forEach((t)->atomicLong.incrementAndGet());
            System.out.println(atomicLong.get());
    
            //java.util.concurrent.atomic.AtomicMarkableReference
            //分段加锁 --> 资源竞争非常激烈的情况下使用
            LongAdder longAdder = new LongAdder();;
            LongStream.range(0,2000)
                    .parallel()
                    .forEach((t)->longAdder.increment());
            System.out.println(longAdder.sumThenReset());
        }
    }

## 三、JUC包中的并发工具类应用？ ##

### 3.1 CountDownLatch对象分析及应用 ###

CountDownLatch是一个辅助同步类，用来协调多个线程之间的同步(例如一个线程等待其它线程的某些操作执行结束再继续执行)。我们可以将CountDownLatch作为一个倒数计数器使用，构建对象时，先设定一个计数初始值，每调用一次 `countDown()` 方法，计数值就会减1。线程可以调用 `CountDownLatch` 的 `await` 方法进入阻塞，当计数值降到0时，所有之前调用 `await` 阻塞的线程会被激活。

CountDownLatch 应用案例：

public class TestCountDownLatch01 { 
    	static String content;
    	public static void main(String[] args)throws Exception { 
    		CountDownLatch cdl=new CountDownLatch(1);
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				content="helloworld";
    				cdl.countDown();
    			}
    		}).start();
    		while(content==null)cdl.await();
    		System.out.println(content.toUpperCase());
    	} 
    }

说明：CountDownLatch的不足之处是，**它的初始计数值一旦降到0，无法重置**。如果需要重置，可以考虑使用CyclicBarrier。

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.Locale;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class CountDownLatchTests { 
    
        static String content;
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch latch = new CountDownLatch(1);
            Thread t = new Thread() { 
                @Override
                public void run() { 
                    content = "helloworld";
                    latch.countDown();
                    while (true) { }
                }
            };
            t.start();
            //while (content == null);
            //TimeUnit.SECONDS.sleep(1);
            //t.join();
            while (content == null) { 
                latch.await();
            }
            System.out.println(content.toUpperCase());
        }
    }

### 3.2 CyclicBarrier对象分析及应用 ###

CyclicBarrier也是一个同步工具类，用来协调多个线程之间的同步（例如让多个线程之间相互等待）。我们可以将CyclicBarrier理解为一个可以循环使用的栅栏，它做的事情就是让线程到达栅栏时被阻塞(调用await方法)，直到到达栅栏的线程数满足指定数量要求时，栅栏才会打开放行。这个应用，其实有点像军训报数，报数总人数满足教官认为的总数时，教官才会安排后面的训练。

CyclicBarrier 应用案例：

public class TestCyclicBarrier01{ 
    	static CyclicBarrier cBarrier = new CyclicBarrier(3,new Runnable() { 
    		@Override
    		public void run() { 
    			System.out.println("step-01 all complete");
    		}
    	});
    	static class WorkTask implements Runnable{ 
    		@Override
    		public void run() { 
    			try{ 
    				String tName=Thread.currentThread().getName();
    				System.out.println("step-01:"+tName);
    				cBarrier.await();
    				System.out.println("step-02:"+tName);
    			}catch(Exception e){ e.printStackTrace();}
    		}
    	}
    	public static void main(String[] args) { 
    		WorkTask task=new WorkTask();
    		for(int i=0;i<3;i++){ 
    			new Thread(task).start();
    		}
    	}
    }

CyclicBarrier典型应用是一组任务，它们并行执行工作，然后在进行下一个步骤之前进行等待，直至所有的任务都完成。

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.TimeUnit;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class CyclicBarrierTests { 
    
        //循环栅栏
        static CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() { 
            @Override
            public void run() { 
                System.out.println("step01 all complete");
            }
        });
    
        static class WorkTask implements Runnable { 
            @Override
            public void run() { 
                String tName = Thread.currentThread().getName();
                System.out.println(tName+"->step-01");
                //等待其它线程完成第一阶段任务以后，再继续执行
                try { 
                    cyclicBarrier.await();
                } catch (Exception e) { 
                    e.printStackTrace();
                }
                System.out.println(tName+"->step-02");
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            WorkTask workTask = new WorkTask();
            for (int i=0; i<3; i++){ 
                new Thread(workTask).start();
            }
    
            TimeUnit.SECONDS.sleep(2);
            cyclicBarrier.reset();
            for (int i=0;i<3;i++) { 
                new Thread(workTask).start();
            }
        }
    }

### 3.3 Semaphore对象分析及应用 ###

Semaphore，又名信号量，此对象可以维护了一定数量的 “许可证”。通过许可证控制同时访问共享资源的最大线程数（类似限流），当有线程想要访问共享资源时，需要先获取(acquire)许可；如果许可不够了，线程需要一直等待，直到许可可用。当线程使用完共享资源后，可以归还(release)许可，以供其它需要的线程使用。这样可以更好的保证合理的使用公共资源。

基于Semaphore实现限流操作

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    
    //限制服务
    class LimitService { 
        private Semaphore permit = new Semaphore(5,true);
        public void doService() { 
            //获取许可
            try { 
                permit.acquire();
                String tName = Thread.currentThread().getName();
                System.out.println(tName + "-> doService");
                TimeUnit.SECONDS.sleep(4);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                //释放许可
                permit.release();
            }
        }
    }
    
    public class SemaphoreTests { 
        public static void main(String[] args) { 
            LimitService limitService = new LimitService();
            for (int i = 0; i<10;i++) { 
                new Thread(()->limitService.doService()).start();
            }
        }
    }

说明，当许可数 ≤ 0代表共享资源不可用。许可数 ＞ 0，代表共享资源可用，且多个线程可以同时访问共享资源。

## 四、JUC包中的线程池应用？ ##

### 4.1 Java线程池简述？ ###

Java中创建线程对象远不像创建一个普通对象那么简单。创建一般的对象，可能仅仅是在JVM的堆里分配一块内存而已；而创建一个线程，却需要调用操作系统内核的API，然后操作系统要为线程分配一系列的资源，这个创建成本一般会很高，所以可以把线程理解为一个重量级的对象，应该避免频繁创建和销毁。Java中为了优化线程对象应用，提供了一些线程池类型的对象。如图所示：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA57uP55CG77yM5aSp5Y-w6aOO5aW95aSn_size_15_color_FFFFFF_t_70_g_se_x_16]

### 4.2 ThreadPoolExecutor对象应用 ###

JUC包中最核心的线程池类型为ThreadPoolExecutor，其常用构造函数如下：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

参数说明：

*  `corePoolSize`：表示线程池保有的核心线程数；
 *  `maximumPoolSize`：表示线程池创建的最大线程数；
 *  `keepAliveTime & unit`：线程空闲时间；
 *  `workQueue`：工作队列(存储任务的的一种容器)；
 *  `threadFactory`：创建线程的工厂；
 *  `handler`：通过这个参数指定任务的拒绝策略；

线程池拒绝策略：

如果线程池中所有的线程都在忙碌，并且工作队列也满了（前提是工作队列是有界队列），那么此时提交任务，线程池就会拒绝接收。ThreadPoolExecutor已经提供了如下几种种策略。

*  `CallerRunsPolicy`：提交任务的线程自己去执行该任务；
 *  `AbortPolicy`：默认的拒绝策略，会throws RejectedExecutionException；
 *  `DiscardPolicy`：直接丢弃任务，没有任何异常抛出；
 *  `DiscardOldestPolicy`：丢弃最老的任务。

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class TaskExecutorUtil { 
        static ThreadFactory threadFactory = new ThreadFactory() { 
            AtomicInteger atomicInteger = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) { 
                return new Thread(r,"thread-pool-"+atomicInteger.getAndIncrement());
            }
        };
        static BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(3);
    
        static ThreadPoolExecutor doCreateThreadPoolExecutor() { 
            return new ThreadPoolExecutor(3,
                                    5,
                                        60,
                                        TimeUnit.SECONDS,
                                        workQueue,
                                        threadFactory,
                                        new ThreadPoolExecutor.CallerRunsPolicy());
        }
    
        static void close(ExecutorService executorService) { 
            try { 
                executorService.shutdown();//设定一个shutdown标记
                executorService.awaitTermination(5,TimeUnit.SECONDS);
            }catch (Exception e) { 
                e.printStackTrace();
            }finally { 
                if (!executorService.isTerminated()) { 
                    System.out.println("task not finish");
                }
                //设定stop标记，不再接收新任务，并且会尝试关闭正在执行的任务，还会返回任务队列
                executorService.shutdownNow();
            }
        }
    }

ThreadPoolExecutor 应用案例分析

创建TaskExecutorUtil工具类，然后在类中添加创建线程池和关闭池的方法。

第一步:定义创建池对象的方法

public static ThreadPoolExecutor doCreateThreadPoolExecutor() { 
    	int corePoolSize=3;
    	int maximumPoolSize=5;
    	long keepAliveTime=60;
    	BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(3);
    	RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
    	ThreadFactory threadFactory=new ThreadFactory() { 
    		AtomicInteger at=new AtomicInteger(1);
    		@Override
    		public Thread newThread(Runnable r) { 
    			return new Thread(r,"pool-thread->"+at.getAndIncrement());
    		}
    	};
    	ThreadPoolExecutor pExecutor = new ThreadPoolExecutor(corePoolSize,
    																maximumPoolSize,
    																keepAliveTime,
    																TimeUnit.SECONDS,
    																workQueue,
    																threadFactory,
    																handler);
    	return pExecutor;
    }

第二步：定义关闭池对象

public  static void doCloseExecutor(ThreadPoolExecutor executor) { 
    	try { 
    		System.out.println("attempt to shutdown executor");
    		executor.shutdown();
    		executor.awaitTermination(5, TimeUnit.SECONDS);
    	}catch (InterruptedException e) { 
    		System.err.println("tasks interrupted");
    	}finally { 
    		if (!executor.isTerminated()) { 
    			System.err.println("cancel non-finished tasks");
    		}
    		executor.shutdownNow();
    		System.out.println("shutdown finished");
    	}
    }

shutdown只是将线程池的状态设置为 `SHUTWDOWN` 状态，正在执行的任务会继续执行下去，新的任务将会被拒绝。但这个方法不会等待提交的任务执行完，我们可以用awaitTermination来等待任务执行完。`shutdownNow()` 方法是线程池处于 `STOP` 状态，此时线程池不再接受新的任务，并且会去尝试终止正在执行的任务，然后清空并返回队列。整个过程类似超市或商场关门。

第三步：应用池对象中的线程执行任务

private static void doTestPoolExecutor01() throws Exception { 
    	ThreadPoolExecutor pExecutor = doCreateThreadPoolExecutor();
    	Future<Integer> future = pExecutor.submit(new Callable<Integer>() { 
    		@Override
    		public Integer call() throws Exception { 
    			TimeUnit.SECONDS.sleep(5);
    			return new Random().nextInt();
    		}
    	});
    	System.out.println("future done? " + future.isDone());
    	Integer result=future.get();
    	System.out.println("future done? " + future.isDone());
    	System.out.println(result);
    	doCloseExecutor(pExecutor);
    }

private static void doTestPoolExecutor02() throws Exception { 
    	ThreadPoolExecutor pExecutor = doCreateThreadPoolExecutor();
    	List<Callable<String>> callables = Arrays.asList(
    												() -> "task1",
    												() -> "task2",
    												() -> "task3");
    	pExecutor.invokeAll(callables).stream().map(future -> { 
    		try { 
    			return future.get();
    		}
    		catch (Exception e) { 
    			throw new IllegalStateException(e);
    		}
    	}).forEach(System.out::println);
    	doCloseExecutor(pExecutor);
    }

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.Arrays;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.*;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class ThreadPoolTests { 
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            //doTestThreadPool01();
            doTestThreadPool02();
        }
    
        private static void doTestThreadPool02() throws InterruptedException { 
            //1.创建池对象
            ThreadPoolExecutor threadPool = TaskExecutorUtil.doCreateThreadPoolExecutor();
            //2.执行批量任务
            List<Callable<String>> tasks = Arrays.asList(() -> "task01", () -> "task02", () -> "task03");
            threadPool.invokeAll(tasks)
                    .stream()
                    .map(f-> { 
                        try { 
                            return f.get();
                        } catch (Exception e) { 
                            throw new IllegalStateException(e);
                        }
                    })
                    .forEach(System.out::println);
            //3.关闭池对象
            TaskExecutorUtil.close(threadPool);
        }
        private static void doTestThreadPool01() throws ExecutionException, InterruptedException { 
            //1.创建池对象
            ThreadPoolExecutor threadPool = TaskExecutorUtil.doCreateThreadPoolExecutor();
            //2.基于池对象中的线程执行任务
            Future<Integer> future = threadPool.submit(new Callable<Integer>() { 
                @Override
                public Integer call() throws Exception { 
                    TimeUnit.SECONDS.sleep(2);
                    return new Random().nextInt();
                }
            });
            System.out.println("future.isDone()->" + future.isDone());
            Integer result = future.get();
            System.out.println("result->" + result);
            System.out.println("future.isDone()->" + future.isDone());
            //3.关闭池对象
            TaskExecutorUtil.close(threadPool);
        }
    }

### 4.3 ScheduledThreadPoolExecutor对象应用 ###

`ScheduledThreadPoolExecutor` 继承自 `ThreadPoolExecutor` 类，其内部将所有的 `Runnable` 任务包装成 `RunnableScheduledFuture` 类型，用于满足任务的延迟和周期性调度。

ScheduledThreadPoolExecutor 应用分析

创建一个Runnable任务，3秒以后执行。

static void doTestSchedule01() { 
    	ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
    	Runnable task = () -> System.out.println("Scheduling: " + System.nanoTime());
    	executor.schedule(task, 3, TimeUnit.SECONDS);
    }

按固定频率执行任务，与任务执行时长无关。

static void doTestSchedule02() { 
    	ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
    	Runnable task = () -> System.out.println("Scheduling: " + System.currentTimeMillis());
    	int initialDelay = 0;
    	int period = 1;
    	executor.scheduleAtFixedRate(task, initialDelay, period, TimeUnit.SECONDS);
    }

按固定时长执行任务，与任务执行时长有关（如果不执行完第n次任务是永远不会再执行第n+1次任务的）。

static void doTestSchedule03() { 
    	ScheduledExecutorService executor =new ScheduledThreadPoolExecutor(1);
    	Runnable task = () -> { 
    		try { 
    			TimeUnit.SECONDS.sleep(2);
    			System.out.println("Scheduling: " + System.currentTimeMillis());
    		}catch (InterruptedException e) { 
    			System.err.println("task interrupted");
    		}
    	};
    	executor.scheduleWithFixedDelay(task, 0, 1, TimeUnit.SECONDS);
    }

**参考代码：**

package com.cy.pj.juc.lock;
    
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /** * @author archibald * @date 2021/9/4 * @apiNote */
    public class ScheduledExecutorTests { 
        public static void main(String[] args) { 
            //doTestScheduleTask01();
            //doTestScheduleTask02();
            doTestScheduleTask03();
        }
    
        // 周期性的任务调度，第n次任务执行结束以后再延迟delay这么长时间再次执行
        private static void doTestScheduleTask03() { 
            ScheduledThreadPoolExecutor es = new ScheduledThreadPoolExecutor(1);
            es.scheduleWithFixedDelay(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    System.out.println("schedule:" + System.currentTimeMillis());
                }
            },3,1,TimeUnit.SECONDS);
        }
    
        // 周期性任务调度，前后两次任务的开始时间相隔period 这么长时间，任务开始时间与任务执行时长没有关系
        private static void doTestScheduleTask02() { 
            ScheduledThreadPoolExecutor es = new ScheduledThreadPoolExecutor(1);
            es.scheduleAtFixedRate(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("schedule:" + System.currentTimeMillis());
                }
            },3,1,TimeUnit.SECONDS);//3秒以后开始执行，每隔1秒执行一次
        }
    
        // 延迟任务调度
        private static void doTestScheduleTask01() { 
            ScheduledThreadPoolExecutor es = new ScheduledThreadPoolExecutor(1);
            es.schedule(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("schedule:" + System.currentTimeMillis());
                }
            },3,TimeUnit.SECONDS);//3秒以后执行任务
        }
    }

## 五、总结 ##

### 5.1 重难点分析 ###

*  JUC包中常用锁的应用(Lock,ReentrantLock,ReadWriteLock,StampedLock)
 *  JUC包中常用原子类应用（LongAtomic,LongAddr,…）
 *  JUC包中并发工具类应用（CountDownLatch,CyclicBarrier,Semaphore）
 *  JUC包线程池的应用（ThreadPoolExecutor,…）

### 5.2 常见FAQ ###

*  ReentrantLock相对synchronized有什么优势？
 *  StampedLocked对象相对于ReadWriteLock对象有什么优势？
 *  Java中无锁对象的应用原理及可能存在问题。
 *  JUC包中常用的并发工具类有哪些？
 *  ThreadPoolExecutor对象构建时常用参数?
 *  ThreadPoolExecutor中关联的阻塞式队列的作用？
 *  ThreadPoolExecutor任务拒绝处理策略？
 *  ScheduledThreadPoolExecutor对象有什么作用？

### 5.3 BUG分析 ###

*  中断异常(InterruptedException)；
 *  监视器异常(IllegalMonitorStateExceptionThreadPoolExecutor)；
 *  参数异常(IllegalArgumentException)；
 *  无效的线程状态异常(IllegalThreadStateException)。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA57uP55CG77yM5aSp5Y-w6aOO5aW95aSn_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/eb869bb7a25447ebab42a38e90bb350c.png