多线程常用工具类—— CountDownLatch、CyclicBarrier、Semaphore

分手后的思念是犯贱 2024-04-17 15:52 121阅读 0赞

# CountDownLatch #

作用：是一组线程等待其他(一组)的线程完成工作以后再执行，可看做加强版join  
常用方法

CountDownLatch(int count) //实例化一个倒计数器，count指定计数个数
     countDown() // 计数减一
     await() //等待，当计数减到0时，所有线程并行执行

举个栗子，下面代码，要保证上面的11个线程都执行完主线程才往下执行，就需要再每个线程里调用countDown保证计数器最后减到0。

public class CountDownTest { 
        private static class CountThread implements  Runnable{ 
           final static CountDownLatch latch = new CountDownLatch(11);
            @Override
            public void run() { 
                System.out.println("kai shi zhi xing==========");
                latch.countDown();
    
            }
        }
        public static void main(String[] args) throws InterruptedException { 
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("2 kai shi zhi xing");
                    CountThread.latch.countDown();
                }
            }).start();
    
            for(int i=0;i<10;i++){ 
                new Thread(new CountThread()).start();
            }
            CountThread.latch.await();//只有latch减到了0才开始执行后面的代码
            System.out.println("主线程开始执行了");
        }
    }

值得一提的是,一个线程里可以调用多次countDown;即计数器在一个线程里减n

# CyclicBarrier #

CyclicBarrier，让一组线程到达一个同步点后(barrier.await()方法)再一起继续运行，在其中任意一个线程未达到同步点，其他到达的线程均会被阻塞。

**await方法**  
调用await方法的线程告诉CyclicBarrier自己已经到达同步点，然后当前线程被阻塞。直到parties个参与线程调用了await方法，CyclicBarrier同样提供带超时时间的await和不带超时时间的await方法：

**构造方法**

public CyclicBarrier(int parties, Runnable barrierAction) 
    public CyclicBarrier(int parties)

第一种构造方法里的第一个参数是到达屏障的线程数，第二个参数是到屏障数达到要求后（嗯，这里可以理解为加法）。首先要执行的任务

还是举个栗子

public class BarrierTest implements Runnable{ 
        private static final CyclicBarrier barrier = new CyclicBarrier(4, new RunThread());
        @Override
        public void run() { 
            long currentTheadId = Thread.currentThread().getId();
            Random rd = new Random();
            try { 
                if(rd.nextBoolean()){ 
                    System.out.println("Thread: "+currentTheadId+"你们先上，我睡一会");
                    Thread.sleep(1000);
                }
                System.out.println("Thread: "+currentTheadId+"===准备就绪");
                barrier.await();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (BrokenBarrierException e) { 
                e.printStackTrace();
            }
            System.out.println("Thread: "+currentTheadId+"===冲突屏障，继续执行");
        }
        private static class RunThread implements Runnable{ 
            @Override
            public void run() { 
                System.out.println("开始执行结束之前的收尾工作");
            }
        }
    
        public static void main(String[] args) { 
            for(int i=0;i<4;i++){ 
                new Thread(new BarrierTest()).start();
            }
        }
    }

上面的打印结果

Thread: 9你们先上，我睡一会
    Thread: 10你们先上，我睡一会
    Thread: 11你们先上，我睡一会
    Thread: 12===准备就绪
    Thread: 9===准备就绪
    Thread: 10===准备就绪
    Thread: 11===准备就绪
    开始执行结束之前的收尾工作
    Thread: 11===冲突屏障，继续执行
    Thread: 12===冲突屏障，继续执行
    Thread: 10===冲突屏障，继续执行
    Thread: 9===冲突屏障，继续执行

**CyclicBarrier和CountDownLatch的区别**  
1、countdownlatch放行由第三者控制（主线程中的 CountThread.latch.await()方法），CyclicBarrier放行由一组线程本身控制（线程run方法中执行barrier.await();）  
2、countdownlatch放行条件》=线程数，CyclicBarrier放行条件=线程数，一个是计数器减法，一个是加法（这么理解）

# Semaphore #

Semaphore用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。还可以用来实现某种资源池限制，或者对容器施加边界。

1、**semaphore.acquire();**  
请求一个信号量，这时候信号量个数-1，当减少到0的时候，下一次acquire不会再执行，只有当执行一个release()的时候，信号量不为0的时候才可以继续执行acquire  
2、**semaphore.release();**  
释放一个信号量，这时候信号量个数+1，  
3、**getQueueLength**  
获取当前等待获取信号量的线程

**就像信号量的字面意思，可以将其理解为一个控制单位，当连接数大于等于其定义时的值时，代码不再往下执行**

举个例子，我们来用Semaphore来控制数据库连接池个数\`\`

public class DBPoolSemaphore { 
       
       private final static int POOL_SIZE = 10;
       private final Semaphore useful,useless;//useful表示可用的数据库连接，useless表示已用的数据库连接
       
       public DBPoolSemaphore() { 
          this. useful = new Semaphore(POOL_SIZE);
          this.useless = new Semaphore(0);
       }
       //存放数据库连接的容器
       private static LinkedList<Connection> pool = new LinkedList<Connection>();
       //初始化池
       static { 
            for (int i = 0; i < POOL_SIZE; i++) { 
                pool.addLast(SqlConnectImpl.fetchConnection());
            }
       }
       /*归还连接*/
       public void returnConnect(Connection connection) throws InterruptedException { 
          if(connection!=null) { 
             System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接！！"
                   +"可用连接数:"+useful.availablePermits());
             useless.acquire();
             synchronized (pool) { 
                pool.addLast(connection);
             }  
             useful.release();
          }
       }
       
       /*从池子拿连接*/
       public Connection takeConnect() throws InterruptedException { 
          useful.acquire();//这里是重点
          Connection conn;
          synchronized (pool) { 
             conn = pool.removeFirst();
          }
          useless.release();
          return conn;
       }
       
    }