线程间的四种通信方式

一、同步

通过Synchronized实现线程间的通信
代码如下：

public class MyObject {

    synchronized public void methodA() {
        //do something....
    }

    synchronized public void methodB() {
        //do some other thing
    }
}

public class ThreadA extends Thread {

    private MyObject object;
//省略构造方法
    @Override
    public void run() {
        super.run();
        object.methodA();
    }
}

public class ThreadB extends Thread {

    private MyObject object;
//省略构造方法
    @Override
    public void run() {
        super.run();
        object.methodB();
    }
}

public class Run {
    public static void main(String[] args) {
        MyObject object = new MyObject();

        //线程A与线程B 持有的是同一个对象:object
        ThreadA a = new ThreadA(object);
        ThreadB b = new ThreadB(object);
        a.start();
        b.start();
    }
}

由于线程A和线程B持有同一个MyObject类的对象object，尽管这两个线程需要调用不同的方法，但是它们是同步执行的，比如：线程B需要等待线程A执行完了methodA()方法之后，它才能执行methodB()方法。这样，线程A和线程B就实现了 通信。

这种方式，本质上就是“共享内存”式的通信。多个线程需要访问同一个共享变量，谁拿到了锁（获得了访问权限），谁就可以执行。

二、while轮询的方式

  import java.util.ArrayList;
  import java.util.List;

  public class MyList {

      private List<String> list = new ArrayList<String>();
      public void add() {
          list.add("elements");
      }
     public int size() {
         return list.size();
     }
 }

 import mylist.MyList;

 public class ThreadA extends Thread {

     private MyList list;

     public ThreadA(MyList list) {
         super();
         this.list = list;
     }

     @Override
     public void run() {
         try {
             for (int i = 0; i < 10; i++) {
                 list.add();
                 System.out.println("添加了" + (i + 1) + "个元素");
                 Thread.sleep(1000);
             }
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }

 import mylist.MyList;

 public class ThreadB extends Thread {

     private MyList list;

     public ThreadB(MyList list) {
         super();
         this.list = list;
     }

     @Override
     public void run() {
         try {
             while (true) {
                 if (list.size() == 5) {
                     System.out.println("==5, 线程b准备退出了");
                     throw new InterruptedException();
                 }
             }
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }

 import mylist.MyList;
 import extthread.ThreadA;
 import extthread.ThreadB;

 public class Test {

     public static void main(String[] args) {
         MyList service = new MyList();

         ThreadA a = new ThreadA(service);
         a.setName("A");
         a.start();

         ThreadB b = new ThreadB(service);
         b.setName("B");
         b.start();
     }
}

线程B通过While循环不断检测(list.size()==5)是否成立，这种方式会浪费处理器资源。因为线程B一直在轮询，而不是当可以执行时，会有通知提醒。
同时，由于线程B每次都取的是本线程的变量，而不是从主内存中取，当线程A的list被更改，线程B就会陷入死循环。

三、notify/wait机制

  import java.util.ArrayList;
  import java.util.List;

  public class MyList {

      private static List<String> list = new ArrayList<String>();

      public static void add() {
          list.add("anyString");
     }

     public static int size() {
         return list.size();
     }
 }


 public class ThreadA extends Thread {

     private Object lock;

     public ThreadA(Object lock) {
         super();
        this.lock = lock;
     }

     @Override
     public void run() {
         try {
             synchronized (lock) {
                 if (MyList.size() != 5) {
                     System.out.println("wait begin "
                             + System.currentTimeMillis());
                     lock.wait();
                     System.out.println("wait end  "
                             + System.currentTimeMillis());
                 }
             }
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }


 public class ThreadB extends Thread {
     private Object lock;

     public ThreadB(Object lock) {
         super();
         this.lock = lock;
     }

     @Override
     public void run() {
         try {
             synchronized (lock) {
                 for (int i = 0; i < 10; i++) {
                     MyList.add();
                     if (MyList.size() == 5) {
                         lock.notify();
                         System.out.println("已经发出了通知");
                     }
                     System.out.println("添加了" + (i + 1) + "个元素!");
                     Thread.sleep(1000);
                 }
             }
        } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }

 public class Run {

     public static void main(String[] args) {

         try {
             Object lock = new Object();

             ThreadA a = new ThreadA(lock);
             a.start();

             Thread.sleep(50);

             ThreadB b = new ThreadB(lock);
             b.start();
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     }
 }

当条件未满足时(list.size() !=5)，线程A调用wait() 放弃CPU，并进入阻塞状态。—不像②while轮询那样占用CPU

当条件满足时，线程B调用 notify()通知 线程A，所谓通知线程A，就是唤醒线程A，并让它进入可运行状态。

这种方式的一个好处就是CPU的利用率提高了。

四、管道通信方式

也就是采用java.io.PipedInputStream 和 java.io.PipedOutputStream进行通信。

public class PipleTest {
    public static void main(String[] args) throws IOException {
        final PipedOutputStream outputStream = new PipedOutputStream();
        final PipedInputStream inputStream = new PipedInputStream(outputStream);
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    outputStream.write("hello world,pipe!".getBytes());
                } catch (IOException e) {
                    e.printStackTrace();
                }finally {
                    try {
                        outputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    int data = inputStream.read();
                    while(data != -1) {
                        System.out.print((char)data);
                        data = inputStream.read();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    try {
                        inputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}

总结:

分布式系统中说的两种通信机制：共享内存机制和消息通信机制。synchronized关键字和while轮询 “属于” 共享内存机制，由于是轮询的条件使用了volatile关键字修饰时，这就表示它们通过判断这个“共享的条件变量“是否改变了，来实现进程间的交流。

参考这篇博客：http://blog.csdn.net/alexlee1986/article/details/21227417