第四節、并發工具類 · 小機的專屬后端開發筆記本

# lock接口為鎖和等待條件提供一個框架的接口和類，它不同于內置的synchronized同步和monitor監視器。 Lock不是Java語言內置的，是在java.util工具包里面的。需要開發人員手動加鎖和解鎖。   ## ReentrantLock 可重入鎖：同一個線程的不同被鎖住的方法可以直接進入。synchronized也是可重入鎖，ReentrantLock可以用來替代synchronized，但是需要手動解鎖。 ~~~ ?ReentrantLock lock = new ReentrantLock(); ?? ?public void method() { ? ? ?// 上鎖 ? lock.lock(); ? ? ?/*-------------------- ? ? ?* 業務操作 ? ? ?*/ ? ? ?// 解鎖 ? ? ?// 一定要寫在finally代碼塊中保證會被釋放掉 ? ? ?lock.unlock(); ?} ~~~ 1. tryLock()，在n秒時間內嘗試獲取鎖對象，不管獲取與否方法都會繼續執行，但是可以根據該方法的返回值來判斷是否獲取到鎖了。 2. lock.lockInterruptibly()，可以被thread.interrupt()打斷。 3. new ReetrantLock(true)，公平鎖。線程會在等待隊列中排隊等待。如果是非公平的線程會直接搶占鎖。多個線程之間會交替執行，而不會讓一個線程獨占鎖。   ## CounDownLatch 當主線程需要在子線程的工作完成后才能繼續運行時可以使用這個類進行阻塞。 ~~~ ?CountDownLatch countDownLatch = new CountDownLathc(2); // 建立2個線程空間 ?new Thread(() -> { ? // 處理 ? countDownLatch.countDown(); // 值減一 ?}); ?countDownLatch.await(); // 阻塞主線程 ~~~ 當countDown的值減為0的時候主線程可以繼續往下運行，否則的話阻塞，等待子線程運行完成。比thread.join()靈活。   ### CyclicBarrier 當線程數達到設定的值的時候就會運行設定的代碼。 ~~~ ?package net.smrobot._thread.day04; ?import java.util.concurrent.BrokenBarrierException; ?import java.util.concurrent.CyclicBarrier; ?public class T07_CyclicBarrierTest { ? ? ?public static void main(String[] args) { ? ? ? ? ?// 滿20人之后調用第二個參數指定的動作 ? ? ? ? ?CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() { ? ? ? ? ? ? ?@Override ? ? ? ? ? ? ?public void run() { ? ? ? ? ? ? ? ? ?System.out.println("barrier start"); ? ? ? ? ? ? } ? ? ? ? }); ? ? ? ? ?for (int i = 0; i < 100; i++) { ? ? ? ? ? ? ?new Thread(() -> { ? ? ? ? ? ? ? ? ?try { ? ? ? ? ? ? ? ? ? ? ?// 等待，每20個線程才會被放行 ? ? ? ? ? ? ? ? ? ? ?barrier.await(); ? ? ? ? ? ? ? ? ? ? ?System.out.println("11"); ? ? ? ? ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? ? ? } catch (BrokenBarrierException e) { ? ? ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? ? ? } ? ? ? ? ? ? }).start(); ? ? ? ? } ? ? } ?} ?? ~~~ 可以用在流量限流或者是有個線程要等其他線程結束了才能運行。 > 備注：限流類：Guava RateLimiter   ### Phaser 分解段運行，在遺傳算法中可以用上。每個階段都會設置棧欄。   ### ReadWriteLock 讀寫鎖，當有讀線程在讀的時候其他線程也可以讀數據，寫線程等待，而寫線程是互斥的。 ~~~ ?package net.smrobot._thread.day04; ?? ?? ?import java.util.Random; ?import java.util.concurrent.locks.Lock; ?import java.util.concurrent.locks.ReadWriteLock; ?import java.util.concurrent.locks.ReentrantLock; ?import java.util.concurrent.locks.ReentrantReadWriteLock; ?? ?public class T09_ReadWriteLockTest { ? ? ?static Lock lock = new ReentrantLock(); ? ? ?private static int value; ?? ? ? ?static ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); ? ? ?static Lock readLock = readWriteLock.readLock(); ? ? ?static Lock writeLock = readWriteLock.writeLock(); ?? ? ? ?public static void main(String[] args) { ? ? ? ? ?// 使用互斥鎖 ?// ? ? ? Runnable readR = () -> read(lock); ?// ? ? ? Runnable writeR = () -> write(lock, new Random().nextInt()); ? ? ? ? ?// 使用讀寫鎖 ? ? ? ? ?Runnable readR = () -> read(readLock); ? ? ? ? ?Runnable writeR = () -> write(writeLock, new Random().nextInt()); ?? ? ? ? ? ?for (int i = 0; i < 10; i++) { ? ? ? ? ? ? ?new Thread(readR).start(); ? ? ? ? } ? ? ? ? ?for (int i = 0; i < 2; i++) { ? ? ? ? ? ? ?new Thread(writeR).start(); ? ? ? ? } ?? ?? ? ? } ?? ? ? ?public static void read(Lock lock) { ? ? ? ? ?try { ? ? ? ? ? ? ?lock.lock(); ? ? ? ? ? ? ?Thread.sleep(1000); ? ? ? ? ? ? ?System.out.println("read over!"); ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? } finally { ? ? ? ? ? ? ?lock.unlock(); ? ? ? ? } ?? ? ? } ?? ? ? ?public static void write(Lock lock, int value) { ? ? ? ? ?try { ? ? ? ? ? ? ?lock.lock(); ? ? ? ? ? ? ?Thread.sleep(1000); ? ? ? ? ? ? ?value = value; ? ? ? ? ? ? ?System.out.println("write over!"); ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? } finally { ? ? ? ? ? ? ?lock.unlock(); ? ? ? ? } ? ? } ?} ?? ~~~ `StampedLock`   ### semaphore 信號量，可以用來限流，可以同時設置多少個線程運行。 ~~~ ?package net.smrobot._thread.day04; ?? ?import java.util.concurrent.Semaphore; ?? ?public class T10_SemaphoreTest { ?? ? ? ?public static void main(String[] args) { ? ? ? ? ?Semaphore s = new Semaphore(2); ? ? ? ? ?// 公平鎖 ? ? ?//Semaphore s = new Semaphore(2, true); ? ? ? ? ?new Thread(() -> { ? ? ? ? ? ? ?try { ? ? ? ? ? ? ? ? ?// 信號量中獲取鎖 ? ? ? ? ? ? ? ? ?s.acquire(); ? ? ? ? ? ? ? ? ?System.out.println("t1 running ..."); ? ? ? ? ? ? ? ? ?Thread.sleep(1000); ? ? ? ? ? ? ? ? ?System.out.println("t1 running ...."); ? ? ? ? ? ? ? ? ?// 釋放鎖 ? ? ? ? ? ? ? ? ?s.release(); ? ? ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? } ? ? ? ? }).start(); ?? ? ? ? ? ?new Thread(() -> { ? ? ? ? ? ? ?try { ? ? ? ? ? ? ? ? ?s.acquire(); ? ? ? ? ? ? ? ? ?System.out.println("t2 running ..."); ? ? ? ? ? ? ? ? ?Thread.sleep(1000); ? ? ? ? ? ? ? ? ?System.out.println("t2 running ...."); ? ? ? ? ? ? ? ? ?s.release(); ? ? ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? } ? ? ? ? }).start(); ?? ? ? } ?} ?? ~~~   ### Exchanger 用于線程之間通信交換數據，只能用于兩個線程之間交換數據，兩個線程同時使用exchanger。 ~~~ ?package net.smrobot._thread.day04; ?? ?import java.util.concurrent.Exchanger; ?? ?public class T11_ExchangerTest { ? ? ?static Exchanger<String> exchanger = new Exchanger<>(); ? ? ?public static void main(String[] args) { ?? ? ? ? ? ?new Thread(() -> { ? ? ? ? ? ? ?String s1 = "t1"; ? ? ? ? ? ? ?try { ? ? ? ? ? ? ? ? ?s1 = exchanger.exchange(s1); ? ? ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? } ? ? ? ? ? ? ?System.out.println(Thread.currentThread().getName() + s1); ? ? ? ? }, "T1").start(); ?? ? ? ? ? ?new Thread(() -> { ? ? ? ? ? ? ?String s2 = "t2"; ? ? ? ? ? ? ?try { ? ? ? ? ? ? ? ? ?s2 = exchanger.exchange(s2); ? ? ? ? ? ? } catch (InterruptedException e) { ? ? ? ? ? ? ? ? ?e.printStackTrace(); ? ? ? ? ? ? } ? ? ? ? ? ? ?System.out.println(Thread.currentThread().getName() + s2); ? ? ? ? }, "T2").start(); ?? ? ? } ?} ?? ~~~ ### LockSupport 直接使用里面的靜態方法，可以暫停某個線程運行與重啟。 ~~~ ?// 暫停線程的運行 ?LockSupport.park(); ?? ?// 喚醒線程 ?LockSupport.unpark(thread); ~~~ 可以先調用unpark(t)，此時park()不起作用。