Java 并发进阶：Semaphore 在连接池和资源池中的应用与性能分析

Java 并发进阶：Semaphore 在连接池和资源池中的应用与性能分析

你好，我是老码农。今天我们来聊聊 Java 并发编程中一个非常实用的工具——Semaphore，它在连接池和资源池等场景中扮演着关键角色。如果你已经对并发编程有一定了解，想进一步提升技术水平，那么这篇文章绝对适合你。

1. 什么是 Semaphore？

Semaphore（信号量）是 Java 并发包 java.util.concurrent 中的一个同步器，用于控制对共享资源的访问。你可以把它想象成一个停车场，Semaphore就像停车位的数量。当线程需要访问共享资源时，它需要先“获取”一个信号量（相当于获得一个停车位），才能继续执行。当线程使用完资源后，它需要“释放”信号量（相当于释放一个停车位），以便其他线程使用。

Semaphore 主要维护了一个许可证（permits） 的计数器。你可以通过构造函数指定许可证的数量。当一个线程调用 acquire() 方法时，它会尝试获取一个许可证。如果许可证计数器大于0，则线程成功获取一个许可证，计数器减1，线程继续执行。如果许可证计数器为0，则线程会被阻塞，直到其他线程释放许可证。当一个线程调用 release() 方法时，它会释放一个许可证，计数器加1。如果此时有线程在等待获取许可证，那么其中一个线程会被唤醒，并获取一个许可证。

1.1 核心方法

• Semaphore(int permits): 构造函数，用于初始化信号量，permits 指定了许可证的数量。
• acquire(): 获取一个许可证。如果许可证不足，则阻塞当前线程，直到获取到许可证。该方法有多个重载版本，例如 acquire(int permits)，用于获取指定数量的许可证。
• release(): 释放一个许可证。该方法有多个重载版本，例如 release(int permits)，用于释放指定数量的许可证。
• tryAcquire(): 尝试获取一个许可证，如果获取成功，返回 true，否则返回 false，不会阻塞当前线程。该方法有多个重载版本，例如 tryAcquire(long timeout, TimeUnit unit)，在指定时间内尝试获取许可证。
• availablePermits(): 获取当前可用的许可证数量。
• drainPermits(): 获取并返回立即可用的所有许可证，并将许可证数量设置为0。

1.2 简单示例

import java.util.concurrent.Semaphore;

public class SemaphoreExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个 Semaphore，初始许可证数量为 2
        Semaphore semaphore = new Semaphore(2);

        // 创建 5 个线程
        for (int i = 0; i < 5; i++) {
            int threadId = i;
            new Thread(() -> {
                try {
                    // 获取一个许可证
                    semaphore.acquire();
                    System.out.println("线程 " + threadId + " 获取了许可证，开始执行任务...");
                    // 模拟任务执行时间
                    Thread.sleep(2000);
                    System.out.println("线程 " + threadId + " 执行任务完毕，释放许可证...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 释放许可证
                    semaphore.release();
                }
            }).start();
        }
    }
}

在这个例子中，我们创建了一个初始许可证数量为 2 的 Semaphore。这意味着最多只有两个线程可以同时执行任务。其他线程会被阻塞，直到有线程释放许可证。你可以运行这段代码，观察线程的执行顺序，体会 Semaphore 的作用。

2. Semaphore 在连接池中的应用

连接池是数据库访问中常用的技术，它通过预先创建一定数量的数据库连接，并将这些连接存储在一个池中，以便在需要时快速获取连接，避免了频繁创建和销毁连接的开销。Semaphore 在连接池中可以用来限制并发连接的数量，防止数据库过载。

2.1 连接池的简单实现

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;

public class ConnectionPool {
    private final String url;
    private final String user;
    private final String password;
    private final int initialPoolSize;
    private final List<Connection> connectionPool;
    private final Semaphore semaphore;

    public ConnectionPool(String url, String user, String password, int initialPoolSize) throws SQLException {
        this.url = url;
        this.user = user;
        this.password = password;
        this.initialPoolSize = initialPoolSize;
        this.connectionPool = new ArrayList<>(initialPoolSize);
        this.semaphore = new Semaphore(initialPoolSize);
        initializePool();
    }

    private void initializePool() throws SQLException {
        for (int i = 0; i < initialPoolSize; i++) {
            Connection connection = createConnection();
            connectionPool.add(connection);
        }
    }

    private Connection createConnection() throws SQLException {
        // 实际应用中，这里应该使用数据库连接的配置信息来创建连接
        return DriverManager.getConnection(url, user, password);
    }

    public Connection getConnection() throws InterruptedException {
        semaphore.acquire(); // 获取一个许可证
        Connection connection = null;
        synchronized (connectionPool) {
            if (!connectionPool.isEmpty()) {
                connection = connectionPool.remove(0);
            }
        }
        if (connection == null) {
            // 如果连接池中没有可用的连接，可以考虑创建新的连接
            try {
                connection = createConnection();
            } catch (SQLException e) {
                semaphore.release(); // 如果创建连接失败，释放许可证
                throw new RuntimeException("Failed to create a new connection.", e);
            }
        }
        return connection;
    }

    public void releaseConnection(Connection connection) {
        if (connection != null) {
            synchronized (connectionPool) {
                connectionPool.add(connection);
            }
            semaphore.release(); // 释放一个许可证
        }
    }

    public void close() throws SQLException {
        synchronized (connectionPool) {
            for (Connection connection : connectionPool) {
                connection.close();
            }
            connectionPool.clear();
        }
    }

    public int getAvailableConnections() {
        return semaphore.availablePermits();
    }
}

在这个例子中，ConnectionPool 类维护了一个数据库连接池。Semaphore 用于控制并发连接的数量，initialPoolSize 指定了连接池的大小，也就是 Semaphore 的初始许可证数量。getConnection() 方法用于获取一个数据库连接，它首先调用 semaphore.acquire() 获取一个许可证。如果许可证不足，则阻塞当前线程，直到获取到许可证。然后，从连接池中获取一个连接。如果连接池中没有可用的连接，则创建一个新的连接。releaseConnection() 方法用于释放一个数据库连接，它将连接放回连接池，并调用 semaphore.release() 释放一个许可证。

2.2 使用示例

import java.sql.Connection;
import java.sql.SQLException;

public class ConnectionPoolExample {
    public static void main(String[] args) throws SQLException, InterruptedException {
        // 数据库连接信息
        String url = "jdbc:mysql://localhost:3306/your_database";
        String user = "your_user";
        String password = "your_password";
        int initialPoolSize = 3;

        // 创建连接池
        ConnectionPool connectionPool = new ConnectionPool(url, user, password, initialPoolSize);

        // 创建 5 个线程，模拟并发访问数据库
        for (int i = 0; i < 5; i++) {
            int threadId = i;
            new Thread(() -> {
                Connection connection = null;
                try {
                    // 获取连接
                    connection = connectionPool.getConnection();
                    System.out.println("线程 " + threadId + " 获取了连接，可用连接数：" + connectionPool.getAvailableConnections());
                    // 模拟数据库操作
                    Thread.sleep(1000);
                } catch (InterruptedException | SQLException e) {
                    e.printStackTrace();
                } finally {
                    // 释放连接
                    if (connection != null) {
                        connectionPool.releaseConnection(connection);
                        System.out.println("线程 " + threadId + " 释放了连接，可用连接数：" + connectionPool.getAvailableConnections());
                    }
                }
            }).start();
        }

        // 等待一段时间，让线程执行完毕
        Thread.sleep(5000);

        // 关闭连接池
        connectionPool.close();
    }
}

在这个例子中，我们创建了一个 ConnectionPool，并使用多个线程并发地从连接池中获取和释放连接。你可以运行这段代码，观察线程的执行顺序，体会 Semaphore 在限制并发连接数量方面的作用。你会发现，最多只有 initialPoolSize 个线程可以同时获得数据库连接。

3. Semaphore 在资源池中的应用

资源池是一种管理有限资源的机制，例如线程池、对象池等。Semaphore 可以用来控制对资源池中资源的访问，确保资源的合理分配和使用。

3.1 资源池的简单实现

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;

public class ResourcePool<T> {
    private final List<T> resources;
    private final Semaphore semaphore;
    private final ResourceFactory<T> resourceFactory;

    public interface ResourceFactory<T> {
        T create();
    }

    public ResourcePool(int poolSize, ResourceFactory<T> resourceFactory) {
        this.resources = new ArrayList<>(poolSize);
        this.semaphore = new Semaphore(poolSize);
        this.resourceFactory = resourceFactory;
        initializePool(poolSize);
    }

    private void initializePool(int poolSize) {
        for (int i = 0; i < poolSize; i++) {
            T resource = resourceFactory.create();
            resources.add(resource);
        }
    }

    public T getResource() throws InterruptedException {
        semaphore.acquire(); // 获取一个许可证
        T resource = null;
        synchronized (resources) {
            if (!resources.isEmpty()) {
                resource = resources.remove(0);
            }
        }
        if (resource == null) {
            // 如果资源池中没有可用的资源，可以考虑创建新的资源
            resource = resourceFactory.create();
        }
        return resource;
    }

    public void releaseResource(T resource) {
        if (resource != null) {
            synchronized (resources) {
                resources.add(resource);
            }
            semaphore.release(); // 释放一个许可证
        }
    }

    public int getAvailableResources() {
        return semaphore.availablePermits();
    }
}

在这个例子中，ResourcePool 类维护了一个资源池。Semaphore 用于控制并发访问资源的数量。ResourceFactory 接口用于创建资源。getResource() 方法用于获取一个资源，它首先调用 semaphore.acquire() 获取一个许可证。然后，从资源池中获取一个资源。如果资源池中没有可用的资源，则创建一个新的资源。releaseResource() 方法用于释放一个资源，它将资源放回资源池，并调用 semaphore.release() 释放一个许可证。

3.2 使用示例

public class ResourcePoolExample {
    public static void main(String[] args) throws InterruptedException {
        // 定义资源工厂
        ResourcePool.ResourceFactory<String> stringFactory = () -> {
            // 模拟创建资源
            return "Resource-" + System.currentTimeMillis();
        };

        // 创建资源池
        int poolSize = 3;
        ResourcePool<String> resourcePool = new ResourcePool<>(poolSize, stringFactory);

        // 创建 5 个线程，模拟并发访问资源
        for (int i = 0; i < 5; i++) {
            int threadId = i;
            new Thread(() -> {
                String resource = null;
                try {
                    // 获取资源
                    resource = resourcePool.getResource();
                    System.out.println("线程 " + threadId + " 获取了资源: " + resource + ", 可用资源数：" + resourcePool.getAvailableResources());
                    // 模拟资源使用
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 释放资源
                    if (resource != null) {
                        resourcePool.releaseResource(resource);
                        System.out.println("线程 " + threadId + " 释放了资源: " + resource + ", 可用资源数：" + resourcePool.getAvailableResources());
                    }
                }
            }).start();
        }

        // 等待一段时间，让线程执行完毕
        Thread.sleep(5000);
    }
}

在这个例子中，我们创建了一个 ResourcePool，并使用多个线程并发地从资源池中获取和释放资源。你可以运行这段代码，观察线程的执行顺序，体会 Semaphore 在控制资源访问方面的作用。你可以根据自己的需求，将资源池中的资源替换成不同的类型，例如线程、数据库连接、网络连接等。

4. Semaphore 与锁机制的对比

Semaphore 和锁（例如 ReentrantLock）都是用于控制并发访问的工具，但它们的应用场景和实现方式有所不同。

4.1 相同点

• 互斥性: 都可以实现对共享资源的互斥访问，避免多个线程同时修改共享资源，导致数据不一致的问题。
• 阻塞: 当资源不可用时，都可以使线程阻塞，直到资源可用。

4.2 不同点

4.3 性能对比

通常情况下，Semaphore 的性能与 ReentrantLock 相当，甚至在某些情况下，Semaphore 的性能更好。这是因为 Semaphore 的实现相对简单，而且可以根据需要调整许可证的数量，从而更好地控制并发度。然而，Semaphore 的开销也会受到许可证数量的影响，如果许可证数量过多，可能会导致资源浪费。选择使用 Semaphore 还是锁，需要根据具体的应用场景和性能要求进行权衡。

5. Semaphore 的高级用法

5.1 信号量与线程池的结合

Semaphore 可以与线程池结合使用，用于限制任务的并发执行数量。例如，你可以使用 Semaphore 来限制线程池中同时运行的任务数量，防止线程池过载。

import java.util.concurrent.*;

public class SemaphoreThreadPool {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个固定大小的线程池
        int corePoolSize = 5;
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                corePoolSize, corePoolSize, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(10));

        // 创建一个 Semaphore，限制并发任务数量
        Semaphore semaphore = new Semaphore(3);

        // 提交 10 个任务
        for (int i = 0; i < 10; i++) {
            int taskId = i;
            threadPool.execute(() -> {
                try {
                    // 获取许可证
                    semaphore.acquire();
                    System.out.println("任务 " + taskId + " 获取了许可证，开始执行...");
                    // 模拟任务执行时间
                    Thread.sleep(2000);
                    System.out.println("任务 " + taskId + " 执行完毕，释放许可证...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 释放许可证
                    semaphore.release();
                }
            });
        }

        // 关闭线程池
        threadPool.shutdown();
        threadPool.awaitTermination(1, TimeUnit.MINUTES);
    }
}

在这个例子中，我们创建了一个线程池，并使用 Semaphore 限制了同时运行的任务数量为 3。你可以运行这段代码，观察任务的执行顺序，体会 Semaphore 与线程池结合使用的效果。

5.2 信号量与限流的结合

Semaphore 还可以用于实现限流，例如限制某个 API 的访问频率。你可以使用 Semaphore 来限制单位时间内允许通过的请求数量，防止系统过载。

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class RateLimiter {
    private final Semaphore semaphore;
    private final long interval;
    private final int permits;
    private long lastAcquireTime;

    public RateLimiter(int permits, long interval, TimeUnit unit) {
        this.semaphore = new Semaphore(permits);
        this.interval = unit.toMillis(interval);
        this.permits = permits;
        this.lastAcquireTime = System.currentTimeMillis();
    }

    public boolean tryAcquire() {
        long currentTime = System.currentTimeMillis();
        long elapsedTime = currentTime - lastAcquireTime;

        // 如果超过了时间间隔，则重置信号量
        if (elapsedTime >= interval) {
            synchronized (this) {
                currentTime = System.currentTimeMillis(); // 再次获取当前时间，防止并发问题
                elapsedTime = currentTime - lastAcquireTime;
                if (elapsedTime >= interval) {
                    semaphore.release(permits - semaphore.availablePermits()); // 释放所有许可
                    lastAcquireTime = currentTime;
                }
            }
        }

        // 尝试获取许可证
        return semaphore.tryAcquire();
    }

    public static void main(String[] args) throws InterruptedException {
        // 创建一个限流器，每秒允许 2 个请求
        RateLimiter rateLimiter = new RateLimiter(2, 1, TimeUnit.SECONDS);

        // 模拟发送 5 个请求
        for (int i = 0; i < 5; i++) {
            if (rateLimiter.tryAcquire()) {
                System.out.println("请求 " + i + " 通过");
                // 模拟处理请求
                Thread.sleep(500);
            } else {
                System.out.println("请求 " + i + " 被限流");
            }
        }
    }
}

在这个例子中，我们创建了一个 RateLimiter，每秒最多允许 2 个请求通过。tryAcquire() 方法用于尝试获取一个许可证。如果获取成功，则返回 true，否则返回 false。你可以运行这段代码，观察请求的执行情况，体会 Semaphore 在限流方面的作用。

5.3 信号量与读写锁的结合

Semaphore 还可以与读写锁结合使用，实现更灵活的并发控制。例如，你可以使用 Semaphore 来控制读操作的数量，限制同时读取的线程数量。

import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteSemaphore {
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Semaphore readSemaphore;
    private final int maxReaders;
    private int readCount = 0;

    public ReadWriteSemaphore(int maxReaders) {
        this.maxReaders = maxReaders;
        this.readSemaphore = new Semaphore(maxReaders);
    }

    public void read(Runnable task) throws InterruptedException {
        readSemaphore.acquire();
        lock.readLock().lock();
        try {
            readCount++;
            task.run();
        } finally {
            readCount--;
            lock.readLock().unlock();
            readSemaphore.release();
        }
    }

    public void write(Runnable task) {
        lock.writeLock().lock();
        try {
            task.run();
        } finally {
            lock.writeLock().unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReadWriteSemaphore readWriteSemaphore = new ReadWriteSemaphore(3);

        // 模拟读操作
        for (int i = 0; i < 5; i++) {
            int readerId = i;
            new Thread(() -> {
                try {
                    readWriteSemaphore.read(() -> {
                        System.out.println("Reader " + readerId + " is reading.");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("Reader " + readerId + " finished reading.");
                    });
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        // 模拟写操作
        new Thread(() -> {
            readWriteSemaphore.write(() -> {
                System.out.println("Writer is writing.");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Writer finished writing.");
            });
        }).start();

        // 等待一段时间，让线程执行完毕
        Thread.sleep(5000);
    }
}

在这个例子中，我们结合了读写锁和 Semaphore。Semaphore 用于限制并发读操作的数量，ReadWriteLock 用于实现读写互斥。你可以运行这段代码，观察读写操作的执行顺序，体会 Semaphore 在控制读操作并发度方面的作用。

6. 总结

Semaphore 是一个非常强大的并发工具，它提供了灵活的并发控制机制，可以应用于多种场景，例如连接池、资源池、线程池、限流等。通过掌握 Semaphore 的使用方法和应用场景，可以帮助你编写出更高效、更可靠的并发程序。希望这篇文章能帮助你更深入地理解 Semaphore，并能够在实际项目中灵活运用。

记住，在实际应用中，要根据具体的场景和需求，选择合适的并发控制工具。Semaphore、ReentrantLock、ReadWriteLock 等工具各有优缺点，需要根据实际情况进行权衡。多思考，多实践，才能真正掌握并发编程的精髓。

希望这篇文章对你有所帮助！