Java并发-定义锁以及消费者-生产者模式实现
一、引子
对于Java并发的锁结构，我们常常使用的是synchonized结构，而由于其灵活度较低，所以在Java-5后提出了Lock接口，以及AbstractQueuedSynchronizer抽象类供我们方便且安全地来实现自定义锁结构，下面从代码出发来开始这篇文章的阅读。
本文就两个实现方式来阐述“生产者-消费者模式”背景下的锁应用，第一种方式是使用Lock接口的自定义实现类来实现，第二种方式是使用synchronized关键字来实现。愿读者在两种不同的实现方式对比中发现各自使用的特点。
二、Lock接口实现
需求：设计一个同步工具：该工具在同一时刻， 只允许至多两个线程同时访问， 超过两个线程的访问将被阻塞， 我们将这个同步工具命名为TwinsLock。并且以生产者和消费者的角度来验证此锁是否成功编写。
package concurrency_basic.chapter16_自定义同步组件;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
public class TwinsLock implements Lock {
private final Sync sync = new Sync(2);
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
if (count <= 0) {
throw new IllegalArgumentException("count must large than zero.");
}
setState(count);
}
public int tryAcquireShared(int reduceCount) {
for (; ; ) {
int current = getState();
int newCount = current - reduceCount;
if (newCount < 0 || compareAndSetState(current,
newCount)) {
return newCount;
}
}
}
public boolean tryReleaseShared(int returnCount) {
for (; ; ) {
int current = getState();
int newCount = current + returnCount;
if (compareAndSetState(current, newCount)) {
return true;
}
}
}
}
public void lock() {
sync.acquireShared(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
public void unlock() {
sync.releaseShared(1);
}
@Override
public Condition newCondition() {
return null;
}
//以下是TwinsLock类是否成功编写的测试代码
public static void main(String[] args) {
final Lock lock = new TwinsLock();
class Worker extends Thread {
public void run() {
lock.lock();
try {
Thread.sleep(2000);
System.out.println(Thread.currentThread().getName());
Thread.sleep(100);
System.out.println();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
// 启动10个线程
for (int i = 0; i < 10; i++) {
Worker w = new Worker();
w.start();
}
}
}控制台输出：
Thread-0
Thread-1
Thread-2
Thread-5
Thread-6
Thread-4
Thread-7
Thread-3
Thread-8
Thread-9三、代码分析
首先，我们由控制台输出可见，我们的确成功地创建了一个最多支持两个线程同时工作的共享锁。
但是如果要求读者朋友不加基础地直接理解以上代码，恐怕对于部分人有所难度。所以我先介绍一下由jdk1.5之后提供的锁设计模式，学了这个之后，理解代码相对容易非常多了。下面我以ReentrantLock类作为一个例子来说明自定义锁的设计模式。
我们先不管右下角的Condition接口。先看看其余接口以及类在锁构造过程中所起到的作用：
Lock接口：锁是面向使用者的，它定义了使用者与锁交互的接口（比如可以允许两个线程并行访问），隐藏了实现细节；
AQS抽象类：同步器面向的是锁的实现者，它简化了锁的实现方式， 隐藏了同步状态管理、 线程的排队、 等待与唤醒等底层操作。
所以上述代码的逻辑是：首先我们写一个静态的内部类Sync，其需要继承AQS抽象类。其主要功能是：重写AQS类内部获取资源方法：tryAcquireShared以及释放资源的方法：tryReleaseShared，资源获取和释放中涉及了同步器状态的变化，而状态的变化需要调用AQS内部提供了CAS方法：compareAndSetState。而AQS中涉及线程排队、休眠、唤醒等操作代码我们并不需要实现，我们所需做的就是关于线程获得到资源/释放资源时，修改同步器的状态，而这个状态将决定线程是否被唤醒，是否将尝试抢夺资源的锁放入等待队列并休眠。
public int tryAcquireShared(int reduceCount) {
for (; ; ) {
int current = getState();
int newCount = current - reduceCount;
if (newCount < 0 || compareAndSetState(current,
newCount)) {
return newCount;
}
}
}
public boolean tryReleaseShared(int returnCount) {
for (; ; ) {
int current = getState();
int newCount = current + returnCount;
if (compareAndSetState(current, newCount)) {
return true;
}
}
} 而Lock方法的实现TwinsLock类的相关方法重写，只需简单地调用AQS抽象类实现：Sync静态内部类对象的若干方法即可：
public void lock() {
sync.acquireShared(1);
}
public void unlock() {
sync.releaseShared(1);
} 其中acquireShared(1)以及releaseShared(1)方法的入口参数值的大小可以认为是对线程资源消耗程度的描述，在这个类中，我们可以认为其都会消耗同步器中资源单位1（总共资源单位为2），如果将值改为2，相当于两个资源会被一个线程锁占据，这样一来，锁就只允许只有一个线程占据资源，进行执行了。
四、传统方式实现生产者消费者模式
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
/**
* @author Fisherman
*/
public class TraditionalLock {
private static final Object MONITOR = new Object();
private AtomicInteger number = new AtomicInteger(2);
public void lock() {
synchronized (MONITOR) {
while (number.get() <= 0) {
try {
MONITOR.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
number.decrementAndGet();
}
}
public void unlock() {
synchronized (MONITOR) {
number.incrementAndGet();
MONITOR.notifyAll();
}
}
public static void main(String[] args) {
final TraditionalLock lock = new TraditionalLock();
class Worker extends Thread {
public void run() {
lock.lock();
try {
Thread.sleep(2000);
System.out.println(Thread.currentThread().getName());
Thread.sleep(100);
System.out.println();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
// 启动10个线程
for (int i = 0; i < 10; i++) {
Worker w = new Worker();
w.start();
}
}
}控制台输出：
Thread-1
Thread-0
Thread-8
Thread-9
Thread-2
Thread-7
Thread-4
Thread-3
Thread-5
Thread-6 可见我们使用传统方式也实现了限制线程运行数目为2的生产者消费者模式，但是与Lock接口实现的锁相比，显然传统的实现方式在lock、unlock方法需要更多的细节。需要设置同步监视器，需要额外调用 wait/notifyAll方法，并且由于使用了synchronized进行上锁，所以在资源消耗上比CAS实现的同步操作更加耗费内存资源。综上所述，使用Lock接口实现的自定义锁更加灵活、耗费更少的资源、对开发者更加友善。