并发编程常用工具类(二) SymaPhore实现线程池

1.symaPhore简介

symaphore(信号量)用来控制同时访问某个资源的线程数量，一般用在并发流量控制。个人对它的理解相当于是接待室每次只能接待固定数量的人，当达到最高接待数的时候，其他人就会被拦截在外等待，当前面接待完走出接待室，才会继续接待下面的人。

2.symaphore使用

symaphore有两个构造方法：构造方法Semaphore(int permits)接受一个int参数，表示可用的许可证数量，内部默认创建一个非公平锁；构造方法Semaphore(int permits, boolean fair)接受一个

int和一个boolean值，分别表示可用许可证数量和是否使用公平锁。（公平锁和非公平锁后面文章会单独提到）

一般在做流量控制的时候，我们就可以通过控制许可证数量来控制并发数的大小，接下来具体聊聊怎么实现对线程池的控制，代码如下：

 1 public class DBPoolSemaphore {

 2

 3     private final static int POOL_SIZE = 10;

 4     private final Semaphore useful,useless;//useful表示可用的数据库连接，useless表示已用的数据库连接

 5

 6     public DBPoolSemaphore() {

 7         this. useful = new Semaphore(POOL_SIZE);

 8         this.useless = new Semaphore(0);

 9     }

10

11     //存放数据库连接的容器

12     private static LinkedList<Connection> pool = new LinkedList<Connection>();

13     //初始化池

14     static {

15         for (int i = 0; i < POOL_SIZE; i++) {

16             pool.addLast(SqlConnectImpl.fetchConnection());

17         }

18     }

19

20     /*归还连接*/

21     public void returnConnect(Connection connection) throws InterruptedException {

22         if(connection!=null) {

23             System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接！！"

24                     +"可用连接数:"+useful.availablePermits());

25             useless.acquire();

26             synchronized (pool) {

27                 pool.addLast(connection);

28             }

29             useful.release();

30         }

31     }

32

33     /*从池子拿连接*/

34     public Connection takeConnect() throws InterruptedException {

35         useful.acquire();

36         Connection conn;

37         synchronized (pool) {

38             conn = pool.removeFirst();

39         }

40         useless.release();

41         return conn;

42     }

43 }

首先创建了两个symaphore对象，分别用来表示已用线程池和可用线程池，在设计拿连接和归还连接时，分别先后调用acquire()和release(),acquire()是用来获取许可，release()归还许可，相当于在拿连接时先去可用连接池获取许可，获取到才会继续执行，否则阻塞等待，直到有连接池归还了连接，可用线程许可中可以获取到，获取数据库连接，之后将不可用连接许可增加；归还连接刚好相反。本质就是通过控制可用和不可用许可数目，达到控制并发流量的效果。

下面是我设计的一段执行上面代码的示例：

 1 public class AppTest {

 2

 3     private static DBPoolSemaphore dbPool = new DBPoolSemaphore();

 4

 5     //业务线程

 6     private static class BusiThread extends Thread{

 7         @Override

 8         public void run() {

 9             Random r = new Random();//让每个线程持有连接的时间不一样

10             long start = System.currentTimeMillis();

11             try {

12                 Connection connect = dbPool.takeConnect();

13                 System.out.println("Thread_"+Thread.currentThread().getId()

14                         +"_获取数据库连接共耗时【"+(System.currentTimeMillis()-start)+"】ms.");

15                 SleepTools.ms(100+r.nextInt(100));//模拟业务操作，线程持有连接查询数据

16                 System.out.println("查询数据完成，归还连接！");

17                 dbPool.returnConnect(connect);

18             } catch (InterruptedException e) {

19             }

20         }

21     }

22

23     public static void main(String[] args) {

24         for (int i = 0; i < 50; i++) {

25             Thread thread = new BusiThread();

26             thread.start();

27         }

28     }

29

30 }

有兴趣的同学可以使用上述代码跑一下，可以看到前10个线程不耗时间，可以直接获取到，后面的会阻塞（花费时间越来越长），归还一个才会去获取连接，因为我的连接许可设置为10，所以每次最高并发数为10。

其他方法

Semaphore还提供一些其他方法：

int availablePermits() ：返回此信号量中当前可用的许可证数。
int getQueueLength()：返回正在等待获取许可证的线程数。
boolean hasQueuedThreads() ：是否有线程正在等待获取许可证。
void reducePermits(int reduction) ：减少reduction个许可证。是个protected方法。
Collection getQueuedThreads() ：返回所有等待获取许可证的线程集合。是个protected方法。

ps：部分引用自http://ifeve.com/concurrency-semaphore/

巴特西

并发编程常用工具类(二) SymaPhore实现线程池

1.symaPhore简介

2.symaphore使用

其他方法

最新文章

热门文章