(转)Java并发编程总结Hadoop核心源码实例解读

Java并发编程总结---Hadoop核心源码实例解读

(2011-05-26 10:42:44)

转载▼

标签： 并发编程

分类： 编程

程序设计需要同步(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.
 
(2)guarantee that one thread’s changes will be visible to another
 

2）配合使用object的wait和notify，实现对象monitor控制权从一个线程调度到另外一个线程。具体实例请参阅http://blog.sina.com.cn/s/blog_4a1f59bf0100rgxh.html
3）使用ReentrantLock和Condition控制。ReentrantLock和Condition出现在java.util.concurrent包中。下面有从Hadoop源码中摘取出来的一部分的内容作为介绍。

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;
        }
      }
    }