锁

ReentrantLock,ReadWriteLock,CountdownLatch,CyclicBarrier,Phaser,Semaphore,Exchanger

一.各种锁的作用：

如果用了很多锁后,担心死锁逻辑,可以用jstack检查
ReentrantLock和Synchronized相比,最大的特点就是可以根据业务需求灵活控制.
ReentrantReadWriteLock用于一块资源读多写少,又要求避免脏数据的情况
Semaphore用于限流,控制最多同时执行的线程数.
剩下的主要是面试时候吊打面试官(手动狗头):
CountdownLatch和CyclicBarrier类似,需要准备大批量的数据/资源调动,分多个线程去准备,但是等这些数据都准备OK后才能进行下一步动作.(MapReduce?)
CountdownLatch是数到0就完事了,再往下数啥效果都没有,不能循环使用;
CyclicBarrier可以循环使用.
Exchanger可能写游戏的两人交易时用到
Phaser可能在遗传算法时用到

二.ReentrantLock(重要)

什么是可重入锁?字面意思, 可重入锁就是对同样的一把锁可以再锁一次.当一个线程中执行到加锁代码时,发现当前持有这把锁的线程就是自己,那可以执行该代码.
(ReentrantLock内部使用了CAS,有没有使用系统级别锁呢?回头再探究)

三.ReentrantLock特点|使用场景

1.简单使用,替代synchronized
2.可以tryLock(一定时间)
3.可以被马上打断lockInterruptibly
4.可以使用公平锁

简单使用,替代synchronized

这个可以替代synchronized,只是需要手动上锁(lock)和解锁(unlock),要注意lock操作以及接下来的业务代码用try{}包起来,在finally{}中unlock,无论是否有异常,都会释放锁.
使用synchronized时,当加锁代码执行完或者抛异常时,JVM会自动释放锁.

public class T02_ReentrantLock2 {
    Lock lock = new ReentrantLock();
    void m1() {
        try {
            lock.lock(); //synchronized(this)
            for (int i = 0; i < 10; i++) {
                TimeUnit.SECONDS.sleep(1);
                System.out.println(i);
                // 此处验证一下"可重入"特性
                if (i == 2) {
                    m2();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    void m2() {
        try {
            lock.lock();
            System.out.println("m2 ...");
        } finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) {
        T02_ReentrantLock2 rl = new T02_ReentrantLock2();
        new Thread(rl::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 必须等锁被上一个线程释放后,才能继续执行
        new Thread(rl::m2).start();
    }
}

可以有不同的condition

每一个condition可以理解为一个等待队列.
synchronized中,线程wait后,其实是进了一个等待队列,等待被唤醒;当调用notify时随机唤醒等待队列其中一个线程,当调用notifyall时,唤醒等等待线程中的所有线程.
synchronized只有一个默认的等待对了,而ReentrantLock可以有多个等待队列,即多个condition.

看一个生产者/消费者的例子:

/**
 * 写一个固定容量同步容器，拥有put和get方法,以及getCount方法,
 * 能够支持2个生产者线程以及10个消费者线程的阻塞调用
 */
public class WzContainerReentrantLock<T> {
    private int max = 10;
    private int size = 0;
    private LinkedList<T> values = new LinkedList<>();
    private Lock lock = new ReentrantLock();
    private Condition producerCondition = lock.newCondition();
    private Condition consumerCondition = lock.newCondition();
    public T get() {
        T t = null;
        try {
            lock.lock();
            while (size == 0) {
                consumerCondition.await();
            }
            t = values.removeFirst();
            size--;
            producerCondition.signalAll();// 通知生产者
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return t;
    }
    public void put(T t) {
        try {
            lock.lock();
            while (size == this.max) {
                producerCondition.await();
            }
            values.addLast(t);
            size++;
            consumerCondition.signalAll();// 通知消费者
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) {
        WzContainerReentrantLock<String> container = new WzContainerReentrantLock<>();
        //启动消费者线程
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 4; j++) {
                    String s = container.get();
                    System.out.println(s);
                }
            }, "consumer-" + i).start();
        }
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //启动生产者线程
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                for (int j = 0; j < 20; j++) {
                    container.put(Thread.currentThread().getName() + " at " + Instant.now());
                }
            }, "producer-" + i).start();
        }
    }
}

可以tryLock(一定时间)

public class T03_ReentrantLock3 {
    Lock lock = new ReentrantLock();
    void m1() {
        try {
            lock.lock();
            for (int i = 0; i < 10; i++) {
                TimeUnit.SECONDS.sleep(1);
                System.out.println(i);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    /**
     * 使用tryLock进行尝试锁定，不管锁定与否，方法都将继续执行
     * 可以根据tryLock的返回值来判定是否锁定
     * 也可以指定tryLock的时间，由于tryLock(time)抛出异常，所以要注意unclock的处理，必须放到finally中
     */
    void m2() {
        // 当然也可以lock.tryLock(),不指定时间
        /*
        boolean locked = lock.tryLock();
        System.out.println("m2 ..." + locked);
        if(locked) lock.unlock();
        */
        boolean locked = false;
        try {
            long start = System.currentTimeMillis();
            locked = lock.tryLock(5, TimeUnit.SECONDS);
            long end = System.currentTimeMillis();
            System.out.println("m2 ..." + locked + "|waiting time:" + (end - start));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if(locked) {
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) {
        T03_ReentrantLock3 rl = new T03_ReentrantLock3();
        new Thread(rl::m1).start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(rl::m2).start();
    }
}

可以被马上打断lockInterruptibly

lock()方法的锁,只能在当前线程获取到锁后才能被打断;
而lockInterruptibly()可以马上被打断,不必等获取锁.
证明一下:

public class T04_ReentrantLock4 {
    public static void main(String[] args) {
        Lock lock = new ReentrantLock();
        Thread t1 = new Thread(()->{
            try {
                lock.lock();
                System.out.println("t1 start");
                TimeUnit.SECONDS.sleep(10);
                System.out.println("t1 end");
            } catch (InterruptedException e) {
                System.out.println("t1 was interrupted!");
            } finally {
                lock.unlock();
            }
        });
        t1.start();
        Thread t2 = new Thread(()->{
            try {
                // 如果是直接lock的话,t2就不能被直接打断,只能等t1执行完了,t2获取到lock后才能被打断
//                lock.lock();
                lock.lockInterruptibly(); //可以对interrupt()方法做出响应
                System.out.println("t2 start");
                TimeUnit.SECONDS.sleep(5);
                System.out.println("t2 end");
            } catch (InterruptedException e) {
                System.out.println("t2 was interrupted!");
            } finally {
                lock.unlock();
            }
        });
        t2.start();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t2.interrupt(); //打断线程2的等待
    }
}

公平锁

ReentrantLock默认为非公平锁,多个线程在一个队列中去竞争获取锁,随机拿到锁,就跟买彩票一样,不是先买的就先中奖(synchronized也是非公平锁);
公平锁就是像买饭排队一样,比较”公平”.

ReentrantLock可以指定是否为公平锁,默认非公平锁;
synchronized只有非公平锁

这个不太好写测试用例,回头再证明吧…
创建ReentrantLock时,传入fair参数为true,创建的就是公平锁

public class T05_ReentrantLock5 extends Thread {
    //参数为true表示为公平锁，请对比输出结果
    private static ReentrantLock lock = new ReentrantLock(true);
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            System.out.println(Thread.currentThread().getName() + " will lock at" + System.currentTimeMillis());
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + " get lock at" + System.currentTimeMillis());
            } finally {
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        T05_ReentrantLock5 rl = new T05_ReentrantLock5();
        Thread th1 = new Thread(rl);
        Thread th2 = new Thread(rl);
        th1.start();
        th2.start();
    }
}

ReadWriteLock读写锁(重要)

读写锁是读锁和写锁的合称,

读锁:共享锁,其他线程可以同时读,在读操作之间共享资源;但是其他线程不能写,否则会读到脏数据
写锁:排他锁/互斥锁,其他线程既不能写也不能读,也是为了防止脏数据.
其他的锁一般都是排它锁/互斥锁,当一个线程拿到锁后,别的线程就只能阻塞等待,啥也干不了.

主要用于一些内容,写操作不多,但是读操作很多的地方,比如公司的组织架构

读写锁代码示例:

public class T10_TestReadWriteLock {
    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 read(Lock lock, CountDownLatch countDownLatch) {
        try {
            lock.lock();
            //模拟读取操作
            Thread.sleep(1000);
            countDownLatch.countDown();
            System.out.println("read over!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public static void write(Lock lock, int v, CountDownLatch countDownLatch) {
        try {
            lock.lock();
            // 写操作,模拟耗时1秒
            Thread.sleep(1000);
            value = v;
            countDownLatch.countDown();
            System.out.println("write over!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        // 使用CountDownLatch方便计时,下面有CountdownLatch说明
        CountDownLatch countDownLatch = new CountDownLatch(20);
        // 普通互斥锁
//        Runnable readR = ()-> read(lock, countDownLatch);
//        Runnable writeR = ()->write(lock, new Random().nextInt(), countDownLatch);
        // 读写锁
        Runnable readR = () -> read(readLock, countDownLatch);
        Runnable writeR = () -> write(writeLock, new Random().nextInt(), countDownLatch);
        long start = System.currentTimeMillis();
        for (int i = 0; i < 18; i++) {
            new Thread(readR).start();
        }
        for (int i = 0; i < 2; i++) {
            new Thread(writeR).start();
        }
        countDownLatch.await();
        long end = System.currentTimeMillis();
        System.out.println("cust milliseconds:" + (end - start));
    }
}

Semaphore信号量

有点类似于令牌桶算法.用于控制并发的最大数量,限流.
有时候一个任务的某一步骤特别耗时,很多个线程同时执行的话,每个线程都很慢,可能大家都超时失败,为了防止这一现象,可以用信号量控制最大并发数.
假设希望最大并发数为n,那就创建一个有n个令牌的桶,每次线程执行特定方法时,去申请令牌(acquire),拿不到就阻塞;拿到了才继续执行,执行完再把令牌放回桶里(release).

代码示例:

public class T11_TestSemaphore {
    public static void main(String[] args) {
//        Semaphore s = new Semaphore(3);
        Semaphore s = new Semaphore(3, true);
        //允许一个线程同时执行
//        Semaphore s = new Semaphore(1);
        for (int i = 1; i <= 10; i++) {
            new Thread(() -> {
                // 可以做其他的事情,不受信号量控制
                String name = Thread.currentThread().getName();
                try {
                    s.acquire();
                    System.out.println(name + " start...");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    s.release();
                    System.out.println(name + " end...");
                }
                // 可以做其他的事情,不受信号量控制
            }).start();
        }
    }
}

CountdownLatch

见名知意,倒数计数(而非计时)的门闩,计数结束就开门了,然后可以继续执行指令.
使用场景:用于在一个线程中有些特殊操作,需要在业务逻辑上等待其他几个线程完成特定操作后才能执行.
比线程的join()方法更灵活一些,因为不需要等其他线程全部执行完,而是手动灵活控制.
做个类比:饭店中的传菜员拿来一页订单,客人要求几个菜一起上,传菜员就要等大厨们都做完装盘后才能把才端给客人;而且大厨炒完菜还要刷锅,这时候传菜员是不需要等待的.
看一个例子:

public class T06_TestCountDownLatch {
    public static void main(String[] args) {
        // 使用线程的join方法
//        usingJoin();
        // 使用门闩
        usingCountDownLatch();
    }
    private static void usingCountDownLatch() {
        Thread[] threads = new Thread[100];
        CountDownLatch latch = new CountDownLatch(threads.length);
        for(int i=0; i<threads.length; i++) {
            threads[i] = new Thread(()->{
                int result = 0;
                // 业务逻辑a,假设本操作是一些操作的前置条件
                for(int j=0; j<10000; j++) {
                    result += j;
                }
                // 门闩倒数计数(减1),原子操作
                latch.countDown();
                // 下面可以有业务逻辑b,c,d无所谓了
                try {
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName() + "at" + new Date());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        for (int i = 0; i < threads.length; i++) {
            threads[i].start();
        }
        try {
            // 插上门闩,等倒数计数完了(到0)才能继续执行
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("end latch at" + new Date());
    }
    // 这个作为对比
    private static void usingJoin() {
        Thread[] threads = new Thread[100];
        for(int i=0; i<threads.length; i++) {
            threads[i] = new Thread(()->{
                int result = 0;
                for(int j=0; j<10000; j++) {
                    result += j;
                }
            });
        }
        for (int i = 0; i < threads.length; i++) {
            threads[i].start();
        }
        // 等threads中的线程全都结束后才能继续执行
        for (int i = 0; i < threads.length; i++) {
            try {
                threads[i].join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("end join");
    }
}

CyclicBarrier

见名知意,循环栅栏.
就像装鸡蛋的托盘一样,鸡蛋把托盘装满后才去装箱.

首先创建一个CyclicBarrier,它规定当一定数量的线程调用CyclicBarrier的await()方法后,CyclicBarrier会执行一些操作(也可以啥都不做);
其他线程调用CyclicBarrier的await()方法后,线程开始阻塞,等待barrier满了(各个线程就绪)后才继续执行.
CyclicBarrier的barrierAction由该批次进入的最后一个线程执行

下面是个例子,可以看出是先Ready后那些线程才能继续执行.

public class T07_TestCyclicBarrier {
    public static void main(String[] args) {
        // 线程满了后啥都不做,这个应该用不到
        //CyclicBarrier barrier = new CyclicBarrier(20);
        // lambda表达式,和下面的匿名内部类是等效的
        CyclicBarrier barrier = new CyclicBarrier(20, () -> System.out.println("Ready,go! @ " + Instant.now()));
        /*CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() {
            @Override
            public void run() {
                System.out.println("Ready,go! @ " + Instant.now());
            }
        });*/
        for (int i = 0; i < 100; i++) {
            new Thread(() -> {
                try {
                    TimeUnit.SECONDS.sleep(1);
                    barrier.await();
                    System.out.println(Thread.currentThread().getName() + "GO ON @ " + Instant.now());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

Phaser(极少用到)

phase:相位,阶段;phaser就是按照不同的相位阶段去执行不同的命令.

有点像是CountdownLatch和CyclicBarrier的结合:先注册一定的数量parties作为参与的线程数,大家都准备就绪后触发一次onAdvance操作;也可以随时修改这个parties数量.
触发onAdvance时可以获取到当前是第几个相位(第几次触发),和当前有多少个注册的parties

可以这么理解:CyclicBarrier是只有一个栅栏,Phaser是纵向好几个栅栏,每个栅栏触发时可以有不同的操作.

关键方法:
0.自定义一个类继承Phaser,并重写onAdvance方法.每次parties到达相位后,会调用onAdvance方法
1.phaser.bulkRegister(int parties)
批量注册parties,有点类似于CountdownLatch的倒数计数的初始化
2.phaser.arriveAndAwaitAdvance()
Arrives at this phaser and awaits others.
各parties准备就绪后到达相位,等待其他parties后才继续执行,注意下一次相位仍会参与
3.phaser.arriveAndDeregister()
Arrives at this phaser and deregisters from it without waiting for others to arrive.
到达这个相位,并且从中注销,不需等其他parties到达就可继续执行,不再参与Phaser规则了

应用场景:
遗传算法(大概是用计算机去模拟达尔文进化策略),这个我没去详细了解.
以简单的结婚流程为例:
1.首先关键人物(比如一对新人,伴郎伴娘和亲朋好友等客人)都要到场;2.然后走个结婚的流程,新人发表感言,客人吃饭,新人敬酒;3.客人陆续离场;4.新人入洞房,相互拥抱.
这些步骤顺序不能乱,行为的执行者也不能乱.(只有在大家离场后,两位新人才能入洞房,哈哈)

代码示例:

public class T09_TestPhaserWedding {
    static Random r = new Random();
    static MarriagePhaser phaser = new MarriagePhaser();
    static void milliSleep(int milli) {
        try {
            TimeUnit.MILLISECONDS.sleep(milli);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        phaser.bulkRegister(7);
        // 5位客人
        for(int i=0; i<5; i++) {
            new Thread(new Person("p" + i)).start();
        }
        // 2位新人
        new Thread(new Person("bridegroom")).start();
        new Thread(new Person("bride")).start();
    }
    static class MarriagePhaser extends Phaser {
        /**
         *
         * @param phase 当前是第几个阶段.the current phase number on entry to this method,
         * before this phaser is advanced
         * @param registeredParties 目前多少线程参与.the current number of registered parties
         * @return {@code true} if this phaser should terminate
         */
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            switch (phase) {
                case 0:
                    System.out.println("everyone arrived!" + registeredParties);
                    System.out.println();
                    return false;
                case 1:
                    System.out.println("everyone finished eating!" + registeredParties);
                    System.out.println();
                    return false;
                case 2:
                    System.out.println("everyone left!" + registeredParties);
                    System.out.println();
                    return false;
                case 3:
                    System.out.println("The Newlywed Time!" + registeredParties);
                    return true;
                default:
                    return true;
            }
        }
    }
    static class Person implements Runnable {
        String name;
        public Person(String name) {
            this.name = name;
        }
        public void arrive() {
            milliSleep(r.nextInt(1000));
            System.out.printf("%s arrived!\n", name);
            phaser.arriveAndAwaitAdvance();
        }
        public void eat() {
            milliSleep(r.nextInt(1000));
            System.out.printf("%s finish eating!\n", name);
            phaser.arriveAndAwaitAdvance();
        }
        public void leave() {
            milliSleep(r.nextInt(1000));
            System.out.printf("%s leave!\n", name);
            phaser.arriveAndAwaitAdvance();
            // 所有人离去后,保洁人员才打扫座位
            System.out.printf("everyone was left, clean %s's seat\n", name);
        }
        private void hug() {
            if(name.equals("bridegroom") || name.equals("bride")) {
                milliSleep(r.nextInt(1000));
                System.out.printf("%s hug!\n", name);
                phaser.arriveAndAwaitAdvance();
            } else {
                phaser.arriveAndDeregister();
                // 注销后,不必等本次相位的所有parties到达,可以随意执行
                System.out.printf("%s Deregister\n", name);
            }
        }
        @Override
        public void run() {
            arrive();
            eat();
            leave();
            hug();
        }
    }
}

Exchanger交换器

用于两个线程间交换数据.
中间有个类似CyclicBarrier的操作,第一个线程准备好而第二个线程没准备好前,第一个会阻塞;只有两个线程都准备好后才会开始交换,然后继续执行

应用场景:两人做交易(不知道各位有没有玩过零几年很火的游戏<传奇>?)

代码示例:

public class T12_TestExchanger {
    static Exchanger<String> exchanger = new Exchanger<>();
    public static void main(String[] args) {
        new Thread(()->{
            String source = "T1";
            String getted = null;
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println("T1 ready @" + new Date());
                getted = exchanger.exchange(source);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " get:" + getted + "@" + new Date());
        }, "t1").start();
        new Thread(()->{
            String source = "T2";
            String getted = null;
            try {
                System.out.println("T2 ready @" + new Date());
                getted = exchanger.exchange(source);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " get:" + getted + "@" + new Date());
        }, "t2").start();
        // 如果再来个T3,程序会一直阻塞下去
        /*new Thread(()->{
            String source = "T3";
            String getted = null;
            try {
                System.out.println("T3 ready @" + new Date());
                getted = exchanger.exchange(source);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " get:" + getted + "@" + new Date());
        }, "t3").start();*/
    }
}