Java并发25：Semaphore与Exchanger - 控制并发与数据交换

Semaphore semaphore = new Semaphore(3);  // 3个许可证

semaphore.acquire();  // 获取许可证（阻塞）
try {
    // 访问资源
} finally {
    semaphore.release();  // 释放许可证
}

public class ConnectionPool {
    private final Semaphore semaphore;
    private final List<Connection> connections;

    public ConnectionPool(int size) {
        this.semaphore = new Semaphore(size);
        this.connections = new ArrayList<>(size);
        for (int i = 0; i < size; i++) {
            connections.add(createConnection());
        }
    }

    public Connection getConnection() throws InterruptedException {
        semaphore.acquire();  // 获取许可证
        return connections.remove(0);
    }

    public void releaseConnection(Connection conn) {
        connections.add(conn);
        semaphore.release();  // 释放许可证
    }
}

public class RateLimiter {
    private final Semaphore semaphore;

    public RateLimiter(int maxConcurrent) {
        this.semaphore = new Semaphore(maxConcurrent);
    }

    public void execute(Runnable task) {
        try {
            if (semaphore.tryAcquire(100, TimeUnit.MILLISECONDS)) {
                try {
                    task.run();
                } finally {
                    semaphore.release();
                }
            } else {
                throw new RuntimeException("Too many requests");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class ParkingLot {
    private final Semaphore spaces;

    public ParkingLot(int capacity) {
        this.spaces = new Semaphore(capacity);
    }

    public boolean park() {
        return spaces.tryAcquire();  // 尝试停车
    }

    public void leave() {
        spaces.release();  // 离开停车场
    }

    public int availableSpaces() {
        return spaces.availablePermits();
    }
}

Semaphore semaphore = new Semaphore(10);

// 获取许可证
semaphore.acquire();  // 阻塞
semaphore.acquire(3);  // 一次获取多个

// 尝试获取
if (semaphore.tryAcquire()) {  // 非阻塞
    // 获取成功
}

if (semaphore.tryAcquire(1, TimeUnit.SECONDS)) {  // 超时
    // 1秒内获取成功
}

// 释放许可证
semaphore.release();
semaphore.release(3);  // 一次释放多个

// 查询
int available = semaphore.availablePermits();  // 可用许可证数
int waiting = semaphore.getQueueLength();  // 等待线程数

// 非公平（默认，性能好）
Semaphore unfair = new Semaphore(10);

// 公平（FIFO，避免饥饿）
Semaphore fair = new Semaphore(10, true);

Exchanger<String> exchanger = new Exchanger<>();

// 线程1
new Thread(() -> {
    String data = exchanger.exchange("Data from Thread1");
    System.out.println("Thread1 received: " + data);
}).start();

// 线程2
new Thread(() -> {
    String data = exchanger.exchange("Data from Thread2");
    System.out.println("Thread2 received: " + data);
}).start();

// 输出：
// Thread1 received: Data from Thread2
// Thread2 received: Data from Thread1

public class ProducerConsumerExchanger {
    private final Exchanger<Buffer> exchanger = new Exchanger<>();

    class Producer implements Runnable {
        @Override
        public void run() {
            Buffer buffer = new Buffer();
            try {
                while (true) {
                    fillBuffer(buffer);  // 填充数据
                    buffer = exchanger.exchange(buffer);  // 交换空buffer
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    class Consumer implements Runnable {
        @Override
        public void run() {
            Buffer buffer = new Buffer();
            try {
                while (true) {
                    buffer = exchanger.exchange(buffer);  // 交换满buffer
                    processBuffer(buffer);  // 处理数据
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

public class DataValidator {
    private final Exchanger<Data> exchanger = new Exchanger<>();

    public void validate() {
        // 线程1：从数据库读取
        new Thread(() -> {
            Data dbData = readFromDatabase();
            Data fileData = exchanger.exchange(dbData);
            if (!dbData.equals(fileData)) {
                alert("Data mismatch!");
            }
        }).start();

        // 线程2：从文件读取
        new Thread(() -> {
            Data fileData = readFromFile();
            Data dbData = exchanger.exchange(fileData);
            // 数据已交换，可以对比
        }).start();
    }
}

public class GeneticAlgorithm {
    private final Exchanger<Individual> exchanger = new Exchanger<>();

    public void evolve() {
        // 种群1
        new Thread(() -> {
            Individual best = selectBest(population1);
            Individual other = exchanger.exchange(best);  // 交换最优个体
            crossover(best, other);  // 交叉繁殖
        }).start();

        // 种群2
        new Thread(() -> {
            Individual best = selectBest(population2);
            Individual other = exchanger.exchange(best);
            crossover(best, other);
        }).start();
    }
}

Exchanger<String> exchanger = new Exchanger<>();

try {
    String data = exchanger.exchange("myData", 5, TimeUnit.SECONDS);
    System.out.println("Exchanged: " + data);
} catch (TimeoutException e) {
    System.out.println("Exchange timeout");
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

// 基于AQS的共享模式
public class Semaphore {
    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        Sync(int permits) {
            setState(permits);  // 初始许可证数
        }

        // acquire()
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 || compareAndSetState(available, remaining))
                    return remaining;
            }
        }

        // release()
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (compareAndSetState(current, next))
                    return true;
            }
        }
    }
}

// 使用槽位（slot）机制
public class Exchanger<V> {
    private volatile Node[] arena;  // 多槽位（高并发）
    private volatile Node slot;     // 单槽位（低并发）

    public V exchange(V x) throws InterruptedException {
        Object v = slotExchange(x, false, 0);
        if (v == null)
            v = arenaExchange(x, false, 0);  // 槽位冲突，使用arena
        return (V) v;
    }
}