java并发

原理简单。两个线程，一个操作PipedInputStream,一个操作 PipedOutputStream。PipedOutputStream写入的数据先缓存在Buffer中,如果 Buffer满，此线程wait。PipedInputStream读出Buffer中的数据，如果Buffer 没数据，此线程wait。

jdk1.5中的阻塞队列可实现同样功能。

1. 例1 这个例子实际上只是单线程，还谈不上线程间通信，但不妨一看。

   http://hi.baidu.com/ecspell/blog/item/7b02d3133ab555005aaf53f5.html

   package io;
   import java.io.*;
   public class PipedStreamTest {
       public static void main(String[] args) {
           PipedOutputStream ops=new PipedOutputStream();
           PipedInputStream pis=new PipedInputStream();
           try{
               ops.connect(pis);//实现管道连接
               new Producer(ops).run();
               new Consumer(pis).run();
           }catch(Exception e){
               e.printStackTrace();
           }
   
       }
   }
   
   //生产者
   class Producer implements Runnable{
       private PipedOutputStream ops;
       public Producer(PipedOutputStream ops)
       {
           this.ops=ops;
       }
   
       public void run()
       {
           try{
               ops.write("hell,spell".getBytes());
               ops.close();
           }catch(Exception e)
               {e.printStackTrace();}
       }
   }
   
   //消费者
   class Consumer implements Runnable{
       private PipedInputStream pis;
       public Consumer(PipedInputStream pis)
       {
           this.pis=pis;
       }
   
       public void run()
       {
           try{
               byte[] bu=new byte[100];
               int len=pis.read(bu);
               System.out.println(new String(bu,0,len));
               pis.close();
           }catch(Exception e)
               {e.printStackTrace();}
       }
   } 
   

2. 例2 对上面的程序做少许改动就成了两个线程。

   package io;
   import java.io.*;
   public class PipedStreamTest {
       public static void main(String[] args) {
           PipedOutputStream ops=new PipedOutputStream();
           PipedInputStream pis=new PipedInputStream();
           try{
               ops.connect(pis);//实现管道连接
               Producer p = new Producer(ops);
               new Thread(p).start();
               Consumer c = new Consumer(pis);
               new Thread(c).start();
           }catch(Exception e){
               e.printStackTrace();
           }
   
       }
   }
   
   //生产者
   class Producer implements Runnable{
       private PipedOutputStream ops;
       public Producer(PipedOutputStream ops)
       {
           this.ops=ops;
       }
   
       public void run()
       {
           try{
               for(;;){
                   ops.write("hell,spell".getBytes());
                   ops.close();
               }
           }catch(Exception e)
               {e.printStackTrace();}
       }
   }
   
   //消费者
   class Consumer implements Runnable{
       private PipedInputStream pis;
       public Consumer(PipedInputStream pis)
       {
           this.pis=pis;
       }
   
       public void run()
       {
           try{
               for(;;){
                   byte[] bu=new byte[100];
                   int len=pis.read(bu);
                   System.out.println(new String(bu,0,len));
               }
               pis.close();
           }catch(Exception e)
               {e.printStackTrace();}
       }
   }
   

3. 例3. 这个例子更加贴进应用

   import java.io.*;
          
   public class PipedIO { //程序运行后将sendFile文件的内容拷贝到receiverFile文件中
       public static void main(String args[]){       
           try{//构造读写的管道流对象       
               PipedInputStream pis=new PipedInputStream();       
               PipedOutputStream pos=new PipedOutputStream();       
               //实现关联       
               pos.connect(pis);       
               //构造两个线程，并且启动。           
               new Sender(pos,"c:\\text2.txt").start();           
               new Receiver(pis,"c:\\text3.txt").start();         
           }catch(IOException e){       
               System.out.println("Pipe Error"+ e);       
           }       
       }       
   }       
   //线程发送       
   class Sender extends Thread{           
       PipedOutputStream pos;       
       File file;       
       //构造方法       
       Sender(PipedOutputStream pos, String fileName){       
           this.pos=pos;       
           file=new File(fileName);       
       }          
       //线程运行方法       
       public void run(){          
           try{       
               //读文件内容       
               FileInputStream fs=new FileInputStream(file);       
               int data;       
               while((data=fs.read())!=-1){       
                   //写入管道始端       
                   pos.write(data);       
               }       
               pos.close();                        
           }       
           catch(IOException e) {       
               System.out.println("Sender Error" +e);       
           }       
       }       
   }
          
   //线程读       
   class Receiver extends Thread{       
       PipedInputStream pis;       
       File file;       
       //构造方法       
       Receiver(PipedInputStream pis, String fileName){         
           this.pis=pis;       
           file=new File(fileName);       
       }          
       //线程运行       
       public void run(){          
           try {       
               //写文件流对象       
               FileOutputStream fs=new FileOutputStream(file);       
               int data;       
               //从管道末端读       
               while((data=pis.read())!=-1){
          
                   //写入本地文件       
                   fs.write(data);       
               }       
               pis.close();            
           }       
           catch(IOException e){       
               System.out.println("Receiver Error" +e);       
           }       
       }       
   }
   

阻塞队列可以代替管道流方式来实现进水管/排水管模式（生产者/消费者).JDK1.5提供了几个现成的阻塞队列. 现在来看ArrayBlockingQueue的代码如下：

这里是一个阻塞队列

BlockingQueue<Object> blockingQ = new ArrayBlockingQueue<Object> 10;

一个线程从队列里取

for(;;){
    Object o = blockingQ.take();//队列为空，则等待（阻塞）
}

另一个线程往队列存

for(;;){
    blockingQ.put(new Object());//队列满，则等待（阻塞）
}

可见，阻塞队列使用起来比管道简单。

可以使用线程管理任务。还可以使用jdk1.5提供的一组类来更方便的管理任务。从这些类里我们可以体会一种面向任务的思维方式。这些类是：

1. Executor接口。使用方法：

   Executor executor = anExecutor;//生成一个Executor实例。
   executor.execute(new RunnableTask1());
   

   用意：使用者只关注任务执行，不用操心去关注任务的创建、以及执行细节等这些第三方实现者关心的问题。也就是说，把任务的调用执行和任务的实现解耦。

   实际上，JDK1.5中已经有该接口出色的实现。够用了。

2. Executors是一个如同Collections一样的工厂类或工具类，用来产生各种不同接口的实例。
3. ExecutorService接口它继承自Executor. Executor只管把任务扔进 executor()里去执行，剩余的事就不管了。而ExecutorService则不同，它会多做点控制工作。比如：

   class NetworkService {
       private final ServerSocket serverSocket;
       private final ExecutorService pool;
   
       public NetworkService(int port, int poolSize) throws IOException {
           serverSocket = new ServerSocket(port);
           pool = Executors.newFixedThreadPool(poolSize);
       }
    
       public void serve() {
           try {
               for (;;) {
                   pool.execute(new Handler(serverSocket.accept()));
               }
           } catch (IOException ex) {
               pool.shutdown(); //不再执行新任务
           }
       }
   }
   
   class Handler implements Runnable {
       private final Socket socket;
       Handler(Socket socket) { this.socket = socket; }
       public void run() {
           // read and service request
       }
   }
   

   ExecutorService(也就是代码里的pool对象）执行shutdown后，它就不能再执行新任务了，但老任务会继续执行完毕，那些等待执行的任务也不再等待了。

4. 任务提交者与执行者通讯

   public static void main(String args[])throws Exception {
       ExecutorService executor = Executors.newSingleThreadExecutor();
       Callable<String> task = new Callable<String>(){
           public String call()throws Exception{
               return "test";
           }
       };
       Future<String> f = executor.submit(task); 
       String result = f.get();//等待（阻塞）返回结果
       System.out.println(result);
       executor.shutdown();                
   }
   

   Executors.newSingleThreadExecutor()取得的Executor实例有以下特性:

   1. 任务顺序执行. 比如：

      executor.submit(task1);
      executor.submit(task2);
      

      必须等task1执行完，task2才能执行。

   2. task1和task2会被放入一个队列里，由一个工作线程来处理。即：一共有2个线程(主线程、处理任务的工作线程）。
5. 其它的类请参考Java Doc

本节例子来自温少的Java并发教程，可能会有改动。向温少致敬。

1. CountDownLatch 门插销计数器
   1. 启动线程，然后等待线程结束。即常用的主线程等所有子线程结束后再执行的问题。

      public static void main(String[] args)throws Exception {
          // TODO Auto-generated method stub
          final int count=10;
          final CountDownLatch completeLatch = new CountDownLatch(count);//定义了门插销的数目是10
                      
          for(int i=0;i<count;i++){
              Thread thread = new Thread("worker thread"+i){
                      public void run(){
                          //do xxxx                                   
                          completeLatch.countDown();//减少一根门插销
                      }
                  };
              thread.start();
          }           
          completeLatch.await();//如果门插销还没减完则等待。
      } 
      

      JDK1.4时，常用办法是给子线程设置状态，主线程循环检测。易用性和效率都不好。

   2. 启动很多线程，等待通知才能开始

      public static void main(String[] args) throws Exception {
          // TODO Auto-generated method stub
          final CountDownLatch startLatch = new CountDownLatch(1);//定义了一根门插销
      
          for (int i = 0; i < 10; i++) {
              Thread thread = new Thread("worker thread" + i) {
                      public void run() {
                          try {
                              startLatch.await();//如果门插销还没减完则等待
                          } catch (InterruptedException e) {
      
                          }
                          // do xxxx
                      }
                  };
              thread.start();
          }
          startLatch.countDown();//减少一根门插销
      }
      

2. CycliBarrier. 等所有线程都达到一个起跑线后才能开始继续运行。

   public class CycliBarrierTest implements Runnable {
       private CyclicBarrier barrier;
   
       public CycliBarrierTest(CyclicBarrier barrier) {
           this.barrier = barrier;
       }
   
       public void run() {
           //do xxxx;
           try {
               this.barrier.await();//线程运行至此会检查是否其它线程都到齐了，没到齐就继续等待。到齐了就执行barrier的run函数体里的内容
           } catch (Exception e) {
   
           }
       }
   
       /**
        * @param args
        */
       public static void main(String[] args) {
           //参数2代表两个线程都达到起跑线才开始一起继续往下执行
           CyclicBarrier barrier = new CyclicBarrier(2, new Runnable() {
                   public void run() {
                       //do xxxx;
                   }
               });
           Thread t1 = new Thread(new CycliBarrierTest(barrier));         
           Thread t2 = new Thread(new CycliBarrierTest(barrier));
           t1.start();
           t2.start();
       }
   
   }
   

   这简化了传统的用计数器+wait/notifyAll来实现该功能的方式。

1. Amdahl定律. 给定问题规模，可并行化部分占12%，那么即使把并行运用到极致，系统的性能最多也只能提高1/(1-0.12)=1.136倍。即：并行对提高系统性能有上限。
2. Gustafson定律. Gustafson定律说Amdahl定律没有考虑随着cpu的增多而有更多的计算能力可被使用。其本质在于更改问题规模从而可以把Amdahl定律中那剩下的88%的串行处理并行化，从而可以突破性能门槛。本质上是一种空间换时间。
3. Sun-Ni定律. 是前两个定律的进一步推广。其主要思想是计算的速度受限于存储而不是CPU的速度. 所以要充分利用存储空间等计算资源，尽量增大问题规模以产生更好/更精确的解.

计算机识别物体需要飞速的计算，以至于芯片发热发烫，而人在识别物体时却一目了然，却并不会导致某个脑细胞被烧热烧焦（夸张）而感到不适，是由于大脑是一个分布式并行运行系统，就像google用一些廉价的linux服务器可以进行庞大复杂的计算一样，大脑内部无数的神经元的独自计算，互相分享成果，从而瞬间完成需要单个cpu万亿次运算才能有的效果。试想，如果在并行处理领域有所创建，将对计算机的发展和未来产生不可估量的影响。当然，其中的挑战也可想而知：许多的问题是并不容易轻易就“分割”的了的。