多线程与高并发3 各种锁

（1）ReentrantLock

• 可重入
• 可通过构造参数设置时公平锁还是非公平锁
• 需要明文释放锁，而synchronized是自动释放的
• 可响应中断
• 可在获取锁是设置超时时间
• 通知队列

可重入：

package day03;

import java.util.concurrent.TimeUnit;

/**
 * @author: zdc
 * @date: 2020-03-25
 */
//可重入
public class _1ReentrantLock {
    synchronized void m1(){
        for (int i = 0; i < 10; i++) {
            System.out.println(i);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if(i==5)
                m2();
        }
    }

    private synchronized void m2() {
        System.out.println("m2.............");
    }

    public static void main(String[] args) {
        new _1ReentrantLock().m1();
    }
}

需手动释放锁：

package day03;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: zdc
 * @date: 2020-03-25
 */
public class _2Reentrantlock {
    ReentrantLock lock = new ReentrantLock();

    void m1() {
        lock.lock();
        try {
            for (int i = 0; i < 10; i++) {
                System.out.println(Thread.currentThread().getName() + "--" + i);
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //执行完后 必须手动释放锁
            lock.unlock();
        }
    }
        

    public static void main(String[] args) {
        _2Reentrantlock demo = new _2Reentrantlock();
        for (int i = 0; i < 20; i++) {
            new Thread(demo::m1, "" + i).start();
        }
    }
}

tryLock尝试锁定：

package day03;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: zdc
 * @date: 2020-03-25
 */
//尝试锁定
public class _3Reentrantlock {
    private ReentrantLock lock = new ReentrantLock();
    void m1(){
        lock.lock();
        System.out.println("m1....start");
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("m1....end");
        lock.unlock();
    }
   private boolean locked = false;
    void m2(){
        try {
            //tryLock会返回个标记
          locked =lock.tryLock(4,TimeUnit.SECONDS);
            System.out.println("m2-"+locked);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            if(locked)
                lock.unlock();
            System.out.println("m2结束");
        }
    }

    public static void main(String[] args) {
        _3Reentrantlock demo = new _3Reentrantlock();
        new Thread(demo::m1).start();
        new Thread(demo::m2).start();
    }
}

lock.interrupt() 

（2）CountDownLatch 

CountDownLatch的API还是很简单的，主要就是countDown和await两个方法。CountDownLatch实例化时将count设置为AQS的state，每次countDown时CAS将state设置为state - 1，await时首先会检查当前state是否为0，如果为0则代表所有的任务完成了，await结束，否则主线程将循环重试，直到线程被中断或者任务完成或者等待超时。

 

package day03;

import java.util.concurrent.CountDownLatch;

/**
 * @author: zdc
 * @date: 2020-03-26
 */
public class _4CountDownLatch {
    public static void main(String[] args) {
        int number=100;
        Thread[] threads = new Thread[100];
        CountDownLatch latch = new CountDownLatch(threads.length-20);
        for (int i = 0; i < threads.length; i++) {
            threads[i]=new Thread(()->{
                System.out.println("hehe");
                latch.countDown();
            });
        }
        for (int i = 0; i < threads.length; i++) {
            threads[i].start();
        }
        try {
            latch.await(); //当countDown减到0后释放锁 后续可以执行
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("over");
    }
}

（3）CyclicBarrier  栅栏   它的作用就是会让所有线程都等待完成后才会继续下一步行动。 https://www.jianshu.com/p/bdf47236fc3a

https://www.jianshu.com/p/333fd8faa56e

注意CountDownLatch中任务执行线程调用完了countDown方法之后就会退出，不会等待最后一个任务完成才会退出，这也是CountDownLatch和CyclicBarrier的一个区别

CyclicBarrier较CountDownLatch而言主要多了两个功能：

1. 支持重置状态，达到循环利用的目的。这也是Cyclic的由来。CyclicBarrier中有一个内部类Generation，代表当前的同步处于哪一个阶段。当最后一个任务完成，执行任务的线程会通过nextGeneration方法来重置Generation。也可以通过CyclicBarrier的reset方法来重置Generation。
2. 支持barrierCommand，当最后一个任务运行完成，执行任务的线程会检查CyclicBarrier的barrierCommand是否为null，如果不为null，则运行该任务。

public class CyclicBarrierDemo {

    static class TaskThread extends Thread {
        
        CyclicBarrier barrier;
        
        public TaskThread(CyclicBarrier barrier) {
            this.barrier = barrier;
        }
        
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                System.out.println(getName() + " 到达栅栏 A");
                barrier.await();
                System.out.println(getName() + " 冲破栅栏 A");
                
                Thread.sleep(2000);
                System.out.println(getName() + " 到达栅栏 B");
                barrier.await();
                System.out.println(getName() + " 冲破栅栏 B");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) {
        int threadNum = 5;
        CyclicBarrier barrier = new CyclicBarrier(threadNum, new Runnable() {
            
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + " 完成最后任务");
            }
        });
        
        for(int i = 0; i < threadNum; i++) {
            new TaskThread(barrier).start();
        }
    }
    
}

（4）Phaser

是CountDownLatch和CyclicBarrier的综合体，是栅栏，但不是循环的，而是分阶段的，每个阶段都有不同的线程可以走，但有的线程到了某个阶段就停止了。每个阶段可以有不同数量的线程等待前进到另一个阶段。线程通过调用 arriAndAwaitAdvance() 来阻止它到达屏障，这是一种阻塞方法。当数量到达等于注册的数量时，程序的执行将继续，并且数量将增加。当线程完成其工作时，我们应该调用arrivalAndDeregister（）方法来表示在此特定阶段不再考虑当前线程。

https://blog.csdn.net/u010739551/article/details/51083004

public class _6Phaser {
    static Phaser phaser = new MarriagePhaser();

    static class MarriagePhaser extends Phaser { //实现phaser接口
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            switch (phase) {//四个阶段
                case 0:
                    allArrive(registeredParties);
                    return false;
                case 1:
                    allEat(registeredParties);
                    return false;
                case 2:
                    allLeave(registeredParties);
                    return false;
                case 3:
                    over(registeredParties);
                    return true;
                default:
                    return true;
            }
        }

        private boolean over(int registeredParties) {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("步入洞房后结束" + registeredParties);
            return true;
        }

        private boolean allLeave(int registeredParties) {
            System.out.println("所有人离开" + registeredParties);
            return false;
        }

        private boolean allEat(int registeredParties) {
            System.out.println("所有人吃饭" + registeredParties);
            return false;
        }

        private boolean allArrive(int registeredParties) {
            System.out.println("所有人到齐" + registeredParties);
            return false;
        }
    }

    static class Person implements Runnable {
        private String name;

        public Person(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            phaser.arriveAndAwaitAdvance(); //等
            phaser.arriveAndAwaitAdvance();
            phaser.arriveAndAwaitAdvance();
            if (name.equals("新郎") || name.equals("新娘")) {
                System.out.println(Thread.currentThread().getName() + "开始步入洞房");
                phaser.arriveAndAwaitAdvance();//
            } else {
                phaser.arriveAndDeregister();//不是新郎新娘的 下车
            }
        }
    }

    public static void main(String[] args) {
        phaser.bulkRegister(7);//7个人
        for (int i = 0; i < 5; i++) {
            new Thread(new Person("zzz" + i)).start();
        }
        new Thread(new Person("新娘"), "新郎").start();
        new Thread(new Person("新郎"), "新娘").start();
    }
}

（5）ReadWriteLock 读写锁

ReadWriteLock是JDK5中提供的读写分离锁。读写分离锁可以有效地帮助减少锁竞争，以提高系统性能。用锁分离的机制来提升性能非常容易理解，比如线程A1,A2,A3进行写操作，B1,B2,B3进行读操作，如果使用重入锁或者内部锁（synchronized）则论路上说所有读之间、读与写之间、写与写之间都是穿行操作的。当B1进行读取时，B2、B3则需要等待锁。由于读操作并不对数据的完整性造成破坏，这种等待显然是不合理的。因此，读写锁就有了发挥功能的余地。

在这种情况下，读写锁允许多个线程同时读，使得B1,B2,B3之间真正并行。但是，考虑都数据完整性，写写操作和读写操作间依然时需要相互等待和持有锁的。总的来说，读写锁的访问约束如下表：

 

当读线程来时加一把锁允许其他读线程可以读，但写线程不可以进。写线程时把整个线程全部锁定。

对于读写锁访问约束如下：
（1）读-读不互斥：读读之间不阻塞
（2）读-写互斥：读阻塞写，写也会阻塞读
（3）写-写互斥：写写阻塞

public class _7ReadLock {
    public static void read(Lock lock){
        try {
            lock.lock();
            TimeUnit.SECONDS.sleep(1);
            System.out.println("read over");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
    public static void write(Lock lock,String something){
        try {
            lock.lock();
            TimeUnit.SECONDS.sleep(1);
            System.out.println("write over");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ReentrantLock lock1 = new ReentrantLock();

 /*       Runnable read = ()->read(lock1);  //ReentrantLock 读线程只能一个个读
        Runnable write = ()->write(lock1,"something to write");
*/
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        Runnable read = ()->read(readWriteLock.readLock());  //ReentrantLock 读线程只能一个个读 而readWriteLock.readLock()是共享锁，可以并行读
        Runnable write = ()->write(readWriteLock.writeLock(),"something to write");


        for (int i = 0; i < 20; i++) {
            new Thread(read).start();
        }
        for (int i = 0; i < 2; i++) {
            new Thread(write).start();
        }

    }
}

（6）Semaphore

限流 

void acquire():从此信号量获取一个许可前线程将一直阻塞。相当于一辆车占了一个车位。

 void release():释放一个许可，将其返回给信号量。就如同车开走返回一个车位

public class _8Semaphore {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();

        Semaphore semaphore = new Semaphore(3);//只允许三个线程同时执行

        for (int i = 0; i < 10; i++) {
            final int num = i;
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        semaphore.acquire();//获取一个许可  初始有三个  当用完后，必须等着
                        System.out.println("num="+num);
                        TimeUnit.SECONDS.sleep((int)(Math.random()*10+1)); //失眠 便于分析线程时 三个三个一组的
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        semaphore.release();//释放一个许可
                    }
                }
            });
        }
    }
}

（7）exchange

java.util.concurrent包中的Exchanger类可用于两个线程之间交换信息。可简单地将Exchanger对象理解为一个包含两个格子的容器，通过exchanger方法可以向两个格子中填充信息。当两个格子中的均被填充时，该对象会自动将两个格子的信息交换，然后返回给线程，从而实现两个线程的信息交换。

public class _9Exchanger {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();

        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();
    }
}