闭锁、栅栏、交换机、信号量学习笔记

Dear 丶 2022-06-06 04:10 219阅读 0赞

1.闭锁(CountDownLatch)

通过一个计数器实现，原理就是不停地检查join线程是否存活，如果存活则一直等待，作用是m(m>=1)个线程等待其他的线程执行玩之后然后继续往下执行，类似join()方法，但功能更强大，可以操作更多的线程。

![20171027090930396.png][]

测试代码:

public class TestTemp {
        public static void main(String[] args) throws InterruptedException {
            CountDownLatch latch = new CountDownLatch(3);//计数器初始值设置为3
            Waiter waiter = new Waiter(latch);
            Decrementer decrementer = new Decrementer(latch);
            new Thread(waiter)
                    .start();
            new Thread(decrementer).start();
            Thread.sleep(4000);
        }
    
    }
     class Waiter implements Runnable{
        CountDownLatch latch = null;
        public Waiter(CountDownLatch latch) {
            this.latch = latch;
        }
        public void run() {
            try {
                latch.await();//阻塞在这边
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Waiter Released");
        }
    }
    class Decrementer implements Runnable {
        CountDownLatch latch = null;
        public Decrementer(CountDownLatch latch) {
            this.latch = latch;}
        public void run() {
            try {
                Thread.sleep(1000);
                this.latch.countDown();//计数器减1
                Thread.sleep(1000);
                this.latch.countDown();//计数器减1
                Thread.sleep(1000);
                this.latch.countDown();//计数器减1
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

当计数器的值减少为0的时候,Waiter线程就能够不再阻塞，继续往下运行了。

2.栅栏

![Center][]

使用场景:所有线程相互等待，直到所有线程都到达某一点时才打开栅栏，然后线程继续执行。

public class TestTemp {
        public static void main(String[] args) throws InterruptedException {
            Runnable barrier1Action = new Runnable() {
                public void run() {
                    System.out.println("BarrierAction 1 executed ");
                }
            };
            Runnable barrier2Action = new Runnable() {
                public void run() {
                    System.out.println("BarrierAction 2 executed ");
                }
            };
            CyclicBarrier barrier1 = new CyclicBarrier(2, barrier1Action);//有两个线程等待就执行barrier1Action
            CyclicBarrier barrier2 = new CyclicBarrier(2, barrier2Action);//有两个线程等待就执行barrier2Action
            CyclicBarrierRunnable barrierRunnable1 =
                    new CyclicBarrierRunnable(barrier1, barrier2);
            CyclicBarrierRunnable barrierRunnable2 =
                    new CyclicBarrierRunnable(barrier1, barrier2);
            new Thread(barrierRunnable1).start();
            new Thread(barrierRunnable2).start();
        }
    
    
    
    }
     class CyclicBarrierRunnable implements Runnable{
        CyclicBarrier barrier1 = null;
        CyclicBarrier barrier2 = null;
        public CyclicBarrierRunnable(CyclicBarrier barrier1, CyclicBarrier barrier2) {
            this.barrier1 = barrier1;
            this.barrier2 = barrier2;
        }
        public void run() {
            try {
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() +" waiting at barrier 1");
                this.barrier1.await();//等在这里呢
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() +
                        " waiting at barrier 2");
                this.barrier2.await();
                System.out.println(Thread.currentThread().getName() +
                        " done!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

可以看到触发的条件仅仅和两个线程本身有关，而闭锁和外部的条件(计数器)来触发，它们的触发条件是不同的。

3.交换机

![Center 1][]

使用场景:用户线程之间的数据交互,替代wait和notify

public class TestTemp {
        public static void main(String[] args) throws InterruptedException {
            Exchanger exchanger = new Exchanger();
            ExchangerRunnable exchangerRunnable1 =
                    new ExchangerRunnable(exchanger, "A");
            ExchangerRunnable exchangerRunnable2 =
                    new ExchangerRunnable(exchanger, "B");
            new Thread(exchangerRunnable1).start();
            new Thread(exchangerRunnable2).start();
        }
    }
     class ExchangerRunnable implements Runnable{
        Exchanger exchanger = null;
        Object object = null;
        public ExchangerRunnable(Exchanger exchanger, Object object) {
            this.exchanger = exchanger;
            this.object = object;
        }
        public void run() {
            try {
                Object previous = this.object;
                this.object = this.exchanger.exchange(this.object);
                System.out.println(Thread.currentThread().getName() +
                        " exchanged " + previous + " for " + this.object
                );
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

4.信号量

作用:

1. 保护一个重要(代码)部分防止一次超过 N 个线程进入。  
2. 在两个线程之间发送信号。

public class TestTemp {
    
        public static void main(String[] args) throws InterruptedException {
            Service service = new Service();
            ThreadA a = new ThreadA(service);
            a.setName("A");
            ThreadB b = new ThreadB(service);
            b.setName("B");
            ThreadC c = new ThreadC(service);
            c.setName("C");
            a.start();
            b.start();
            c.start();
        }
        
    }
    class Service {
        private Semaphore semaphore = new Semaphore(1);//同一个时刻，只有一个线程可以执行acquire和release部分的代码
        public void testMethod() {
            try {
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName()
                        + " begin timer=" + System.currentTimeMillis());
                Thread.sleep(5000);
                System.out.println(Thread.currentThread().getName()
                        + "   end timer=" + System.currentTimeMillis());
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    class ThreadA extends Thread {
    
        private Service service;
    
        public ThreadA(Service service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.testMethod();
        }
    
    }
    class ThreadB extends Thread {
    
        private Service service;
    
        public ThreadB(Service service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.testMethod();
        }
    
    }
     class ThreadC extends Thread {
    
        private Service service;
    
        public ThreadC(Service service) {
            super();
            this.service = service;
        }
        @Override
        public void run() {
            service.testMethod();
        }
    
    }

[20171027090930396.png]: https://www.2cto.com/uploadfile/Collfiles/20171027/20171027090930396.png
[Center]: /images/20220606/a86f2dbf8377485d97d400ce24995b60.png
[Center 1]: /images/20220606/934c5842a20948eb87e76f5c70a2c4e5.png