java 多线程中Semaphore信号量

 Semaphore信号量：
           可以维护当前访问自身的线程个数，并提供了同步机制。使用Semaphore可以控制同时访问资源的线程个数，例如实现一个文件允许的并发访问数。
 
     获得一项前，每个线程必须从信号量获取许可(semaphore.acquire())，从而保证可以使用该项。该线程结束后，将项返回到池中并将许可返回到该信号量，从而允许其他线程获取该项。注意，调用 acquire() 时无法保持同步锁,因为这会阻止将项返回到池中。信号量封装所需的同步，以限制对池的访问，这同维持该池本身一致性所需的同步是分开的。  故，需要的同步的代码还需要同步，Semaphore只是将线程并发的数目进行限制的一种手段。
 
     单个信号量的Semaphore对象可以实现互斥锁的功能，并且可以是由一个线程获得了"锁"，再由另一个线程释放锁，这可应用于死锁恢复的一些场合。
 
 两个构造方法：
   public Semaphore(int permits)
    创建具有给定的许可数和非公平的公平设置的 Semaphore
   public Semaphore(int permits, boolean fair)
    创建具有给定的许可数和给定的公平设置的 Semaphore。
 
    permits - 初始的可用许可数目。此值可能为负数，在这种情况下，必须在授予任何获取前进行释放。
    fair - 如果此信号量保证在争用时按先进先出的顺序授予许可，则为 true；否则为 false。
 
 Semaphore类中的方法的详细说明，请参阅API

 

package com.etrip.concurrent.executor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * 
 * Semaphore信号量：
 *           可以维护当前访问自身的线程个数，并提供了同步机制。使用Semaphore可以控制同时访问资
 * 源的线程个数，例如实现一个文件允许的并发访问数。

 * 
 * 
 * @author longgangbai
 * 
 */
public class SemaphoreTest {
	
	public static void main(String[] args) {
		// 根据任务的数量具体的分配线程数的线程池
		int threadCount = 10;
		
		ExecutorService service = Executors.newFixedThreadPool(threadCount);
		
		// 创建信号量：最大值为3，即只能允许同时并发访问资源的线程数为3个
		final Semaphore sp = new Semaphore(3,true);
		
		for (int i = 0; i < threadCount; i++) {
			Runnable runnable = new Runnable() {
				public void run() {
					try {
						sp.acquire();// 获取许可
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
					System.out.println("线程" + Thread.currentThread().getName() + "进入，当前已有" + (3 - sp.availablePermits()) + "个并发");
					try {
						Thread.sleep(2000L);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
					System.out.println("线程" + Thread.currentThread().getName() + "即将离开");
					
					sp.release();// 释放许可,必须要释放许可，如果不释放会造成死锁现象。
					// 下面代码有时候执行不准确，因为其没有和上面的代码合成原子单元
					System.out.println("线程" + Thread.currentThread().getName() + "已离开，当前已有" + (3 - sp.availablePermits()) + "个并发");
				}
			};
			//向线程池中添加执行的线程
			service.execute(runnable);
		}
		//关闭线程池
		service.shutdown();
	}
}

 

 

使用信号量创建简单对象池

 

package com.etrip.concurrent.executor;

import java.util.concurrent.Semaphore;
/**
 * 通过信号量实现对象池的应用
 * 
 * @author longgangbai
 */
public class SemaphorePool  {
	   private static final int MAX_AVAILABLE = 100;
	   
	   // Not a particularly efficient data structure; just for demo

	   protected Object[] items =null;
	   protected boolean[] used = new boolean[MAX_AVAILABLE];
	   
	   private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);

	   
	   public SemaphorePool(Object[] items){
		   this.items=items;
	   }
	   
	   /**
	    * 获取对象池中的对象
	    * @return
	    * @throws InterruptedException
	    */
	   public Object getItem() throws InterruptedException {
	     available.acquire();
	     return getNextAvailableItem();
	   }
 
       /**
        * 返还对象到对象池中
        * @param x
        */
	   public void putItem(Object x) {
	     if (markAsUnused(x))
	       available.release();
	   }
     
	   /**
	    * 获取一个可用的对象
	    * @return
	    */
	   protected synchronized Object getNextAvailableItem() {
	     for (int i = 0; i < MAX_AVAILABLE; ++i) {
	       if (!used[i]) {
	          used[i] = true;
	          return items[i];
	       }
	     }
	     return null; // not reached
	   }

	   /**
	    * 标记对象为未用状态
	    * @param item
	    * @return
	    */
	   protected synchronized boolean markAsUnused(Object item) {
	     for (int i = 0; i < MAX_AVAILABLE; ++i) {
	       if (item == items[i]) {
	          if (used[i]) {
	            used[i] = false;
	            return true;
	          } else
	            return false;
	       }
	     }
	     return false;
	   }

	 }