The Concurrency API introduced in Java 5 has a Semaphore class. It provides the same concept of semaphore as invented by Edsger Dijkstra. It can be used to control the access to the pool of resources.
Each Semaphore instance contains a count of permits. This count is usually equal to count of resources that are available for access. This count of permits is set only during the instantiation using the constructors of Semaphore class. There are no methods to directly increase this count.
Acquiring and releasing a permit
Once a permit is acquired, the access to the resource is granted. When the use of resource is over the permit should be released.
The acquire(), acquire(int permits), acquireUninterruptibly() and acquireUninterruptibly(int permits) methods are called before gaining the access to the resource. If the permit is acquired it means that the resource can be accessed. The acquire() and acquire(int permits) are blocking methods and InterruptedException needs to handled when they are called. When the thread calling these blocking methods is interrupted, the InterruptedException is thrown.
The acquire() method acquires a permit from the semaphore. It blocks until the permit is available or the thread waiting to acquire a permit is interrupted.
The acquire(int permits) method acquires the number of permits passed as argument to this method and blocks until all of them are available or the thread waiting to acquire them is interrupted.
The acquireUninterruptibly() method also acquires the permit and blocks until one is available even if the thread waiting to acquire a permit is interrupted.
The current count of available permits can be obtained by the method availablePermits(). If no permits are available, the blocking acquire methods will remain blocked until one is available.
The reducePermits(int reduction) method can be called to reduce the count of available permits. This method does not block for the given number of permits to become available for reduction.
When the use of a resource is over, a permit should be released. A permit is available after it is release. The release() and release(int permits) methods can be called for releasing permits.
The release() method releases a permit and it is available for acquisition after release. The release(int permits) releases the given number of permits. This method blocks until the given number of permits are available for release.
Ensure that an acquire method is called before calling release method as calling the latter before former would increase the permit count.
When a thread is blocked to acquire a permit, some other thread should call release method or interrupt the waiting thread, if applicable.
Fairness setting
When multiple threads are waiting to acquire a permit, there is no guarantee that the thread which went first in the waiting state will be first one to get the permit. This could lead a thread to starve. This situation can be avoided by setting the fairness setting as true while instantiating this class. The Semaphore(int permits, boolean fair) constructor should be used to set it.
The fairness setting can be found by calling the isFair() method.
Non-blocking methods to acquire a permit
The tryAcquire(), tryAcquire(int permits), tryAcquire(long timeout, TimeUnit unit) and tryAcquire(int permits, long timeout, TimeUnit unit) methods are the non-blocking methods used to obtain a permit.
The tryAcquire() method when called will try to obtain a permit if one is available. If there is no available permit, this method would not block and returns false. If the permit is acquired this method returns true. This method when called would try to acquire a permit immediately even if there are other threads waiting to obtain a permit. This behavior does not honor the fairness setting. In order to honor the fairness setting, the method tryAcquire(0, TimeUnit.SECONDS) should be called.
The tryAcquire(int permits) tries to obtain the number of permits passed as argument to this method only if all are available when this method is called. If the permits are acquired this method returns true otherwise it returns false. Like tryAcquire(), this method too does not honor the fairness setting. In order to honor the fairness setting the tryAcquire(permits, 0, TimeUnits.SECONDS) method should be called.
The tryAcquire(long timeout, TimeUnit unit) and tryAcquire(int permits, long timeout, TimeUnit unit) are non-blocking acquire methods with timeout. These method also honor the fairness setting. When they are invoked, they will wait for the specified time period to obtain a permit until the period is elapsed, the permit is acquired or the thread which called these methods is interrupted. The InterruptedException is thrown when the thread calling these methods is interrupted and it needs to be handled.
The tryAcquire(long timeout, TimeUnit unit) method tries to acquire only one permit. If the permit is acquired this method returns true. When the timeout is elapsed and no permit is acquired this method returns false.
The tryAcquire(int permits, long timeout, TimeUnit unit) method tries to acquire the number of permits passed as the first argument to this method. If the given number of permits are available this method returns true. If the given number of permits are not acquired before the timeout is elapsed this method returns false.
Other methods
The collection of threads waiting to acquire permit can be obtained by the getQueuedThreads() method. The getQueueLength() method returns the number of threads waiting to acquire permit. The hasQueuedThreads() method can be used to find out whether there are any threads waiting to acquire permit. If there are waiting threads, this method will return true.
Producer Consumer example using Semaphore
Using Semaphore class eliminates the need of calling wait() and notify() methods directly in the code. Listing 1 given below is the Producer-Consumer implementation using Semaphore class. A Queue instance is shared by the Producer and Consumer threads. The maximum of 5 items can be present in the FIFO Queue at a time, hence the number of permits defined while instantiating Semaphore class is 5. The putItem() method is called by the Producer thread whereas the takeItem() method is called by the Consumer thread. Since there is only one Producer thread which can keep waiting for the permit, the fairness is not set to true. The code ensures that the release() is not called before acquire(). The partial output of this code is given in Listing 2.
public class Queue {
private Semaphore semaphore;
private LinkedList<Integer> list;
public Queue(int maxItems) {
semaphore = new Semaphore(maxItems);
list = new LinkedList<Integer>();
}
public synchronized void putItem(int number) {
try {
if(semaphore.availablePermits() > 0) {
semaphore.acquire();
list.addLast(number);
System.out.println("Produce : " + number);
System.out.println("Elements in queue " +
list);
}
else {
System.out.println("Wait for consumer");
}
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
public synchronized void takeItem() {
if(!list.isEmpty()) {
System.out.println("Consume : " +
list.pollFirst());
semaphore.release();
}
else {
System.out.println("Wait for producer");
}
}
}
public class Producer implements Runnable {
private Queue queue;
private Random random = new Random();
public Producer(Queue queue) {
this.queue = queue;
}
public void run() {
while(true) {
queue.putItem(Math.abs(random.nextInt(1000)));
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class Consumer implements Runnable {
private Queue queue;
public Consumer(Queue queue) {
this.queue = queue;
}
public void run() {
while(true) {
queue.takeItem();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class ProducerConsumerDemo {
public static void main(String[] args) {
Queue queue = new Queue(5);
Executor producer =
Executors.newSingleThreadExecutor();
Executor consumer =
Executors.newSingleThreadExecutor();
producer.execute(new Producer(queue));
consumer.execute(new Consumer(queue));
}
}
Listing 1. Producer-Consumer example using Semaphore
Produce : 205 Elements in queue [205] Consume : 205 Produce : 866 Elements in queue [866] Consume : 866 Produce : 956 Elements in queue [956] Produce : 239 Elements in queue [956, 239] Consume : 956 Produce : 173 Elements in queue [239, 173] Produce : 723 Elements in queue [239, 173, 723] Consume : 239 Produce : 804 Elements in queue [173, 723, 804] Produce : 203 Elements in queue [173, 723, 804, 203] Consume : 173 Produce : 878 Elements in queue [723, 804, 203, 878] Produce : 118 Elements in queue [723, 804, 203, 878, 118] Consume : 723 Produce : 471 Elements in queue [804, 203, 878, 118, 471] Wait for consumer Consume : 804 Produce : 141 Elements in queue [203, 878, 118, 471, 141] Wait for consumer Consume : 203 Produce : 731 Elements in queue [878, 118, 471, 141, 731]
Listing 2. Partial output of code in Listing 1