Java 多线程与并发 - ReentrantLock

public class LockTest {
    private final Lock lock = new ReentrantLock();
    // 容器未满
    private final Condition notFull = lock.newCondition();
    // 容器非空
    private final Condition notEmpty = lock.newCondition();
    // 容器
    private final Object[] items = new Object[20];

    int putIndex;
    int takeIndex;
    int count;

    public void put(Object o) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length) {
                // 容器满了，阻塞生产者
                notFull.await();
            }
            if (++putIndex == items.length) {
                putIndex = 0;
            }
            ++count;
            // 放入产品，唤醒消费者
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                // 容器为空，阻塞生产者消费者
                notEmpty.await();
            }
            final Object item = items[takeIndex];
            if (++takeIndex == items.length) {
                takeIndex = 0;
            }
            --count;
            // 取走产品，唤醒生产者
            notFull.signal();
            return item;
        } finally {
            lock.unlock();
        }
    }

}

@Test
public void test() throws InterruptedException {
    final LockTest lockTest = new LockTest();
    final ExecutorService producer = Executors.newFixedThreadPool(5);
    final ExecutorService consumer = Executors.newFixedThreadPool(5);
    for (int i = 0; i < 25; i++) {
        String name = "put-" + i;
        producer.submit(() -> {
            try {
                System.out.println("生产：" + name);
                lockTest.put(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    Thread.sleep(10);
    System.out.println("容器中有 20 个产品，生产者线程 5 个等待中");
    for (int i = 0; i < 25; i++) {
        consumer.submit(() -> {
            try {
                System.out.printf("消费：%s\n", lockTest.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
    Thread.sleep(10);
    System.out.println("容器中有 0 个产品，消费者线程 5 个等待中");
}

public void method() {
    lock.lock();
    try {
        lock.lock();
    } finally {
        lock.unlock();
    }
}

@Test
public void test1() {
    final MyLock myLock = new MyLock();
    lock.lock();
    try {
        lock.lock();
    } finally {
        lock.unlock();
    }
}

private static class MyLock extends AbstractQueuedSynchronizer {
    public void lock() {
        acquire(1);
    }

    public void unlock() {
        release(1);
    }

    @Override
    protected boolean tryAcquire(int arg) {
        return compareAndSetState(0, 1);
    }

    @Override
    protected boolean tryRelease(int arg) {
        setState(0);
        return true;
    }
}

@Test
public void test2() {
    // true 公平锁，false 非公平锁
    final ReentrantLock lock = new ReentrantLock(true);
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    for (int i = 0; i < 3; i++) {
        executorService.submit(() -> {
            for (int k = 0; k < 2; k++) {
                lock.lock();
                System.out.printf("线程%s 获得锁\n", Thread.currentThread().getName());
                lock.unlock();
            }
        });
    }
}

/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            // CAS 修改成功，设置当前线程获得锁
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 否则调用 AQS 中的 acquire() 方法， acquire() 又调用子类的 tryAcquire() 方法
            // 如果 tryAcquire() 方法仍不能获取锁，则进入阻塞队列等待唤醒
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // state == 0，没有线程获取到锁
        if (compareAndSetState(0, acquires)) {
            // CAS 修改成功，设置当前线程获得锁
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    // 有线程获取到锁，判断是不是当前线程获取到的
    else if (current == getExclusiveOwnerThread()) {
        // 是当前线程获取到的，state + 1（acquires 传进来就是 1）
        int nextc = c + acquires;
        // 同一个线程可重入最大次数为 Integer.MAX_VALUE，超出后再加一即为负数
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    // 获取锁失败，会进入 AQS 的等待队列中
    return false;
}

/**
 * Sync object for fair locks
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        // 直接走 AQS 的 acquire() 流程，会调用 tryAcquire() 方法
        acquire(1);
    }

    /**
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // state == 0，没有线程获取到锁
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                // 阻塞等待队列中没有线程，并且 CAS 修改状态成功，设置当前线程获得锁
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 有线程获取到锁，判断是不是当前线程获取到的
        else if (current == getExclusiveOwnerThread()) {
            // 是当前线程获取到的，重入
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        // 阻塞等待队列中有线程，锁也不是当前线程获取，则进入阻塞等待队列
        return false;
    }
}

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}