Comprehensive guide to using Semaphore for controlling access to limited resources and implementing resource pools.

1. Semaphore Basics

Basic Semaphore Usage

import java.util.concurrent.Semaphore;
public class BasicSemaphoreExample {
public static void main(String[] args) {
// Create a semaphore with 3 permits
Semaphore semaphore = new Semaphore(3);
// Create multiple threads that need to acquire permits
for (int i = 1; i <= 10; i++) {
final int threadId = i;
new Thread(() -> {
try {
System.out.println("Thread " + threadId + " waiting for permit");
semaphore.acquire(); // Block until permit is available
System.out.println("Thread " + threadId + " acquired permit");
// Simulate work with the resource
Thread.sleep(2000);
System.out.println("Thread " + threadId + " releasing permit");
semaphore.release();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
// Stagger thread creation
try { Thread.sleep(200); } catch (InterruptedException e) {}
}
}
}

Semaphore with Try-Acquire

public class TryAcquireSemaphore {
private final Semaphore semaphore;
public TryAcquireSemaphore(int permits) {
this.semaphore = new Semaphore(permits);
}
public boolean tryAccessResource() {
// Non-blocking attempt to acquire permit
if (semaphore.tryAcquire()) {
try {
System.out.println(Thread.currentThread().getName() + " acquired permit");
// Use the resource
return true;
} finally {
semaphore.release();
}
} else {
System.out.println(Thread.currentThread().getName() + " could not acquire permit");
return false;
}
}
public boolean tryAccessResourceWithTimeout(long timeout, TimeUnit unit) 
throws InterruptedException {
// Try to acquire with timeout
if (semaphore.tryAcquire(timeout, unit)) {
try {
System.out.println(Thread.currentThread().getName() + " acquired permit after waiting");
// Use the resource
return true;
} finally {
semaphore.release();
}
} else {
System.out.println(Thread.currentThread().getName() + " timeout waiting for permit");
return false;
}
}
public void demonstrateTryAcquire() throws InterruptedException {
Semaphore sem = new Semaphore(2);
// Acquire both permits
sem.acquire(2);
Thread thread = new Thread(() -> {
boolean acquired = sem.tryAcquire();
System.out.println("Try acquire result: " + acquired);
try {
boolean acquiredWithTimeout = sem.tryAcquire(1, TimeUnit.SECONDS);
System.out.println("Try acquire with timeout result: " + acquiredWithTimeout);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
thread.start();
Thread.sleep(500);
// Release one permit
sem.release();
thread.join();
}
}

2. Database Connection Pool

Basic Connection Pool with Semaphore

public class DatabaseConnectionPool {
private final Semaphore semaphore;
private final List<Connection> availableConnections;
private final List<Connection> usedConnections;
private final int poolSize;
public DatabaseConnectionPool(int poolSize) {
this.poolSize = poolSize;
this.semaphore = new Semaphore(poolSize, true); // Fair semaphore
this.availableConnections = new ArrayList<>(poolSize);
this.usedConnections = new ArrayList<>(poolSize);
initializePool();
}
private void initializePool() {
for (int i = 0; i < poolSize; i++) {
availableConnections.add(createConnection());
}
System.out.println("Initialized connection pool with " + poolSize + " connections");
}
public Connection getConnection() throws InterruptedException {
semaphore.acquire(); // Wait for available permit
return getAvailableConnection();
}
public Connection getConnection(long timeout, TimeUnit unit) 
throws InterruptedException, TimeoutException {
if (semaphore.tryAcquire(timeout, unit)) {
return getAvailableConnection();
} else {
throw new TimeoutException("Timeout waiting for database connection");
}
}
private synchronized Connection getAvailableConnection() {
Connection connection = availableConnections.remove(availableConnections.size() - 1);
usedConnections.add(connection);
System.out.println("Connection acquired. Available: " + availableConnections.size() + 
", Used: " + usedConnections.size());
return connection;
}
public synchronized void releaseConnection(Connection connection) {
if (usedConnections.remove(connection)) {
availableConnections.add(connection);
semaphore.release(); // Release the permit
System.out.println("Connection released. Available: " + availableConnections.size() + 
", Used: " + usedConnections.size());
}
}
public synchronized void closeAllConnections() {
for (Connection conn : availableConnections) {
closeConnection(conn);
}
for (Connection conn : usedConnections) {
closeConnection(conn);
}
availableConnections.clear();
usedConnections.clear();
}
public synchronized int getAvailableConnectionsCount() {
return availableConnections.size();
}
public synchronized int getUsedConnectionsCount() {
return usedConnections.size();
}
// Mock connection implementation
private Connection createConnection() {
return new MockConnection();
}
private void closeConnection(Connection connection) {
// Close the connection
System.out.println("Closing connection: " + connection);
}
// Mock Connection class
static class MockConnection implements Connection {
private final String id = UUID.randomUUID().toString().substring(0, 8);
@Override
public String toString() {
return "Connection-" + id;
}
// Implement other Connection methods as needed...
@Override public void close() {}
@Override public boolean isClosed() { return false; }
// ... other Connection interface methods
}
// Demonstration
public static void main(String[] args) throws Exception {
DatabaseConnectionPool pool = new DatabaseConnectionPool(3);
ExecutorService executor = Executors.newFixedThreadPool(10);
for (int i = 1; i <= 10; i++) {
final int taskId = i;
executor.submit(() -> {
try {
System.out.println("Task " + taskId + " waiting for connection");
Connection conn = pool.getConnection();
try {
System.out.println("Task " + taskId + " using connection: " + conn);
// Simulate database operation
Thread.sleep(2000);
System.out.println("Task " + taskId + " completed work");
} finally {
pool.releaseConnection(conn);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread.sleep(500); // Stagger task submission
}
executor.shutdown();
executor.awaitTermination(30, TimeUnit.SECONDS);
pool.closeAllConnections();
}
}

3. Advanced Connection Pool with Validation

Robust Connection Pool

public class RobustConnectionPool {
private final Semaphore semaphore;
private final BlockingQueue<PooledConnection> availableConnections;
private final Set<PooledConnection> usedConnections;
private final int maxPoolSize;
private final ConnectionFactory connectionFactory;
private volatile boolean closed = false;
public RobustConnectionPool(int maxPoolSize, ConnectionFactory connectionFactory) {
this.maxPoolSize = maxPoolSize;
this.connectionFactory = connectionFactory;
this.semaphore = new Semaphore(maxPoolSize, true);
this.availableConnections = new LinkedBlockingQueue<>(maxPoolSize);
this.usedConnections = new ConcurrentHashSet<>();
initializePool();
}
private void initializePool() {
for (int i = 0; i < Math.min(2, maxPoolSize); i++) { // Start with 2 connections
createAndAddConnection();
}
}
private void createAndAddConnection() {
try {
Connection rawConnection = connectionFactory.createConnection();
PooledConnection pooledConnection = new PooledConnection(rawConnection);
availableConnections.offer(pooledConnection);
} catch (SQLException e) {
System.err.println("Failed to create connection: " + e.getMessage());
}
}
public Connection getConnection() throws SQLException, InterruptedException {
if (closed) {
throw new SQLException("Connection pool is closed");
}
semaphore.acquire();
return getAvailableConnection();
}
public Connection getConnection(long timeout, TimeUnit unit) 
throws SQLException, InterruptedException, TimeoutException {
if (closed) {
throw new SQLException("Connection pool is closed");
}
if (!semaphore.tryAcquire(timeout, unit)) {
throw new TimeoutException("Timeout waiting for connection");
}
return getAvailableConnection();
}
private Connection getAvailableConnection() throws SQLException {
try {
PooledConnection connection = availableConnections.poll(1, TimeUnit.SECONDS);
if (connection != null && connection.isValid(1)) {
usedConnections.add(connection);
return connection;
} else {
// Create new connection if none available or existing is invalid
Connection newConnection = connectionFactory.createConnection();
PooledConnection pooledConnection = new PooledConnection(newConnection);
usedConnections.add(pooledConnection);
return pooledConnection;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SQLException("Interrupted while waiting for connection", e);
} catch (SQLException e) {
semaphore.release(); // Release permit if connection creation failed
throw e;
}
}
private void returnConnection(PooledConnection connection) {
if (usedConnections.remove(connection)) {
if (connection.isValid(1) && availableConnections.size() < maxPoolSize) {
availableConnections.offer(connection);
} else {
closeConnectionQuietly(connection.getRealConnection());
// Create replacement connection if pool is not full
if (availableConnections.size() < maxPoolSize && !closed) {
createAndAddConnection();
}
}
semaphore.release();
}
}
public void close() {
closed = true;
// Close all connections
availableConnections.forEach(conn -> 
closeConnectionQuietly(conn.getRealConnection()));
availableConnections.clear();
usedConnections.forEach(conn -> 
closeConnectionQuietly(conn.getRealConnection()));
usedConnections.clear();
}
private void closeConnectionQuietly(Connection connection) {
try {
if (!connection.isClosed()) {
connection.close();
}
} catch (SQLException e) {
// Log but don't throw
System.err.println("Error closing connection: " + e.getMessage());
}
}
public int getAvailableCount() {
return availableConnections.size();
}
public int getUsedCount() {
return usedConnections.size();
}
public int getTotalCount() {
return getAvailableCount() + getUsedCount();
}
// PooledConnection wrapper
private class PooledConnection implements Connection {
private final Connection realConnection;
private volatile boolean isClosed = false;
public PooledConnection(Connection realConnection) {
this.realConnection = realConnection;
}
public Connection getRealConnection() {
return realConnection;
}
@Override
public void close() throws SQLException {
if (!isClosed) {
isClosed = true;
returnConnection(this);
}
}
public boolean isValid(int timeout) {
try {
return !realConnection.isClosed() && realConnection.isValid(timeout);
} catch (SQLException e) {
return false;
}
}
// Delegate all other Connection methods to realConnection
@Override public Statement createStatement() throws SQLException { 
return realConnection.createStatement(); 
}
@Override public PreparedStatement prepareStatement(String sql) throws SQLException { 
return realConnection.prepareStatement(sql); 
}
// ... implement all other Connection interface methods
}
// Connection factory interface
public interface ConnectionFactory {
Connection createConnection() throws SQLException;
}
}

4. Thread Pool with Semaphore Control

Bounded Thread Pool

public class BoundedThreadPool {
private final ExecutorService executor;
private final Semaphore semaphore;
private final int maxConcurrentTasks;
public BoundedThreadPool(int poolSize, int maxConcurrentTasks) {
this.executor = Executors.newFixedThreadPool(poolSize);
this.semaphore = new Semaphore(maxConcurrentTasks);
this.maxConcurrentTasks = maxConcurrentTasks;
}
public <T> Future<T> submit(Callable<T> task) {
return executor.submit(() -> {
semaphore.acquire();
try {
return task.call();
} finally {
semaphore.release();
}
});
}
public Future<?> submit(Runnable task) {
return executor.submit(() -> {
semaphore.acquire();
try {
task.run();
} finally {
semaphore.release();
}
});
}
public void shutdown() {
executor.shutdown();
}
public boolean awaitTermination(long timeout, TimeUnit unit) 
throws InterruptedException {
return executor.awaitTermination(timeout, unit);
}
public int getAvailablePermits() {
return semaphore.availablePermits();
}
public int getQueueLength() {
return semaphore.getQueueLength();
}
// Demonstration
public static void main(String[] args) throws Exception {
BoundedThreadPool boundedPool = new BoundedThreadPool(10, 3); // Max 3 concurrent tasks
List<Future<String>> futures = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
final int taskId = i;
Future<String> future = boundedPool.submit(() -> {
System.out.println("Task " + taskId + " started. Available permits: " + 
boundedPool.getAvailablePermits());
Thread.sleep(3000); // Simulate work
return "Result from task " + taskId;
});
futures.add(future);
Thread.sleep(500); // Stagger submission
}
// Collect results
for (Future<String> future : futures) {
System.out.println("Completed: " + future.get());
}
boundedPool.shutdown();
boundedPool.awaitTermination(1, TimeUnit.MINUTES);
}
}

5. Rate Limiting with Semaphore

API Rate Limiter

public class RateLimiter {
private final Semaphore semaphore;
private final int maxPermits;
private final ScheduledExecutorService scheduler;
public RateLimiter(int permitsPerSecond) {
this.maxPermits = permitsPerSecond;
this.semaphore = new Semaphore(permitsPerSecond);
this.scheduler = Executors.newScheduledThreadPool(1);
// Schedule permit replenishment
scheduler.scheduleAtFixedRate(() -> {
int currentPermits = semaphore.availablePermits();
int permitsToRelease = maxPermits - currentPermits;
if (permitsToRelease > 0) {
semaphore.release(permitsToRelease);
}
}, 1, 1, TimeUnit.SECONDS);
}
public boolean tryAcquire() {
return semaphore.tryAcquire();
}
public void acquire() throws InterruptedException {
semaphore.acquire();
}
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
return semaphore.tryAcquire(timeout, unit);
}
public void shutdown() {
scheduler.shutdown();
}
// Demonstration
public static void main(String[] args) throws Exception {
RateLimiter rateLimiter = new RateLimiter(5); // 5 requests per second
ExecutorService executor = Executors.newFixedThreadPool(10);
for (int i = 1; i <= 20; i++) {
final int requestId = i;
executor.submit(() -> {
if (rateLimiter.tryAcquire()) {
System.out.println("Request " + requestId + " allowed at " + 
System.currentTimeMillis());
// Process request
} else {
System.out.println("Request " + requestId + " rate limited");
}
});
Thread.sleep(100); // Stagger requests
}
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
rateLimiter.shutdown();
}
}

Token Bucket Rate Limiter

public class TokenBucketRateLimiter {
private final Semaphore semaphore;
private final int capacity;
private final ScheduledExecutorService refillScheduler;
public TokenBucketRateLimiter(int capacity, int refillRate, TimeUnit refillUnit) {
this.capacity = capacity;
this.semaphore = new Semaphore(capacity);
this.refillScheduler = Executors.newScheduledThreadPool(1);
// Calculate refill period
long refillPeriod = refillUnit.toNanos(1) / refillRate;
// Schedule token refill
refillScheduler.scheduleAtFixedRate(() -> {
if (semaphore.availablePermits() < capacity) {
semaphore.release();
}
}, refillPeriod, refillPeriod, TimeUnit.NANOSECONDS);
}
public boolean tryConsume() {
return semaphore.tryAcquire();
}
public void consume() throws InterruptedException {
semaphore.acquire();
}
public boolean tryConsume(int permits) {
return semaphore.tryAcquire(permits);
}
public void consume(int permits) throws InterruptedException {
semaphore.acquire(permits);
}
public int getAvailableTokens() {
return semaphore.availablePermits();
}
public void shutdown() {
refillScheduler.shutdown();
}
}

6. Resource Pool for Expensive Objects

Object Pool with Semaphore

public class ObjectPool<T> {
private final Semaphore semaphore;
private final BlockingQueue<T> availableObjects;
private final Set<T> usedObjects;
private final ObjectFactory<T> objectFactory;
private final int maxSize;
private volatile boolean closed = false;
public ObjectPool(int maxSize, ObjectFactory<T> objectFactory) {
this.maxSize = maxSize;
this.objectFactory = objectFactory;
this.semaphore = new Semaphore(maxSize, true);
this.availableObjects = new LinkedBlockingQueue<>(maxSize);
this.usedObjects = ConcurrentHashMap.newKeySet();
initializePool();
}
private void initializePool() {
for (int i = 0; i < Math.min(3, maxSize); i++) {
createAndAddObject();
}
}
private void createAndAddObject() {
try {
T object = objectFactory.create();
availableObjects.offer(object);
} catch (Exception e) {
System.err.println("Failed to create object: " + e.getMessage());
}
}
public T acquire() throws InterruptedException {
if (closed) {
throw new IllegalStateException("Pool is closed");
}
semaphore.acquire();
return acquireObject();
}
public T acquire(long timeout, TimeUnit unit) 
throws InterruptedException, TimeoutException {
if (closed) {
throw new IllegalStateException("Pool is closed");
}
if (!semaphore.tryAcquire(timeout, unit)) {
throw new TimeoutException("Timeout waiting for object");
}
return acquireObject();
}
private T acquireObject() {
try {
T object = availableObjects.poll(1, TimeUnit.SECONDS);
if (object != null && objectFactory.validate(object)) {
usedObjects.add(object);
return object;
} else {
// Create new object
T newObject = objectFactory.create();
usedObjects.add(newObject);
return newObject;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
semaphore.release(); // Release permit on interruption
throw new RuntimeException("Interrupted while acquiring object", e);
} catch (Exception e) {
semaphore.release(); // Release permit on creation failure
throw new RuntimeException("Failed to acquire object", e);
}
}
public void release(T object) {
if (usedObjects.remove(object)) {
if (objectFactory.validate(object) && availableObjects.size() < maxSize) {
availableObjects.offer(object);
} else {
objectFactory.destroy(object);
if (!closed && availableObjects.size() < maxSize) {
createAndAddObject();
}
}
semaphore.release();
}
}
public void close() {
closed = true;
// Destroy all objects
availableObjects.forEach(objectFactory::destroy);
availableObjects.clear();
usedObjects.forEach(objectFactory::destroy);
usedObjects.clear();
}
public int getAvailableCount() {
return availableObjects.size();
}
public int getUsedCount() {
return usedObjects.size();
}
// Object factory interface
public interface ObjectFactory<T> {
T create();
boolean validate(T object);
void destroy(T object);
}
// Example: Expensive parser pool
public static class ParserPool extends ObjectPool<DocumentBuilder> {
public ParserPool(int maxSize) {
super(maxSize, new ParserFactory());
}
public String parseXml(String xml) throws Exception {
DocumentBuilder parser = acquire();
try {
Document doc = parser.parse(new InputSource(new StringReader(xml)));
return doc.getDocumentElement().getTextContent();
} finally {
release(parser);
}
}
}
static class ParserFactory implements ObjectFactory<DocumentBuilder> {
@Override
public DocumentBuilder create() {
try {
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
return factory.newDocumentBuilder();
} catch (ParserConfigurationException e) {
throw new RuntimeException("Failed to create parser", e);
}
}
@Override
public boolean validate(DocumentBuilder parser) {
return parser != null;
}
@Override
public void destroy(DocumentBuilder parser) {
// Parser doesn't need explicit cleanup
}
}
}

7. Producer-Consumer with Bounded Buffer

Bounded Buffer with Semaphore

public class BoundedBuffer<T> {
private final Semaphore emptySlots;
private final Semaphore fullSlots;
private final Queue<T> buffer;
private final int capacity;
public BoundedBuffer(int capacity) {
this.capacity = capacity;
this.emptySlots = new Semaphore(capacity);
this.fullSlots = new Semaphore(0);
this.buffer = new LinkedList<>();
}
public void put(T item) throws InterruptedException {
emptySlots.acquire(); // Wait for empty slot
synchronized (this) {
buffer.offer(item);
}
fullSlots.release(); // Signal that a slot is full
}
public T take() throws InterruptedException {
fullSlots.acquire(); // Wait for full slot
T item;
synchronized (this) {
item = buffer.poll();
}
emptySlots.release(); // Signal that a slot is empty
return item;
}
public boolean offer(T item) {
if (emptySlots.tryAcquire()) {
synchronized (this) {
buffer.offer(item);
}
fullSlots.release();
return true;
}
return false;
}
public T poll() {
if (fullSlots.tryAcquire()) {
T item;
synchronized (this) {
item = buffer.poll();
}
emptySlots.release();
return item;
}
return null;
}
public int size() {
synchronized (this) {
return buffer.size();
}
}
public boolean isEmpty() {
synchronized (this) {
return buffer.isEmpty();
}
}
public boolean isFull() {
return emptySlots.availablePermits() == 0;
}
// Demonstration
public static void main(String[] args) throws Exception {
BoundedBuffer<Integer> buffer = new BoundedBuffer<>(5);
// Producer
Thread producer = new Thread(() -> {
try {
for (int i = 1; i <= 10; i++) {
buffer.put(i);
System.out.println("Produced: " + i + " (Buffer size: " + buffer.size() + ")");
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
// Consumer
Thread consumer = new Thread(() -> {
try {
for (int i = 1; i <= 10; i++) {
Integer item = buffer.take();
System.out.println("Consumed: " + item + " (Buffer size: " + buffer.size() + ")");
Thread.sleep(200);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
}

8. Best Practices and Common Patterns

Error Handling and Resource Management

public class SemaphoreBestPractices {
// ✅ GOOD: Always release in finally block
public void properResourceHandling(Semaphore semaphore) {
try {
semaphore.acquire();
// Use the resource
performCriticalOperation();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
// Handle interruption
} finally {
semaphore.release();
}
}
// ✅ GOOD: Use try-with-resources pattern
public void withAutoCloseable(Semaphore semaphore) throws Exception {
try (SemaphoreResource resource = new SemaphoreResource(semaphore)) {
resource.use();
} // Semaphore automatically released
}
// ✅ GOOD: Handle timeouts gracefully
public void withTimeoutHandling(Semaphore semaphore) {
try {
if (semaphore.tryAcquire(1, TimeUnit.SECONDS)) {
try {
performTimeSensitiveOperation();
} finally {
semaphore.release();
}
} else {
handleTimeout();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
handleInterruption();
}
}
// ❌ BAD: Might not release on exception
public void badPractice(Semaphore semaphore) throws InterruptedException {
semaphore.acquire();
performRiskyOperation(); // If this throws, semaphore won't be released
semaphore.release();
}
// Helper class for auto-closeable pattern
static class SemaphoreResource implements AutoCloseable {
private final Semaphore semaphore;
private boolean acquired = false;
public SemaphoreResource(Semaphore semaphore) throws InterruptedException {
this.semaphore = semaphore;
this.semaphore.acquire();
this.acquired = true;
}
public void use() {
if (!acquired) {
throw new IllegalStateException("Resource not acquired");
}
// Use the resource
}
@Override
public void close() {
if (acquired) {
semaphore.release();
acquired = false;
}
}
}
private void performCriticalOperation() {
// Critical section
}
private void performTimeSensitiveOperation() {
// Time-sensitive operation
}
private void performRiskyOperation() {
// Operation that might throw exception
}
private void handleTimeout() {
System.out.println("Operation timed out");
}
private void handleInterruption() {
System.out.println("Operation was interrupted");
}
}

Key Benefits of Semaphore for Resource Pooling

Semaphores provide a robust mechanism for controlling access to limited resources, making them ideal for implementing resource pools, rate limiters, and bounded buffers in concurrent applications.