Java高并发(三)——多线程协作，同步控制

ゝ一世哀愁。 2022-04-15 03:27 190阅读 0赞

继上一篇：[Java高并发——多线程基础][Java] 中讲到，共享资源的合理使用，才能够使多线程程序有条不紊的运行。其中我们通过synchronized来实现临界区资源的是否可以访问。而，这篇我们来重点总结synchronized的增强替代版锁，以及其它JDK并发包提供的一些同步控制的功能。

好，还是先看下知识的总结思维导图，然后分开进行总结：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70][]

一，ReentrantLock（重入锁）：1，顾名思义就是像一把锁，我们可以锁住，又可以打开，从而控制资源的同步访问。而其中重入特性指的是同一个线程，可以反复的进入；2，中断响应，对于synchronized只有保持等待，和继续执行两种情况；而ReentrantLock在等待的过程，我们可以通知其放弃等待（类似生活中约会，你等了一会朋友没到，但是朋友遇到突发情况不能来了，给你打了电话通知你，你就不等了）；3，申请等待时间：就是指定等待时间，在指定时间没得到，则放弃；4，公平锁：指定fair为true则进行先到先得，而不是随机选取。这里看下ReentrantLock的相关例子：

//1,ReentrantLock例子
    public class ReentrantLockTest implements Runnable{
        public static ReentrantLock lock  =new ReentrantLock();
    
        public static int i =0;
    
        public void run() {
            for (int j = 0; j < 1000; j++) {
                lock.lock();
                //lock.lock();
                try {
                    i++;
                }finally {
                    lock.unlock();
                    //lock.unlock();
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
            Thread t1 = new Thread(reentrantLockTest);
            Thread t2 = new Thread(reentrantLockTest);
    
            t1.start();
            t2.start();
            t1.join();
            t2.join();
            System.out.println(i);
        }
    }
    
    
    //2，lock1.lockInterruptibly()中断后可放弃
    public class InterruptLock implements Runnable {
        public static ReentrantLock lock1 = new ReentrantLock();
        public static ReentrantLock lock2 = new ReentrantLock();
    
        int lock;
    
        public InterruptLock(int lock) {
            this.lock = lock;
        }
    
        public void run() {
            try {
                if (lock == 1) {
                    lock1.lockInterruptibly();
                    Thread.sleep(500);
                    lock2.lockInterruptibly();
                } else {
                    lock2.lockInterruptibly();
                    Thread.sleep(500);
                    lock1.lockInterruptibly();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (lock1.isHeldByCurrentThread()) {
                    lock1.unlock();
                }
                if (lock2.isHeldByCurrentThread()) {
                    lock2.unlock();
                }
                System.out.println(Thread.currentThread().getName() + "线程退出");
    
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            InterruptLock interruptLock1 = new InterruptLock(1);
            InterruptLock interruptLock2 = new InterruptLock(2);
            Thread t1 = new Thread(interruptLock1);
            Thread t2 = new Thread(interruptLock2);
            t1.start();
            t2.start();
            Thread.sleep(1000);
            t2.interrupt();
    
        }
    }
    
    
    
    //3，申请等待时间例子：如果直接使用tryLock()如果拿不到则直接返回，不会等待
    public class TimeLock implements Runnable {
    
        public static ReentrantLock lock = new ReentrantLock();
    
        public void run() {
            try {
                if (lock.tryLock(5, TimeUnit.SECONDS)) {
                    System.out.println( Thread.currentThread().getName() + "get lock success");
                    Thread.sleep(6000);
                } else {
                    System.out.println( Thread.currentThread().getName() + "get lock failed");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }
        }
    
        public static void main(String[] args) {
            TimeLock timeLock = new TimeLock();
            Thread t1 = new Thread(timeLock);
            Thread t2 = new Thread(timeLock);
            t1.start();
            t2.start();
        }
    }
    
    
    //4,公平锁，先到先得
    public class FairLock implements Runnable {
        public static ReentrantLock fairLock = new ReentrantLock(true);
    
        public void run() {
            while (true){
                try{
                    fairLock.lock();
                    System.out.println(Thread.currentThread().getName() + "get lock");
                }finally {
                    fairLock.unlock();
                }
            }
        }
    
        public static void main(String[] args) {
            FairLock fairLock = new FairLock();
            Thread t1 = new Thread(fairLock,"thread1");
            Thread t2 = new Thread(fairLock,"thread2");
            t1.start();
            t2.start();
        }
    }

二，Condition条件：记得上篇博客中我们wait()、notify()等待和通知，Condition也可以实现类似的功能，配合锁进行使用。提供的方法await()、awaitUninterruptibly()、awaitNanos(long nanosTimeout)、await(long time,TimeUnit unit)、signal()、signalAll()等，也很容易理解。这里提一下：生产者消费者模型中，如果库房慢了，则生产者await；如果库房没了，则消费者await；生产者生成一个则signal；消费者消费一个则signal。是不是非常实用。好看个简单例子：

public class ConditionTest implements Runnable {
    
        public static ReentrantLock reentrantLock = new ReentrantLock();
        public static Condition condition = reentrantLock.newCondition();
    
        public void run() {
    
            try {
                reentrantLock.lock();
                condition.await();
                System.out.println("Thread is going on");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                reentrantLock.unlock();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            ConditionTest conditionTest = new ConditionTest();
            Thread t1 = new Thread(conditionTest);
            t1.start();
            Thread.sleep(2000);
            reentrantLock.lock();
            condition.signal();
            reentrantLock.unlock();
        }
    }

三，Semaphore信号量：这个也挺容易理解的。无论是synchronized还是lock都是一次只能一个线程获取资源，而信号量可以多个同时访问。其中方法有：acquire()、acquireUninterruptibly()、tryAcquire()、tryAcquire(long timeout,TimeUnit unit)、release()等。看个简单的例子：

public class SemaphoreTest implements  Runnable {
        final Semaphore semaphore =new Semaphore(5);
    
        public void run() {
            try {
                semaphore.acquire();
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() + "done");
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(20);
            final SemaphoreTest semaphoreTest = new SemaphoreTest();
            for (int i = 0; i <20 ; i++) {
                executorService.execute(semaphoreTest);
            }
        }
    }

四，ReadWriteLock读写锁：现实业务场景中往往都是读多写少，而读不会带来数据的不一致性，所以就有了读写锁，读读不阻塞、读写阻塞、写写阻塞，对于读远远大于写的非常使用。看个简单例子：

public class ReadWriteLockTest {
        private static Lock lock = new ReentrantLock();
        private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        private static Lock readLock = reentrantReadWriteLock.readLock();
        private static Lock writeLock = reentrantReadWriteLock.writeLock();
        private int value;
    
        public Object handleRead(Lock lock) throws InterruptedException {
            try {
                lock.lock();
                Thread.sleep(1000);
                return value;
            } finally {
                lock.unlock();
            }
    
        }
    
        public void handleWrite(Lock lock, int index) throws InterruptedException {
            try {
                lock.lock();
                Thread.sleep(1000);
                value = index;
            } finally {
                lock.unlock();
            }
        }
    
    
        public static void main(String[] args) {
            final ReadWriteLockTest readWriteLockTest = new ReadWriteLockTest();
            Runnable readRunnale = new Runnable() {
                public void run() {
                    try {
                        readWriteLockTest.handleRead(readLock);
                        //readWriteLockTest.handleRead(lock);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
    
    
            Runnable writeRunnale = new Runnable() {
                public void run() {
                    try {
                        readWriteLockTest.handleWrite(readLock, new Random().nextInt());
                        //readWriteLockTest.handleWrite(lock,new Random().nextInt());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
    
            for (int i = 0; i < 18; i++) {
                new Thread(readRunnale).start();
            }
            for (int i = 0; i < 2; i++) {
                new Thread(writeRunnale).start();
            }
        }
    }

五，CountDownLatch倒计时器：也是非常实用的，可以让主线等待一组子线程执行完毕再进行业务的处理。看个简单例子：

public class CountDownLatchTest implements Runnable {
        static final CountDownLatch countDownLatch = new CountDownLatch(10);
        static final CountDownLatchTest countDownLatchTest = new CountDownLatchTest();
    
        public void run() {
            try {
                Thread.sleep(new Random().nextInt(10)*100);
                System.out.println("check complete");
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                countDownLatch.countDown();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            for (int i = 0; i <10 ; i++) {
                executorService.execute(countDownLatchTest);
            }
    
            countDownLatch.await();
            System.out.println("all complete");
            executorService.shutdown();
        }
    }

六，CyclicBarrier循环栅栏：增强版CountDownLatch，但是是可以循环使用。这里看个经典例子：司令下达命令，10个士兵一起去完成一项任务：10个士兵首先集合报道，然后去执行任务，执行完了再汇报司令，司令宣布完成任务。相当于计数了两次，循环使用了。好看个例子：

public class CyclicBarrierTest {
    
        public static class Soldier implements Runnable {
            private String soldier;
            private final CyclicBarrier cyclicBarrier;
    
            Soldier(CyclicBarrier cyclicBarrier, String soldierName) {
                this.soldier = soldierName;
                this.cyclicBarrier = cyclicBarrier;
            }
    
            public void run() {
    
                try {
                    //等待所有士兵到齐
                    cyclicBarrier.await();
                    doWork();
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
    
            void doWork() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(soldier + "do work");
            }
        }
    
    
        public static class BarrierRun implements Runnable {
    
            boolean flag;
            int n;
    
            public BarrierRun(boolean flag, int n) {
                this.flag = flag;
                this.n = n;
            }
    
            public void run() {
                if (flag) {
                    System.out.println("soldier" + n + "个，do work");
                } else {
                    System.out.println("soldier" + n + "个，集合完毕");
                    flag = true;
                }
            }
        }
    
    
        public static void main(String[] args) {
            final int n = 10;
            Thread[] allSoldier = new Thread[n];
            boolean flag = false;
            CyclicBarrier cyclicBarrier = new CyclicBarrier(n, new BarrierRun(flag, n));
            //
            System.out.println("集合");
            for (int i = 0; i < n; i++) {
                System.out.println(i + "+soldier come");
                allSoldier[i] = new Thread(new Soldier(cyclicBarrier, "soldier" + n));
                allSoldier[i].start();
    
            }
        }
    }

七，LockSupport：上篇讲了挂起（suspend）和继续执行（resume），其中都是JDK不建议使用的，也说到它的不好处。而LockSupport的静态方法park()可以阻塞当前线程，unpark()继续执行，利用的就是信号量的原理，它为每个线程准备了一个许可证，如果许可证可用，则park()立即返回，并消费许可，如果不可用则进行阻塞；而unpark()则使这个许可变为可用。许可为唯一的。

public class LockSupportTest {
    
        public static Object u = new Object();
        static ChangeObjectThread t1 =new ChangeObjectThread("t1");
        static ChangeObjectThread t2 =new ChangeObjectThread("t2");
    
        public static class ChangeObjectThread extends Thread{
            public ChangeObjectThread(String name){
                super.setName(name);
            }
    
            @Override
            public void run() {
                synchronized (u){
                    System.out.println("in"+ getName());
                    LockSupport.park();
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            t1.start();
            Thread.sleep(100);
            t2.start();
            LockSupport.unpark(t1);
            LockSupport.unpark(t2);
            t1.join();
            t2.join();
        }
    }

好，通过上边几个类的功能都可以很好的满足多线程之间的协作，同步控制资源。只有让数据不出错，才能更高的发挥多线程的高效价值！继续……

[Java]: https://blog.csdn.net/liujiahan629629/article/details/84205788
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70]: /images/20220415/d9d4b25bd20146dabaa40bfde5929070.png