Java并发编程总结---Hadoop核心源码实例解读
程序设计需要同步(synchronization),原因:
1)复杂的功能要求的需要使用多线程编程,线程之间存在读写共享变量。
2)读写共享变量(shared mutual variable),JVM的内存模型(Memory model: decide when and how changes made by one thread become visuble to others)受到其它因素干扰。
3)对共享变量的操作非原子性。例如 i++;就不是原子操作,它分为两部分,(1) 读i (2) i+1写入内存。如果i是线程A和线程B共享的变量,线程A在操作(1)之后,线程调度器调度调度线程B执行i++,因此两个线程在变量i产生了不一致。注意,volatile修饰符是线程操作之前更新数据,但是,上面的问题显然不是更新数据就能解决的。
4)增加互斥区(mutual exclusion)会降低执行效率,但是这是实现数据安全、功能强大的多线程编程最为重要的部分。
5)线程之间需要配合的场景需要并发控制逻辑。
Java并发编程使用的方法:
1) 为代码块和函数添加synchronized,同步的作用有两点:
(1)a means of mutual exclusion, to prevent an object from being observed in an inconsistent state while it’s being modified by another thread.
Hadoop源码使用并发控制的实例:
Map阶段产生<K,V>会先存储在内存中,等到io.sort.mb指定的内存达到阈值(percent)时,会启动spill到本地磁盘的工作。
ReentrantLock与Condition的配合使用,Condition为ReentrantLock锁的等待和释放提供控制逻辑。
例如,使用ReentrantLock加锁之后,可以通过它自身的Condition.await()方法释放该锁,线程在此等待Condition.signal()方法,然后继续执行下去。await方法需要放在while循环中,因此,在不同线程之间实现并发控制,还需要一个volatile的变量,boolean是原子性的变量。因此,一般的并发控制的操作逻辑如下所示:
volatile boolean isProcess = false;
ReentrantLock lock = new ReentrantLock();
Condtion processReady = lock.newCondtion();
thread: run() {
lock.lock();
isProcess = true;
try {
while(!isProcessReady) { //isProcessReady 是另外一个线程的控制变量
processReady.await();//释放了lock,在此等待signal
}catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
isProcess = false;
}
}
}
}
看Hadoop的一段摘取的源码:
private class MapOutputBuffer<K extends Object, V extends Object>
implements MapOutputCollector<K, V>, IndexedSortable {
...
boolean spillInProgress;
final ReentrantLock spillLock = new ReentrantLock();
final Condition spillDone = spillLock.newCondition();
final Condition spillReady = spillLock.newCondition();
volatile boolean spillThreadRunning = false;
final SpillThread spillThread = new SpillThread();
...
public MapOutputBuffer(TaskUmbilicalProtocol umbilical, JobConf job,
TaskReporter reporter
) throws IOException, ClassNotFoundException {
...
spillInProgress = false;
spillThread.setDaemon(true);
spillThread.setName("SpillThread");
spillLock.lock();
try {
spillThread.start();
while (!spillThreadRunning) {
spillDone.await();
}
} catch (InterruptedException e) {
throw new IOException("Spill thread failed to initialize", e);
} finally {
spillLock.unlock();
}
}
protected class SpillThread extends Thread {
@Override
public void run() {
spillLock.lock();
spillThreadRunning = true;
try {
while (true) {
spillDone.signal();
while (!spillInProgress) {
spillReady.await();
}
try {
spillLock.unlock();
sortAndSpill();
} catch (Throwable t) {
sortSpillException = t;
} finally {
spillLock.lock();
if (bufend < bufstart) {
bufvoid = kvbuffer.length;
}
kvstart = kvend;
bufstart = bufend;
spillInProgress = false;
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
spillLock.unlock();
spillThreadRunning = false;
}
}
}