Java并发编程之常用并发工具类

末蓝、 2022-11-13 11:21 304阅读 0赞

# 并发工具类 #

### 文章目录 ###

*  并发工具类
 *   *  一、CountDownLatch
     *   *  1. 概念
         *  2. 代码示例
     *  二、CyclicBarrier
     *   *  1. 概念
         *  2. 代码实现
     *  三、Semaphore
     *   *  1. 概念
         *  2. 代码实现
         *  3. 公平性
     *  四、Exchanger
     *   *  1. 概念
         *  2. 代码实现
     *  五、ReentrantReadWriteLock
     *   *  1. 概念
         *  2. 代码实现
     *  六、StampedLock
     *   *  1. 概念
         *  2. 写锁
         *  3. 乐观读锁
         *  4. 悲观读锁

## 一、CountDownLatch ##

### 1. 概念 ###

*  可以将其理解为一个计数器，创建实例时，构造器中赋予计数器的初始值，实例每调用一次 `countDown()` 计数器的值就减一
 *  实例调用 `await()` ，线程会进入阻塞状态
 *  当计数器的值变成0，会唤醒通过上述方法进入阻塞状态的所有线程，继续向下执行

### 2. 代码示例 ###

1.  多个多线程使用一个计数器
    
        public class CountDownLatchTest1 { 
        
        	public static void main(String[] args) { 
        
        		System.out.println("定义了计数器，初始值为3");
        		final CountDownLatch countDownLatch = new CountDownLatch(3);
        
        		new Thread(new Runnable() { 
        			@Override
        			public void run() { 
        				try { 
        					System.out.println("启动t2线程，调用await方法，阻塞");
        					countDownLatch.await();
        					System.out.println("计数器的值成为0，t2被唤醒，继续运行");
        				} catch (InterruptedException e) { 
        					e.printStackTrace();
        				}
        			}
        		}, "t2").start();
        
        		new Thread(new Runnable() { 
        			@Override
        			public void run() { 
        				try { 
        					System.out.println("启动t3线程，调用await方法，阻塞");
        					countDownLatch.await();
        					System.out.println("计数器的值成为0，t3被唤醒，继续运行");
        				} catch (InterruptedException e) { 
        					e.printStackTrace();
        				}
        			}
        		}, "t3").start();
        
        		new Thread(new Runnable() { 
        			@Override
        			public void run() { 
        				try { 
        					System.out.println("t1线程启动，连续三次将计数器的值减一");
        					countDownLatch.countDown(); //3-1=2
        					countDownLatch.countDown(); //2-1=1
        					countDownLatch.countDown();	//1-1=0
        				} catch (Exception e) { 
        					e.printStackTrace();
        				}
        			}
        		}, "t1").start();
        	}
        }
    
    运行结果：
    
    ![20210326163637632.png][]
2.  也可以定义多个计数器，各个计数器的阻塞和唤醒各不相干

## 二、CyclicBarrier ##

### 1. 概念 ###

*  设置一个公共屏障点，所有线程在屏障点互相等待，所有线程都准备好后，所有线程同时运行
 *  可以循环使用，如下图
    
    1 - 5 表示5个线程，所有线程在屏障点互相等待，全部准备好后，同时启动运行
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70][]

### 2. 代码实现 ###

*  创建实例 `CyclicBarrier barrier = new CyclicBarrier(3);` 构造器参数中声明屏障点的线程数
    
     *  如果等待的线程数少于构造器中设定的初始值，则程序会一直等待下去，无法运行
 *  实例调用 `barrier.await();` 表示此线程在所有线程未到达屏障点之前将一直阻塞
    
     *  最后一个线程调用此方法时，所有阻塞的线程都会立即启动运行
 *  实例调用 `barrier.await(1,TimeUnit.SECONDS);` 表示此线程等待了1秒钟，但是还有线程没有到达屏障点，则此线程会自己先运行，抛出 TimeoutException 异常，导致Barrier被标记为一个已经破坏的Barrier
    
     *  当其他线程调用 `await()` 时发现屏障点已经被破环，抛出 BrokenBarrierException 异常

//所有的子任务都完成时，才执行主任务，这个时候就可以选择使用CyclicBarrier
    
    public class CyclicBarrierTest2 { 
    
    	public static void main(String[] args) throws IOException, InterruptedException, BrokenBarrierException { 
    		
    		// 需要5个等待线程
    		CyclicBarrier barrier = new CyclicBarrier(5);
    
    		// 创建4个等待线程 + 1个main线程
    		new Thread(new Worker(barrier, "worker1")).start();
    		new Thread(new Worker(barrier, "worker2")).start();
    		new Thread(new Worker(barrier, "worker3")).start();
    		new Thread(new Worker(barrier, "worker4")).start();
    
    		System.out.println("main线程等待中");
    		barrier.await();
    		System.out.println("所有的线程都工作完毕了, main线程继续执行!");
    	}
    }
    
    // 要等待的线程
    class Worker implements Runnable { 
    	
    	private final CyclicBarrier barrier;
    	
    	//线程的名字
    	private final String name;
    
    	//构造器
    	public Worker(CyclicBarrier barrier, String name) { 
    		this.barrier = barrier;
    		this.name = name;
    	}
    
    	@Override
    	public void run() { 
    		try { 
    			Thread.sleep(1000);
    			System.out.println(name + "等待中");
    			barrier.await();
    		} catch (InterruptedException | BrokenBarrierException e) { 
    			e.printStackTrace();
    		}
    	}
    }

运行结果：

![20210326163702284.png][]

## 三、Semaphore ##

### 1. 概念 ###

*  Semaphore一个计数信号量，信号量维护了一个许可证集合
 *  通过 `acquire()` 和 `release()` 来获取和释放访问许可证
 *  线程只有获取了许可证才能执行，如果获取不到（许可证集合中的许可证已全部被别的线程获取）则阻塞
 *  线程释放了许可证，其他线程才可以获取

### 2. 代码实现 ###

*  创建实例 `Semaphore semp = new Semaphore(1);` 构造器的参数是许可证的数量
 *  获取许可证 `semp.acquire();`
 *  释放许可证 `semp.release();`

public class SemaphoreTest1 { 
    
    	public static void main(String[] args) { 
    
    		// 创建线程池
    		ExecutorService exec = Executors.newCachedThreadPool();
    
    		// 创建Semaphore实例
    		// 最多同时只能有两个线程获取到许可证
    		final Semaphore semp = new Semaphore(2);
    
    		// 创建三个线程获取许可证
    		for (int index = 0; index < 3; index++) { 
    			Runnable run = new Runnable() { 
    				public void run() { 
    					try { 
    						System.out.println(Thread.currentThread().getName() + "尝试获取许可证");
    
    						// 获取许可证
    						semp.acquire();
    
    						System.out.println(Thread.currentThread().getName() + "成功获取许可证");
    						Thread.sleep(1000);
    						System.out.println(Thread.currentThread().getName() + "释放许可证");
    
    						// 释放许可证
    						semp.release();
    					} catch (InterruptedException e) { 
    					}
    				}
    			};
    			exec.execute(run);
    		}
    		
    		// 退出线程池
    		exec.shutdown();
    	}
    }

运行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 1][]

### 3. 公平性 ###

*  观察上述运行结果可知，1最先尝试获取许可证，但最先成功获取许可证的并不是1，表明默认是非公平的
 *  公平性指的是谁最先尝试获取许可证，谁就应该先获取到许可证
 *  强制公平会影响到并发性能，所以除非确实需要它否则不要启用它
 *  创建公平的实例 `final Semaphore semp = new Semaphore(2, true);`

## 四、Exchanger ##

### 1. 概念 ###

*  此类用于交换两个线程的数据（必须是两个线程）
 *  两个线程必须都到达汇合点才可以交换，若有未到达的，则到达的需要等待

### 2. 代码实现 ###

*  创建实例 `Exchanger<T> exchanger = new Exchanger<>();` 泛型中声明要交换的数据的类型
 *  两个线程同时调用 `exchanger.exchange(T t)` 才可以交换数据，一个先调用则必须等待另一个线程调用
    
     *  返回值是和另一个线程交换后的数据

public class ExchangerTest1 { 
    
    	public static void main(String[] args) { 
    
    		//创建实例，两个线程必须使用同一个交换器
    		Exchanger<String> exchanger = new Exchanger<String>();
    
    		ExchangerRunnable exchangerRunnable1 = new ExchangerRunnable(exchanger, "A");
    		ExchangerRunnable exchangerRunnable2 = new ExchangerRunnable(exchanger, "B");
    		
    		new Thread(exchangerRunnable1).start();
    		new Thread(exchangerRunnable2).start();
    	}
    }
    
    class ExchangerRunnable implements Runnable { 
    
    	Exchanger<String> exchanger = null;
    
    	String object = null; //每个线程要交换的数据
    
    	public ExchangerRunnable(Exchanger<String> exchanger, String object) { 
    		this.exchanger = exchanger;
    		this.object = object;
    	}
    
    	@Override
    	public void run() { 
    		try { 
    			Object previous = this.object; //获取到本线程要交换的数据
    			System.out.println(Thread.currentThread().getName() +"交换前的数据：" + this.object);
    
    			//保证不同线程到达交汇点的时间不同
    			if("C".equals(previous)){ 
    				Thread.sleep(1000);
    				System.out.println(Thread.currentThread().getName() +"对数据C的处理耗时1s");
    			}else if("D".equals(previous)){ 
    				Thread.sleep(2000);
    				System.out.println(Thread.currentThread().getName() +"对数据D的处理耗时2s");
    			}else if("A".equals(previous)){ 
    				Thread.sleep(3000);
    				System.out.println(Thread.currentThread().getName() +"对数据A的处理耗时3s");
    			}else if("B".equals(previous)){ 
    				Thread.sleep(4000);
    				System.out.println(Thread.currentThread().getName() +"对数据B的处理耗时4s");
    			}
    
    			// 两个对象必须在此处汇合，只有一个线程调用change方法是不会进行数据交换的
    			this.object = this.exchanger.exchange(this.object);
    			System.out.println(Thread.currentThread().getName() + " 交换后: " + this.object);
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    	}
    }

运行结果：

![20210326163728373.png][]

## 五、ReentrantReadWriteLock ##

### 1. 概念 ###

*  ReentrantReadWriteLock称为读写锁（写锁 + 悲观读锁），采用读写加锁分离的方式
 *  高并发下读多写少时性能优于ReentrantLock
 *  读读共享，写写互斥，读写互斥
    
     *  共享指的共享锁，互斥指的是排他锁
 *  ReentrantReadWriteLock写锁是可重入锁

### 2. 代码实现 ###

*  创建实例 `ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();`
 *  创建读锁 `ReadLock readLock = rwLock.readLock();`
 *  创建写锁 `WriteLock writeLock = rwLock.writeLock();`
 *  读写锁都是通过调用 `lock()` 和 `unlock()` 进行加锁解锁

public class ReadWriteLockTest1 { 
    
    	private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    	private final ReadLock readLock = rwLock.readLock();
    	private final WriteLock writeLock = rwLock.writeLock();
    
    	//读
    	public void read() { 
    		try { 
    			readLock.lock();
    			System.out.println("当前线程:" + Thread.currentThread().getName() + "read进入...");
    			Thread.sleep(3000);
    		} catch (Exception e) { 
    			e.printStackTrace();
    		} finally { 
                System.out.println("当前线程:" + Thread.currentThread().getName() + "read退出...");
    			readLock.unlock();
    		}
    	}
    
    	//写
    	public void write() { 
    		try { 
    			writeLock.lock();
    			System.out.println("当前线程:" + Thread.currentThread().getName() + "write进入...");
    			Thread.sleep(3000);
    		} catch (Exception e) { 
    			e.printStackTrace();
    		} finally { 
                System.out.println("当前线程:" + Thread.currentThread().getName() + "write退出...");
    			writeLock.unlock();
    		}
    	}
    
    	public static void main(String[] args) { 
    
    		final ReadWriteLockTest1 urrw = new ReadWriteLockTest1();
    
    		Thread t1 = new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				urrw.read();
    			}
    		}, "t1");
    		Thread t2 = new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				urrw.read();
    			}
    		}, "t2");
    		Thread t3 = new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				urrw.write();
    			}
    		}, "t3");
    
    		//读读模式
    		t1.start(); // Read
    		t2.start(); // Read
    	}
    }

运行结果：

![20210326163740524.png][]

t2无需等待t1释放锁才执行

*  将读读模式修改为读写模式

//读写模式
    t1.start(); // Read
    t3.start(); // Write

运行结果：

![20210326163749405.png][]

## 六、StampedLock ##

### 1. 概念 ###

*  StampedLock支持三种锁：写锁、悲观读锁、乐观读锁，相比ReentrantReadWriteLock多了乐观读锁
 *  StampedLock中引入了一个stamp（邮戳）的概念，它代表线程获取到锁的版本，每次获得锁之后都会得到一个新的long值
 *  写锁是排它锁（独占锁），相同时间只能有一个线程获取写锁，其他线程请求 读锁和写锁 都会被阻塞
    
    ![20210326163806830.png][]
 *  StampedLock的写锁是不可重入锁
    
     *  ReentrantLock锁是可重入锁
     *  ReentrantReadWriteLock的写锁是可重入锁

### 2. 写锁 ###

*  创建实例 `StampedLock lock = new StampedLock();`
 *  获取写锁 `long stamp = lock.writeLock();`
 *  释放写锁 `lock.unlockWrite(stamp);` 参数需要传递要释放锁的long值
 *  获取带超时时间的写锁 `lock.tryWriteLock` 如果没有获得写锁，返回值为0，不阻塞，向下运行

**情况一**：StampedLock写锁的正常使用

//创建StampedLock对象
    StampedLock sl = new StampedLock();
    //获取写锁，并且返回long型的stamp
    long stamp = sl.writeLock();
    System.out.println("get write lock,stamp=" + stamp);
    //使用完毕，释放锁，要传入对应的stamp
    sl.unlockWrite(stamp);
    //再次获取写锁
    stamp = sl.writeLock();
    System.out.println("get write lock,stamp=" + stamp);
    System.out.println("每次获取的锁都是新的long值");
    //释放写锁
    sl.unlockWrite(stamp);
    
    //尽量使用try - catch结构，保证释放锁成功

运行结果：

![20210326163821805.png][]

**情况二：体现非重入锁**

StampedLock sl = new StampedLock();
    long stamp = sl.writeLock();
    System.out.println("get write lock,stamp="+stamp);
    
    //当前线程并未释放锁，再次尝试获取写锁
    stamp = sl.writeLock();
    System.out.println("get write lock,stamp="+stamp);
    sl.unlockWrite(stamp);
    
    //线程阻塞，无法第二次获得锁，非可重入

**情况三：带超时时间的锁**

StampedLock sl1 = new StampedLock();
    long stamp1 = sl1.tryWriteLock();
    System.out.println("get StampedLock.tryWriteLock lock1,stamp="+stamp1);
    
    long stamp2 = sl1.tryWriteLock(); //一定会获取失败，返回值为0
    System.out.println("can not get StampedLock.tryWriteLock lock1,stamp="+stamp2);
    
    sl1.unlockWrite(stamp1);

运行结果：

![20210326163909115.png][]

### 3. 乐观读锁 ###

*  创建实例 `StampedLock lock = new StampedLock();`
 *  获取乐观读锁 `long stamp = lock.tryOptimisticRead();`
    
     *  乐观锁认为数据不会轻易的被修改，因此在操作数据前并没有加锁
     *  采用试探的方式，只要当前资源没有写锁就可以获得一个非0的stamp，如果已经存在写锁则返回0，0无法通过validate的验证
     *  乐观读锁本质上没有锁，写锁和悲观读锁会加锁
     *  由于没有真正的加锁，所以并发性能要比悲观读锁高
 *  检查乐观读锁有没有被修改过 `boolean validate = lock.validate(stamp);`
    
     *  验证从获取锁开始是否发生过变化，即有没有加过写锁（写操作会改变数据），加了读锁，不影响
     *  如果这期间锁发生变化则validate返回false，否则返回true
 *  缺点
    
     *  由于没有使用真正的锁，如果数据中途被修改，就会造成数据不一致问题
     *  StampLock使用快照的方式，需要复制一份要操作的变量到方法栈，操作的数据只是一个快照，读线程可能使用的数据不是最新的
     *  validate验证通过之后变量仍然可能还被修改
 *  tryOptimisticRead与validate一定要紧紧挨着使用，否则在获取和验证之间很可能数据被修改
 *  乐观读锁原理
    
     *  先尝试获取乐观读锁，这时候没有写锁就拿到了锁，在真正要使用的时候再验证一下是否发生了改变，如果没有发生改变就可以安心使用

public class StampedLockDemo3 { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            // 创建实例
            StampedLock sl = new StampedLock();
    
            // 获取乐观读锁
            long stamp = sl.tryOptimisticRead();
    
            System.out.println(Thread.currentThread().getName()+" tryOptimisticRead stamp=" + stamp);
    
            // 启动一个新线程
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    long stamp131 = sl.writeLock(); //获取写锁
                    sl.unlockWrite(stamp131); //释放写锁
                }
            }).start();
    
            Thread.sleep(3000);
    
            // 验证
            boolean validate = sl.validate(stamp); //false
    
            // 再次获取一个乐观读锁
            stamp = sl.tryOptimisticRead();
            boolean validate1 = sl.validate(stamp); //true
    
            if(sl.validate(stamp)){ 
                //不安全，验证通过之后仍然可以添加写锁
            }
        }
    }

### 4. 悲观读锁 ###

*  悲观锁认为数据是极有可能被修改的，所以在使用数据之前都需要先加锁，锁未释放之前如果有其他线程想要修改数据（加写锁）就必须阻塞它
 *  悲观读锁是一个共享锁，没有线程占用写锁的情况下，多个线程可以同时获取读锁font>；如果其他线程已经获得了写锁，则获取读锁和写锁都会失败（写锁是排他锁）
    
    ![20210326163924401.png][]
    
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 2][]
 *  读锁可以多次获取（没有写锁占用的情况下），写锁必须在读锁全部释放之后才能获取写锁
 *  在多个线程之间依然存在写写互斥、读写互斥、读读共享的关系
 *  代码实现
    
     *  创建实例 `StampedLock lock = new StampedLock();`
     *  获取悲观读锁 `long stamp1 = lock.readLock();`
     *  释放悲观读锁 `lock.unlockRead(stamp1);`

public class StampedLockDemo2 { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            StampedLock sl2 = new StampedLock();
            long stamp12 = sl2.readLock();
            System.out.println(Thread.currentThread().getName()+" get read lock1,stamp=" + stamp12);
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(Thread.currentThread().getName()+" run");
                    long stamp121 = sl2.writeLock(); //新线程要获取写锁，读锁未释放，必然失败
                    System.out.println(Thread.currentThread().getName()+" get write lock1,stamp="+stamp121);
                }
            }).start();
    
            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + " get unlock read lock1,stamp="+stamp12);
            sl2.unlockRead(stamp12);
        }
    }

运行结果：

![20210326163948201.png][]

[20210326163637632.png]: /images/20221022/65ed598b663b44c09f499f351b1d048f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210326163648223.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[20210326163702284.png]: https://img-blog.csdnimg.cn/20210326163702284.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210326163713429.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[20210326163728373.png]: https://img-blog.csdnimg.cn/20210326163728373.png
[20210326163740524.png]: https://img-blog.csdnimg.cn/20210326163740524.png
[20210326163749405.png]: https://img-blog.csdnimg.cn/20210326163749405.png
[20210326163806830.png]: https://img-blog.csdnimg.cn/20210326163806830.png
[20210326163821805.png]: https://img-blog.csdnimg.cn/20210326163821805.png
[20210326163909115.png]: https://img-blog.csdnimg.cn/20210326163909115.png
[20210326163924401.png]: https://img-blog.csdnimg.cn/20210326163924401.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210326163933269.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80OTM0MzE5MA==,size_16,color_FFFFFF,t_70
[20210326163948201.png]: https://img-blog.csdnimg.cn/20210326163948201.png