ReentrantLock

可以替代synchronized

相对于Synchronized的优势

可以使用trylock

  1. public class T03_ReentrantLock3 {
  2.     Lock lock = new ReentrantLock();
  4.     void m1() {
  5.         try {
  6.             lock.lock();
  7.             for (int i = 0; i < 7; i++) {
  8.                 TimeUnit.SECONDS.sleep(1);
  10.                 System.out.println(i);
  11.             }
  12.         } catch (InterruptedException e) {
  13.             e.printStackTrace();
  14.         } finally {
  15.             lock.unlock();
  16.         }
  17.     }
  19.     /**
  20.      * 使用tryLock进行尝试锁定，不管锁定与否，方法都将继续执行
  21.      * 可以根据tryLock的返回值来判定是否锁定
  22.      * 也可以指定tryLock的时间，由于tryLock(time)抛出异常，所以要注意unclock的处理，必须放到finally中
  23.      */
  24.     void m2() {
  25.         /*
  26.         boolean locked = lock.tryLock();
  27.         System.out.println("m2 ..." + locked);
  28.         if(locked) lock.unlock();
  29.         */
  31.         boolean locked = false;
  33.         try {
  34.             locked = lock.tryLock(5, TimeUnit.SECONDS);
  35.             System.out.println("m2 ..." + locked);
  36.         } catch (InterruptedException e) {
  37.             e.printStackTrace();
  38.         } finally {
  39.             if(locked) lock.unlock();
  40.         }
  42.     }
  44.     public static void main(String[] args) {
  45.         T03_ReentrantLock3 rl = new T03_ReentrantLock3();
  46.         new Thread(rl::m1).start();
  47.         try {
  48.             TimeUnit.SECONDS.sleep(1);
  49.         } catch (InterruptedException e) {
  50.             e.printStackTrace();
  51.         }
  52.         new Thread(rl::m2).start();
  53.     }
  54. }

可以打断线程等待 lockInterruptibly

  1. public class T04_ReentrantLock4 {
  3.     public static void main(String[] args) {
  4.         Lock lock = new ReentrantLock();
  7.         Thread t1 = new Thread(()->{
  8.             try {
  9.                 lock.lock();
  10.                 System.out.println("t1 start");
  11.                 TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
  12.                 System.out.println("t1 end");
  13.             } catch (InterruptedException e) {
  14.                 System.out.println("interrupted!");
  15.             } finally {
  16.                 lock.unlock();
  17.             }
  18.         });
  19.         t1.start();
  21.         Thread t2 = new Thread(()->{
  22.             try {
  23.                 //lock.lock();
  24.                 lock.lockInterruptibly(); //可以对interrupt()方法做出响应
  25.                 System.out.println("t2 start");
  26.                 TimeUnit.SECONDS.sleep(5);
  27.                 System.out.println("t2 end");
  28.             } catch (InterruptedException e) {
  29.                 System.out.println("interrupted!");
  30.             } finally {
  31.                 lock.unlock();
  32.             }
  33.         });
  34.         t2.start();
  36.         try {
  37.             TimeUnit.SECONDS.sleep(1);
  38.         } catch (InterruptedException e) {
  39.             e.printStackTrace();
  40.         }
  41.         t1.interrupt(); //打断线程2的等待
  43.     }
  44. }

公平锁

线程获取锁有先后顺序（通过队列实现）

  1. public class T05_ReentrantLock5 extends Thread {
  3.     private static ReentrantLock lock=new ReentrantLock(true); //参数为true表示为公平锁，请对比输出结果
  4.     public void run() {
  5.         for(int i=0; i<100; i++) {
  6.             lock.lock();
  7.             try{
  8.                 System.out.println(Thread.currentThread().getName()+"获得锁");
  9.             }finally{
  10.                 lock.unlock();
  11.             }
  12.         }
  13.     }
  14.     public static void main(String[] args) {
  15.         T05_ReentrantLock5 rl=new T05_ReentrantLock5();
  16.         Thread th1=new Thread(rl);
  17.         Thread th2=new Thread(rl);
  18.         th1.start();
  19.         th2.start();
  20.     }
  21. }

CountDownLatch

  1. public class T06_TestCountDownLatch {
  2.     public static void main(String[] args) {
  3.         usingJoin();
  4.         usingCountDownLatch();
  5.     }
  7.     private static void usingCountDownLatch() {
  8.         Thread[] threads = new Thread[100];
  9.         CountDownLatch latch = new CountDownLatch(threads.length);
  11.         for(int i=0; i<threads.length; i++) {
  12.             threads[i] = new Thread(()->{
  13.                 int result = 0;
  14.                 for(int j=0; j<10000; j++) result += j;
  15.                 latch.countDown(); // threads - 1,为0时  latch.await() 处继续
  16.             });
  17.         }
  19.         for (int i = 0; i < threads.length; i++) {
  20.             threads[i].start();
  21.         }
  23.         try {
  24.             latch.await();
  25.         } catch (InterruptedException e) {
  26.             e.printStackTrace();
  27.         }
  29.         System.out.println("end latch");
  30.     }
  32.     private static void usingJoin() {
  33.         Thread[] threads = new Thread[100];
  35.         for(int i=0; i<threads.length; i++) {
  36.             threads[i] = new Thread(()->{
  37.                 int result = 0;
  38.                 for(int j=0; j<10000; j++) result += j;
  39.             });
  40.         }
  42.         for (int i = 0; i < threads.length; i++) {
  43.             threads[i].start();
  44.         }
  46.         for (int i = 0; i < threads.length; i++) {
  47.             try {
  48.                 threads[i].join();
  49.             } catch (InterruptedException e) {
  50.                 e.printStackTrace();
  51.             }
  52.         }
  54.         System.out.println("end join");
  55.     }
  56. }

CyclicBarrier

  1. public class T07_TestCyclicBarrier {
  2.     public static void main(String[] args) {
  3.         //CyclicBarrier barrier = new CyclicBarrier(20);
  5.         CyclicBarrier barrier = new CyclicBarrier(20, () -> System.out.println("满人"));
  7.         /*CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() {
  8.             @Override
  9.             public void run() {
  10.                 System.out.println("满人，发车");
  11.             }
  12.         });*/
  14.         for(int i=0; i<100; i++) {
  16.                 new Thread(()->{
  17.                     try {
  18.                         barrier.await(); // 阻塞
  19.                         // ... 执行代码
  20.                     } catch (InterruptedException e) {
  21.                         e.printStackTrace();
  22.                     } catch (BrokenBarrierException e) {
  23.                         e.printStackTrace();
  24.                     }
  25.                 }).start();
  27.         }
  28.     }
  29. }
  30. // result
  31. // 满人
  32. // 满人
  33. // 满人
  34. // 满人
  35. // 满人

场景

限流（Guava RateLimiter）

Phaser

  1. public class T09_TestPhaser2 {
  2.     static Random r = new Random();
  3.     static MarriagePhaser phaser = new MarriagePhaser();
  6.     static void milliSleep(int milli) {
  7.         try {
  8.             TimeUnit.MILLISECONDS.sleep(milli);
  9.         } catch (InterruptedException e) {
  10.             e.printStackTrace();
  11.         }
  12.     }
  14.     public static void main(String[] args) {
  16.         phaser.bulkRegister(7);
  18.         for(int i=0; i<5; i++) {
  20.             new Thread(new Person("p" + i)).start();
  21.         }
  23.         new Thread(new Person("新郎")).start();
  24.         new Thread(new Person("新娘")).start();
  26.     }
  30.     static class MarriagePhaser extends Phaser {
  31.         @Override
  32.         protected boolean onAdvance(int phase, int registeredParties) {
  34.             switch (phase) {
  35.                 case 0:
  36.                     System.out.println("所有人到齐了！" + registeredParties);
  37.                     System.out.println();
  38.                     return false;
  39.                 case 1:
  40.                     System.out.println("所有人吃完了！" + registeredParties);
  41.                     System.out.println();
  42.                     return false;
  43.                 case 2:
  44.                     System.out.println("所有人离开了！" + registeredParties);
  45.                     System.out.println();
  46.                     return false;
  47.                 case 3:
  48.                     System.out.println("婚礼结束！新郎新娘抱抱！" + registeredParties);
  49.                     return true;
  50.                 default:
  51.                     return true;
  52.             }
  53.         }
  54.     }
  57.     static class Person implements Runnable {
  58.         String name;
  60.         public Person(String name) {
  61.             this.name = name;
  62.         }
  64.         public void arrive() {
  66.             milliSleep(r.nextInt(1000));
  67.             System.out.printf("%s 到达现场！\n", name);
  68.             phaser.arriveAndAwaitAdvance();
  69.         }
  71.         public void eat() {
  72.             milliSleep(r.nextInt(1000));
  73.             System.out.printf("%s 吃完!\n", name);
  74.             phaser.arriveAndAwaitAdvance();
  75.         }
  77.         public void leave() {
  78.             milliSleep(r.nextInt(1000));
  79.             System.out.printf("%s 离开！\n", name);
  82.             phaser.arriveAndAwaitAdvance();
  83.         }
  85.         private void hug() {
  86.             if(name.equals("新郎") || name.equals("新娘")) {
  87.                 milliSleep(r.nextInt(1000));
  88.                 System.out.printf("%s 洞房！\n", name);
  89.                 phaser.arriveAndAwaitAdvance();
  90.             } else {
  91.                 phaser.arriveAndDeregister();
  92.                 //phaser.register()
  93.             }
  94.         }
  96.         @Override
  97.         public void run() {
  98.             arrive();
  99.             eat();
  100.             leave();
  101.             hug();
  103.         }
  104.     }
  105. }

ReadWriteLock -> StampedLock

共享锁

排它锁

使用

public class Counter {
    private final ReadWriteLock rwlock = new ReentrantReadWriteLock();
    private final Lock rlock = rwlock.readLock();
    private final Lock wlock = rwlock.writeLock();
    private int[] counts = new int[10];
    public void inc(int index) {
        wlock.lock(); // 加写锁
        try {
            counts[index] += 1;
        } finally {
            wlock.unlock(); // 释放写锁
        }
    }
    public int[] get() {
        rlock.lock(); // 加读锁
        try {
            return Arrays.copyOf(counts, counts.length);
        } finally {
            rlock.unlock(); // 释放读锁
        }
    }
}

Semaphore

使用

public class T11_TestSemaphore {
    public static void main(String[] args) {
        //Semaphore s = new Semaphore(2);
        Semaphore s = new Semaphore(2, true);
        //允许一个线程同时执行
        //Semaphore s = new Semaphore(1);

        new Thread(()->{
            try {
                s.acquire(); // 获取许可 number - 1

                System.out.println("T1 running...");
                Thread.sleep(200);
                System.out.println("T1 running...");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                s.release();
            }
        }).start();

        new Thread(()->{
            try {
                s.acquire();

                System.out.println("T2 running...");
                Thread.sleep(200);
                System.out.println("T2 running...");

                s.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

场景

  • 用于多个共享资源的互斥使用.

  • 用于并发线程数的控制.

    Exchanger

    两线程间交换数据 ```java public class T12_TestExchanger {

    static Exchanger exchanger = new Exchanger<>();

    public static void main(String[] args) {

      new Thread(()->{
      String s = "T1";
      try {
          s = exchanger.exchange(s); // 阻塞
      } catch (InterruptedException e) {
          e.printStackTrace();
      }
      System.out.println(Thread.currentThread().getName() + " " + s);

  }, "t1").start();

    new Thread(()->{
        String s = "T2";
        try {
            s = exchanger.exchange(s);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " " + s);

    }, "t2").start();


}

}

<a name="dmmoq"></a>
## LockSupport
```java
public class T13_TestLockSupport {
    public static void main(String[] args) {
        Thread t = new Thread(()->{
            for (int i = 0; i < 10; i++) {
                System.out.println(i);
                if(i == 5) {
                    LockSupport.park();
//                    LockSupport.park();
                }
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        t.start();

        LockSupport.unpark(t);

    }
}