线程的并发工具类之Semaphore原理和使用（八）

痛定思痛。 2022-12-19 13:15 87阅读 0赞

今天分析 Semaphore原理和使用：

**1、Semaphore**

Semaphore （信号量）是用来控制同时访问特定资源的线程数量，它通过协 调各个线程，以保证合理的使用公共资源。应用场景 Semaphore  可以用于做流 量控制，特别是公用资源有限的应用场景，比如数据库连接。假如有一个需求， 要读取几万个文件的数据，因为都是 IO 密集型任务，我们可以启动几十个线程 并发地读取，但是如果读到内存后，还需要存储到数据库中，而数据库的连接数 只有 10 个，这时我们必须控制只有  10  个线程同时获取数据库连接保存数据，否 则会报错无法获取数据库连接。这个时候，就可以使用 Semaphore  来做流量控 制。。Semaphore  的构造方法  Semaphore （ int permits ）接受一个整型的数字， 表示可用的许可证数量。Semaphore  的用法也很简单，首先线程使用  Semaphore 的 acquire() 方法获取一个许可证，使用完之后调用  release() 方法归还许可证。还 可以用 tryAcquire() 方法尝试获取许可证。

Semaphore  还提供一些其他方法，具体如下。

• intavailablePermits() ：返回此信号量中当前可用的许可证数。

• intgetQueueLength() ：返回正在等待获取许可证的线程数。

• booleanhasQueuedThreads() ：是否有线程正在等待获取许可证。 • void reducePermit （

s int reduction ）：减少  reduction  个许可证，是个  protected

方法。

• Collection getQueuedThreads() ：返回所有等待获取许可证的线程集合，是

个  protected  方法。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**2、用**  **Semaphore**  **实现数据库连接池**

参见代码 ,如   下

import java.sql.Connection;
    import java.util.LinkedList;
    import java.util.concurrent.Semaphore;
    
    /**
     *类说明：演示Semaphore用法，一个数据库连接池的实现
     */
    public class DBPoolSemaphore {
    	
    	private final static int POOL_SIZE = 10;
    	//两个指示器，分别表示池子还有可用连接和已用连接
    	private final Semaphore useful,useless;
    	//存放数据库连接的容器
    	private static LinkedList<Connection> pool = new LinkedList<Connection>();
    	//初始化池
    	static {
            for (int i = 0; i < POOL_SIZE; i++) {
                pool.addLast(SqlConnectImpl.fetchConnection());
            }
    	}
    	public DBPoolSemaphore() {
    		this.useful = new Semaphore(10);
    		this.useless = new Semaphore(0);
    	}
    	
    	/*归还连接*/
    	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 connection;
    		synchronized (pool) {
    			connection = pool.removeFirst();
    		}
    		useless.release();
    		return connection;
    	}
    	
    }

工具类：

*类说明：线程休眠辅助工具类
     */
    public class SleepTools {
    	
    	/**
    	 * 按秒休眠
    	 * @param seconds 秒数
    	 */
        public static final void second(int seconds) {
            try {
                TimeUnit.SECONDS.sleep(seconds);
            } catch (InterruptedException e) {
            }
        }
        
        /**
         * 按毫秒数休眠
         * @param seconds 毫秒数
         */
        public static final void ms(int seconds) {
            try {
                TimeUnit.MILLISECONDS.sleep(seconds);
            } catch (InterruptedException e) {
            }
        }
    }

测试类：

import cn.enjoyedu.tools.SleepTools;
    
    import java.sql.Connection;
    import java.util.Random;
    
    /**
     *类说明：测试数据库连接池
     */
    public class AppTest {
    
    	private static DBPoolSemaphore dbPool = new DBPoolSemaphore();
    	
    	private static class BusiThread extends Thread{
    		@Override
    		public void run() {
    			Random r = new Random();//让每个线程持有连接的时间不一样
    			long start = System.currentTimeMillis();
    			try {
    				Connection connect = dbPool.takeConnect();
    				System.out.println("Thread_"+Thread.currentThread().getId()
    						+"_获取数据库连接共耗时【"+(System.currentTimeMillis()-start)+"】ms.");
    				SleepTools.ms(100+r.nextInt(100));//模拟业务操作，线程持有连接查询数据
    				System.out.println("查询数据完成，归还连接！");
    				dbPool.returnConnect(connect);
    			} catch (InterruptedException e) {
    			}
    		}
    	}
    	
    	public static void main(String[] args) {
            for (int i = 0; i < 50; i++) {
                Thread thread = new BusiThread();
                thread.start();
            }
    	}
    	
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**3、Semaphore**  **注意事项**

参见代码如   下

import cn.enjoyedu.tools.SleepTools;
    
    import java.sql.Connection;
    import java.util.LinkedList;
    import java.util.Random;
    import java.util.concurrent.Semaphore;
    
    /**
     *类说明：演示Semaphore用法，一个数据库连接池的实现
     */
    public class DBPoolNoUseless {
    
    	private final static int POOL_SIZE = 10;
    	private final Semaphore useful;
    	//存放数据库连接的容器
    	private static LinkedList<Connection> pool = new LinkedList<Connection>();
    	//初始化池
    	static {
            for (int i = 0; i < POOL_SIZE; i++) {
                pool.addLast(SqlConnectImpl.fetchConnection());
            }
    	}
    	public DBPoolNoUseless() {
    		this.useful = new Semaphore(10);
    	}
    	
    	/*归还连接*/
    	public void returnConnect(Connection connection) throws InterruptedException {
    		if(connection!=null) {
    			System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接!!"
    					+"可用连接数："+useful.availablePermits());
    			synchronized (pool) {
    				pool.addLast(connection);
    			}
    			useful.release();
    		}
    	}
    	
    	/*从池子拿连接*/
    	public Connection takeConnect() throws InterruptedException {
    		useful.acquire();
    		Connection connection;
    		synchronized (pool) {
    			connection = pool.removeFirst();
    		}
    		return connection;
    	}
    
    	private static DBPoolNoUseless dbPoolNoUseless = new DBPoolNoUseless();
    
    	private static class BusiThread extends Thread{
    		@Override
    		public void run() {
    			Random r = new Random();//让每个线程持有连接的时间不一样
    			long start = System.currentTimeMillis();
    			try {
    				System.out.println("Thread_"+Thread.currentThread().getId()
    						+"_获取数据库连接共耗时【"+(System.currentTimeMillis()-start)+"】ms.");
    				SleepTools.ms(100+r.nextInt(100));//模拟业务操作，线程持有连接查询数据
    				System.out.println("查询数据完成，归还连接！");
    				dbPoolNoUseless.returnConnect(new SqlConnectImpl());
    			} catch (InterruptedException e) {
    			}
    		}
    	}
    
    	public static void main(String[] args) {
    		for (int i = 0; i < 50; i++) {
    			Thread thread = new BusiThread();
    			thread.start();
    		}
    	}
    	
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

到此结束：

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/123669c36b0b45e180d7f2de4e0698fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/aa9069c1b1d3471995403245e86319c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201107225924263.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70