深入浅出Java多线程

http://www.cnblogs.com/LoveJenny/archive/2011/05/24/2053682.html

深入浅出Java多线程（1）-方法 join
Posted on 2008-08-23 23:25 advincenting 阅读(11450) 评论(8)  编辑  收藏 所属分类: JAVA基础知识 
    对于Java开发人员，多线程应该是必须熟练应用的知识点，特别是开发基于Java语言的产品。本文将深入浅出的表述Java多线程的知识点，在后续的系列里将侧重于Java5由Doug Lea教授提供的Concurrent并行包的设计思想以及具体实现与应用。
    如何才能深入浅出呢，我的理解是带着问题，而不是泛泛的看。所以该系列基本以解决问题为主，当然我也非常希望读者能够提出更好的解决问题的方案以及提出更多的问题。由于水平有限，如果有什么错误之处，请大家提出，共同讨论，总之，我希望通过该系列我们能够深入理解Java多线程来解决我们实际开发的问题。
    作为开发人员，我想没有必要讨论多线程的基础知识，比如什么是线程？ 如何创建等 ，这些知识点是可以通过书本和Google获得的。本系列主要是如何理深入解多线程来帮助我们平时的开发，比如线程池如何实现？ 如何应用锁等。 

（1）方法Join是干啥用的？ 简单回答，同步，如何同步？ 怎么实现的？ 下面将逐个回答。
    自从接触Java多线程，一直对Join理解不了。JDK是这样说的：
   join
    public final void join(long millis)throws InterruptedException
    Waits at most millis milliseconds for this thread to die. A timeout of 0 means to wait forever.
大家能理解吗？ 字面意思是等待一段时间直到这个线程死亡，我的疑问是那个线程，是它本身的线程还是调用它的线程的，上代码：
package concurrentstudy;
/**
*
* @author vma
*/
public class JoinTest {
    public static void main(String[] args) {
        Thread t = new Thread(new RunnableImpl());
        t.start();
        try {
            t.join(1000);
            System.out.println("joinFinish");
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
    
        }
    }
}
class RunnableImpl implements Runnable {

    @Override
    public void run() {
        try {
            System.out.println("Begin sleep");
            Thread.sleep(1000);
           System.out.println("End sleep");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}结果是：
Begin sleep
End sleep
joinFinish
明白了吧，当main线程调用t.join时，main线程等待t线程，等待时间是1000，如果t线程Sleep 2000呢
public void run() {
        try {
            System.out.println("Begin sleep");
            // Thread.sleep(1000);
            Thread.sleep(2000);
           System.out.println("End sleep");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
结果是：
Begin sleep
joinFinish
End sleep
也就是说main线程只等1000毫秒，不管T什么时候结束，如果是t.join()呢， 看代码：  
public final void join() throws InterruptedException {
    join(0);
    }
就是说如果是t.join() = t.join(0) 0　JDK这样说的 A timeout of 0 means to wait forever 字面意思是永远等待，是这样吗？
其实是等到t结束后。
这个是怎么实现的吗？ 看JDK代码：
    /**
     * Waits at most <code>millis</code> milliseconds for this thread to
     * die. A timeout of <code>0</code> means to wait forever.
     *
     * @param      millis   the time to wait in milliseconds.
     * @exception  InterruptedException if any thread has interrupted
     *             the current thread.  The <i>interrupted status</i> of the
     *             current thread is cleared when this exception is thrown.
     */
    public final synchronized void join(long millis)
    throws InterruptedException {
    long base = System.currentTimeMillis();
    long now = 0;

    if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
    }

    if (millis == 0) {
        while (isAlive()) {
        wait(0);
        }
    } else {
        while (isAlive()) {
        long delay = millis - now;
        if (delay <= 0) {
            break;
        }
        wait(delay);
        now = System.currentTimeMillis() - base;
        }
    }
    }其实Join方法实现是通过wait（小提示：Object 提供的方法）。 当main线程调用t.join时候，main线程会获得线程对象t的锁（wait 意味着拿到该对象的锁),调用该对象的wait(等待时间)，直到该对象唤醒main线程，比如退出后。

这就意味着main 线程调用t.join时，必须能够拿到线程t对象的锁，如果拿不到它是无法wait的，刚开的例子t.join(1000)不是说明了main线程等待1秒，如果在它等待之前，其他线程获取了t对象的锁，它等待时间可不就是1毫秒了。上代码介绍：
/*
* To change this template, choose Tools | Templates
* and open the template in the editor.
*/
package concurrentstudy;
/**
*
* @author vma
*/
public class JoinTest {
    public static void main(String[] args) {
        Thread t = new Thread(new RunnableImpl());
       new ThreadTest(t).start();
        t.start();
        try {
            t.join();
            System.out.println("joinFinish");
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
    
        }
    }
}
class ThreadTest extends Thread {

    Thread thread;

    public ThreadTest(Thread thread) {
        this.thread = thread;
    }

    @Override
    public void run() {
        holdThreadLock();
    }

    public void holdThreadLock() {
        synchronized (thread) {
            System.out.println("getObjectLock");
            try {
                Thread.sleep(9000);

            } catch (InterruptedException ex) {
             ex.printStackTrace();
            }
            System.out.println("ReleaseObjectLock");
        }

    }
}

class RunnableImpl implements Runnable {

    @Override
    public void run() {
        try {
            System.out.println("Begin sleep");
            Thread.sleep(2000);
           System.out.println("End sleep");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }
}
在main方法中 通过new ThreadTest(t).start();实例化ThreadTest 线程对象， 它在holdThreadLock()方法中，通过 synchronized (thread)，获取线程对象t的锁，并Sleep（9000）后释放，这就意味着，即使
main方法t.join(1000),等待一秒钟，它必须等待ThreadTest 线程释放t锁后才能进入wait方法中，它实际等待时间是9000+1000 MS
运行结果是：
getObjectLock
Begin sleep
End sleep
ReleaseObjectLock
joinFinish

小结：
本节主要深入浅出join及ＪＤＫ中的实现。
在下一节中，我们将要讨论SWing 中的事件方法线程来解决一个网友问到的问题：
如何控制Swing程序在单机只有一个实例，也就是不能运行第二个Main方法。

接深入浅出Java多线程系列（1），本文主要解决的问题是：
如何使其Swing程序只能运行一个实例？
抛开Swing， 我们的程序是通过java 命令行启动一个进程来执行的，该问题也就是说要保证这个进程的唯一性，当然如果能够访问系统的接口，得到进程的信息来判断是否已有进程正在运行，不就解决了吗？但是如何访问系统的接口呢？如何要保证在不同的平台上都是OK的呢？我的思路是用文件锁，当然我相信肯定有更好的方法，呵呵，希望读者能够指出。
文件锁是JDK1.4 NIO提出的，可以在读取一个文件时，获得文件锁，这个锁应该是系统维护的，JVM应该是调用的系统文件锁机制，例子如下：
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
/**
*
* @author vma
*/
public class temp1 {
  public static void main(String args[]) throws FileNotFoundException, InterruptedException, IOException{
    RandomAccessFile r = new RandomAccessFile("d://testData.java","rw");
    FileChannel temp = r.getChannel();
    FileLock fl = temp.lock();
    System.out.println(fl.isValid());
    Thread.sleep(100000);
    temp.close();
  }当代码获得锁后：我们试图编辑这个文件是就会：


如果在启动一个Java Main方法时：
public class temp2 {
  public static void main(String args[]) throws FileNotFoundException, InterruptedException, IOException{
    RandomAccessFile r = new RandomAccessFile("d://testData.java","rw");
    FileChannel temp = r.getChannel();
    FileLock fl = temp.tryLock();
    System.out.println(fl== null);
    temp.close();。返回的结束是 ture , 也就是得不到文件的锁。

这就是对于进程唯一性问题我的解决思路，通过锁定文件使其再启动时得不到锁文件而无法启动。
说到这里，跟今天Swing中的EDT好像还没有关系，对于Swing程序，Main方法中一般像这样：
  public static void main(String[] args) {
    try {
      UIManager.setLookAndFeel(UIManager
          .getCrossPlatformLookAndFeelClassName());
    } catch (Exception e) {
    }

    //Create the top-level container and add contents to it.
    JFrame frame = new JFrame("SwingApplication");
    SwingApplication app = new SwingApplication();
    Component contents = app.createComponents();
    frame.getContentPane().add(contents, BorderLayout.CENTER);

    frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    frame.pack();
    frame.setVisible(true);    启动Jframe后，Main线程就退出了，上面获得文件锁，并持有锁的逻辑往哪里写呢？ 有人会说事件分发线程EDT，真的吗？
    由于我没有做过Swing的项目，仅仅做过个人用的财务管理小软件，还没有深入理解过EDT，不管怎么说先把那段逻辑加到EDT，
    怎么加呢 用SwingUtilities
static void invokeAndWait(Runnable doRun)
          Causes doRun.run() to be executed synchronously on the AWT event dispatching thread.
static void invokeLater(Runnable doRun)
          Causes doRun.run() to be executed asynchronously on the AWT event dispatching thread.
    加上去以后怎么界面没有任何反应了呢？
代码如下：
package desktopapplication1;
import java.awt.BorderLayout;
import java.awt.Component;
import java.awt.GridLayout;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.awt.event.KeyEvent;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.InvocationTargetException;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.swing.BorderFactory;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;
import javax.swing.SwingUtilities;
import javax.swing.UIManager;

public class SwingApplication {
  private static String labelPrefix = "Number of button clicks: ";

  private int numClicks = 0;

  public Component createComponents() {
    final JLabel label = new JLabel(labelPrefix + "0    ");

    JButton button = new JButton("I'm a Swing button!");
    button.setMnemonic(KeyEvent.VK_I);
    button.addActionListener(new ActionListener() {
      public void actionPerformed(ActionEvent e) {
        numClicks++;
        label.setText(labelPrefix + numClicks);
      }
    });
    label.setLabelFor(button);

    /*
     * An easy way to put space between a top-level container and its
     * contents is to put the contents in a JPanel that has an "empty"
     * border.
     */
    JPanel pane = new JPanel();
    pane.setBorder(BorderFactory.createEmptyBorder(30, //top
        30, //left
        10, //bottom
        30) //right
        );
    pane.setLayout(new GridLayout(0, 1));
    pane.add(button);
    pane.add(label);

    return pane;
  }

  public static void main(String[] args) throws InterruptedException {
    try {
      UIManager.setLookAndFeel(UIManager
          .getCrossPlatformLookAndFeelClassName());
    } catch (Exception e) {
    }

    //Create the top-level container and add contents to it.
    JFrame frame = new JFrame("SwingApplication");
    SwingApplication app = new SwingApplication();
    Component contents = app.createComponents();
    frame.getContentPane().add(contents, BorderLayout.CENTER);

    frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    frame.pack();
    frame.setVisible(true);
        try {
            SwingUtilities.invokeAndWait(new getFileLock());
        } catch (InvocationTargetException ex) {
          ex.printStackTrace();
        }
  }
 
}
class getFileLock implements Runnable{

    public void run() {
        try {
            RandomAccessFile r = null;
         try {
                r = new RandomAccessFile("d://testData.java", "rw");
            } catch (FileNotFoundException ex) {
              ex.printStackTrace();
            }
            FileChannel temp = r.getChannel();
            FileLock fl = null;
            try {
                fl = temp.lock();
            } catch (IOException ex) {
                Logger.getLogger(getFileLock.class.getName()).log(Level.SEVERE, null, ex);
            }
   
            System.out.println(fl.isValid());
            try {
                Thread.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException ex) {
               ex.printStackTrace();
            }
            temp.close();
        } catch (IOException ex) {
           ex.printStackTrace();
        }
    }
}
打个断点看看怎么了，断点就在这里     Thread.sleep(Integer.MAX_VALUE); 看看那个线程暂停了 看图片：



看到了吧，我们写的那个getFileLock 是由AWT-EventQueue-0  线程执行，看右下角调用关系， EventDispathThread 启动 Run方法， 然后pumpEvents 取事件，然后从EventQueue取到InvocationEvent 执行Dispath
Dispath调用的就是我们在getFileLock写的run() 方法， ＪＤＫ代码如下：
  public void dispatch() {
    if (catchExceptions) {
        try {
        runnable.run();
        }
        catch (Throwable t) {
                if (t instanceof Exception) {
                    exception = (Exception) t;
                }
                throwable = t;
        }
    }
    else {
        runnable.run();
    }

    if (notifier != null) {
        synchronized (notifier) {
        notifier.notifyAll();
        }
    }
    }  runnable.run();
而如何将我们写的getFileLock加入的那个EventQueue中的呢？当然是SwingUtilities.invokeAndWait(new getFileLock());
看JDK代码：
public static void invokeAndWait(Runnable runnable)
             throws InterruptedException, InvocationTargetException {

        if (EventQueue.isDispatchThread()) {
            throw new Error("Cannot call invokeAndWait from the event dispatcher thread");
        }

    class AWTInvocationLock {}
        Object lock = new AWTInvocationLock();

        InvocationEvent event =
            new InvocationEvent(Toolkit.getDefaultToolkit(), runnable, lock,
                true);

        synchronized (lock) {
            Toolkit.getEventQueue().postEvent(event);
            lock.wait();
        }
Toolkit.getEventQueue().postEvent(event);把我们写的getFileLock 塞进了EventQueue.
这下读者对EDT有个认识了吧。
1. EDT 只有一个线程， 虽然getFileLock是实现Runnable接口，它调用的时候不是star方法启动新线程，而是直接调用run方法。
2. invokeAndWait将你写的getFileLock塞到EventQueue中。
3. Swing 事件机制采用Product Consumer模式 EDT不断的取EventQueue中的事件执行（消费者）。其他线程可以将事件塞入EventQueue中，比如鼠标点击Button是，将注册在BUttion的事件塞入EventQueue中。
所以我们将getFileLock作为事件插入进去后 EDT分发是调用Thread.sleep(Integer.MAX_VALUE)就睡觉了，无暇管塞入EventQueue的其他事件了，比如关闭窗体。

所以绝对不能将持有锁的逻辑塞到EventQueue，而应该放到外边main线程或者其他线程里面。
提到invokeAndWait，还必须说说invokelater 这两个区别在哪里呢?
invokeAndWait与invokelater区别： 看JDK代码：

public static void invokeLater(Runnable runnable) {
        Toolkit.getEventQueue().postEvent(
            new InvocationEvent(Toolkit.getDefaultToolkit(), runnable));
    }
public static void invokeAndWait(Runnable runnable)
             throws InterruptedException, InvocationTargetException {

        if (EventQueue.isDispatchThread()) {
            throw new Error("Cannot call invokeAndWait from the event dispatcher thread");
        }

    class AWTInvocationLock {}
        Object lock = new AWTInvocationLock();

        InvocationEvent event =
            new InvocationEvent(Toolkit.getDefaultToolkit(), runnable, lock,
                true);

        synchronized (lock) {
            Toolkit.getEventQueue().postEvent(event);
            lock.wait();
        }

        Throwable eventThrowable = event.getThrowable();
        if (eventThrowable != null) {
            throw new InvocationTargetException(eventThrowable);
        }
    }invokelater:当在main方法中调用SwingUtils.invokelater,后，把事件塞入EventQueue就返回了，main线程不会阻塞。
invokeAndWait： 当在Main方法中调用SwingUtils.invokeAndWait 后，看代码片段：
        synchronized (lock) {
            Toolkit.getEventQueue().postEvent(event);
            lock.wait();
        }
main线程获得lock 后就wait()了，直到事件分发线程调用lock对象的notify唤醒main线程，否则main 就干等着吧。
这下明白了吧！
总之，对于我们问题最简单的方法就是是main线程里，或者在其他线程里处理。
最后的解决方案是：
package desktopapplication1;
import java.awt.BorderLayout;
import java.awt.Component;
import java.awt.GridLayout;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.awt.event.KeyEvent;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;

import javax.swing.BorderFactory;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;
import javax.swing.UIManager;

public class SwingApplication {
  private static String labelPrefix = "Number of button clicks: ";

  private int numClicks = 0;

  public Component createComponents() {
    final JLabel label = new JLabel(labelPrefix + "0    ");

    JButton button = new JButton("I'm a Swing button!");
    button.setMnemonic(KeyEvent.VK_I);
    button.addActionListener(new ActionListener() {
      public void actionPerformed(ActionEvent e) {
        numClicks++;
        label.setText(labelPrefix + numClicks);
      }
    });
    label.setLabelFor(button);

    /*
     * An easy way to put space between a top-level container and its
     * contents is to put the contents in a JPanel that has an "empty"
     * border.
     */
    JPanel pane = new JPanel();
    pane.setBorder(BorderFactory.createEmptyBorder(30, //top
        30, //left
        10, //bottom
        30) //right
        );
    pane.setLayout(new GridLayout(0, 1));
    pane.add(button);
    pane.add(label);

    return pane;
  }

  public static void main(String[] args) throws InterruptedException {
    try {
      UIManager.setLookAndFeel(UIManager
          .getCrossPlatformLookAndFeelClassName());
    } catch (Exception e) {
    }
    Thread t = new Thread(new getFileLock());
    t.setDaemon(true);
    t.start();
    //Create the top-level container and add contents to it.
    JFrame frame = new JFrame("SwingApplication");
    SwingApplication app = new SwingApplication();
    Component contents = app.createComponents();
    frame.getContentPane().add(contents, BorderLayout.CENTER);

    frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    frame.pack();
    frame.setVisible(true);

  }
 
}
class getFileLock implements Runnable{


    public void run() {
        try {
            RandomAccessFile r = null;
         try {
                r = new RandomAccessFile("d://testData.java", "rw");
            } catch (FileNotFoundException ex) {
              ex.printStackTrace();
            }
            FileChannel temp = r.getChannel();
        
            try {
 
              FileLock fl = temp.tryLock();
              if(fl == null) System.exit(1);
             
            } catch (IOException ex) {
           ex.printStackTrace();
            }
            try {
                Thread.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException ex) {
               ex.printStackTrace();
            }
            temp.close();
        } catch (IOException ex) {
           ex.printStackTrace();
        }
    }
}
在Main方法里启动一个Daemon线程,持有锁，如果拿不到锁，就退出 if(fl == null) System.exit(1);
当然这只是个解决方案，如何友好给给用户提示以及锁定那个文件就要根据具体情况而定了。

　接深入浅出多线程（2）在多线程交互的中，经常有一个线程需要得到另个一线程的计算结果，我们常用的是Future异步模式来加以解决。

　　什么是Future模式呢？Future 顾名思义，在金融行业叫期权，市场上有看跌期权和看涨期权，你可以在现在（比如九月份）购买年底（十二月）的石油，假如你买的是看涨期权，那么如果石油真的涨了，你也可以在十二月份依照九月份商定的价格购买。扯远了，Future就是你可以拿到未来的结果。对于多线程，如果线程A要等待线程B的结果，那么线程A没必要等待B，直到B有结果，可以先拿到一个未来的Future，等B有结果是再取真实的结果。其实这个模式用的很多，比如浏览器下载图片的时候，刚开始是不是通过模糊的图片来代替最后的图片，等下载图片的线程下载完图片后在替换。如图所示：



　　在没有JDK1.5提供的Concurrent之前，我们通过自定义一个结果类，负责结果持有。

　　如下面代码：


双击代码全选 123456789101112131415 packagevincent.blogjava.net; publicclassFutureResult{ 　　privateStringresult; 　　privatebooleanisFinish=false; 　　publicStringgetResult(){ 　　　　returnresult; 　　} 　　publicsynchronizedvoidsetResult(Stringresult){ 　　　　this.result=result; 　　　　this.isFinish=true; 　　} 　　publicsynchronizedbooleanisFinish(){ 　　　　returnisFinish; 　　} }


　　存储结果值和是否完成的Flag。


双击代码全选 123456789101112131415161718 packagevincent.blogjava.net; publicclassGenerateResultThreadextendsThread{ 　　FutureResultfr; 　　publicGenerateResultThread(FutureResultfr){ 　　　　this.fr=fr; 　　} 　　public　voidrun(){ 　　　　//模仿大量耗时计算后（5s）返回结果。 　　　　try{ 　　　　　　System.out.println("GenerateResultThread开始进行计算了！"); 　　　　　　Thread.sleep(5000); 　　　　}catch(InterruptedExceptione){ 　　　　　　//TODOAuto-generatedcatchblock 　　　　　　e.printStackTrace(); 　　　　} 　　　　fr.setResult("ResultByGenerateResultThread"); 　　} }


　　计算具体业务逻辑并放回结果的线程。


双击代码全选 1234567891011121314151617181920212223242526 packagevincent.blogjava.net; publicclassMain{ 　　/** 　　*@paramargs 　　*@throwsInterruptedException 　　*/　　publicstaticvoidmain(String[]args)throwsInterruptedException{ 　　　　//TODOAuto-generatedmethodstub 　　　FutureResultfr=newFutureResult(); 　　　newGenerateResultThread(fr).start(); 　　　//main线程无需等待，不会被阻塞。 　　　//模仿干自己的活2s。 　　　Thread.sleep(2000); 　　　//估计算完了吧取取试试。 　　　System.out.println("过来2s了，看看有结果吗？"); 　　　if(!fr.isFinish()){System.out.println("还没有完成呢！继续干自己活吧！");} 　　　//模仿干自己的活4s。 　　　Thread.sleep(4000); 　　　System.out.println("过来4s了，看看有结果吗？"); 　　　if(fr.isFinish()){ 　　　　　System.out.println("完成了！"); 　　　　　System.out.println("Result："+fr.getResult()); 　　　　　 　　　} 　　} }


　　Main方法需要GenerateResultThread线程计算的结果，通过这种模式，main线程不需要阻塞。结果如下：

　　GenerateResultThread开始进行计算了！

　　过来2s了，看看有结果吗？

　　还没有完成呢！ 继续干自己活吧！

　　过来4s了，看看有结果吗？

　　完成了！

　　Result：ResultByGenerateResultThread

　　在JDK1.5 Concurrent 中，提供了这种Callable的机制。我们只要实现Callable接口中的Call方法，Call方法是可以返回任意类型的结果的。如下：


双击代码全选 123456789101112131415161718192021 packagevincent.blogjava.net; importjava.util.concurrent.Callable; importjava.util.concurrent.ExecutionException; importjava.util.concurrent.Future; importjava.util.concurrent.FutureTask; publicclassConcurrentImpl{ publicstaticvoidmain(String[]args)throwsInterruptedException,Exception{ 　　FutureTask　fr=newFutureTask(newReturnresult()); 　　newThread(fr).start(); 　　　//main线程无需等待，不会被阻塞。 　　　//模仿干自己的活2s。 　　　Thread.sleep(2000); 　　　//估计算完了吧取取试试。 　　　System.out.println("过来2s了，看看有结果吗？"); 　　　if(!fr.isDone()){System.out.println("还没有完成呢！继续干自己活吧！");} 　　　//模仿干自己的活4s。 　　　Thread.sleep(4000); 　　　System.out.println("过来4s了，看看有结果吗？"); 　　　if(fr.isDone()){ 　　　　　System.out.println("完成了！"); 　　


深入浅出多线程（4）对CachedThreadPool OutOfMemoryError问题的一些想法时间:2011-09-07 BlogJava vincent
接系列3，在该系列中我们一起探讨一下CachedThreadPool。

线程池是Conncurrent包提供给我们的一个重要的礼物。使得我们没有必要维 护自个实现的心里很没底的线程池了。但如何充分利用好这些线程池来加快我们 开发与测试效率呢？当然是知己知彼。本系列就说说对CachedThreadPool使用的 一下问题。

下面是对CachedThreadPool的一个测试，程序有问题吗？

package net.blogjava.vincent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class CachedThreadPoolIssue {
/**
* @param args
*/
public static void main(String[] args) {

ExecutorService es = Executors.newCachedThreadPool();
for(int i = 1; i<8000; i++)
es.submit(new task());
}
}
class task implements Runnable{
@Override
public void run() {
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

如果对JVM没有特殊的设置，并在Window平台上，那么就会有一 下异常的发生：

Exception in thread "main" java.lang.OutOfMemoryError: unable to create new native thread
　 at java.lang.Thread.start0(Native Method)
　 at java.lang.Thread.start(Unknown Source)
　 at java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize (Unknown Source)
　 at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
　 at java.util.concurrent.AbstractExecutorService.submit(Unknown Source)
　 at net.blogjava.vincent.CachedThreadPoolIssue.main (CachedThreadPoolIssue.java:19)


本文来自编程入门网：http://www.bianceng.cn/Programming/Java/201109/29119.htm

看看Doc对该线程池的介绍：

Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available. These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks. Calls to execute will reuse previously constructed threads if available. If no existing thread is available, a new thread will be created and added to the pool. Threads that have not been used for sixty seconds are terminated and removed from the cache. Thus, a pool that remains idle for long enough will not consume any resources. Note that pools with similar properties but different details (for example, timeout parameters) may be created using ThreadPoolExecutor constructors.

有以下几点需要注意：

1. 指出会重用先前的线程，不错。

2. 提高了短Task的吞吐量。

3. 线程如果60s没有使用就会移除出Cache。

好像跟刚才的错误没有关系，其实就第一句话说了问题，它会按需要创建新 的线程，上面的例子一下提交8000个Task，意味着该线程池就会创建8000线程， 当然，这远远高于JVM限制了。

注：在JDK1.5中，默认每个线程使用1M内存，8000M !!! 可能吗！！

所以我感觉这应该是我遇到的第一个Concurrent不足之处，既然这么设计， 那么就应该在中Doc指出，应该在使用避免大量Task提交到给 CachedThreadPool.

可能读者不相信，那么下面的例子说明了他创建的Thread。

在ThreadPoolExecutor提供的API中，看到它提供beforeExecute 和 afterExecute两个可以在子类中重载的方法，该方法在线程池中线程执行Task之 前与之后调用。所以我们在beforeExexute中查看目前线程编号就可以确定目前 的线程数目.

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class CachedThreadPoolIssue {
/**
* @param args
*/
public static void main(String[] args) {
ExecutorService es = new LogThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
for(int i = 1; i<8000; i++)
es.submit(new task());
}
}
class task implements Runnable{
@Override
public void run() {
try {
Thread.sleep(600000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
class LogThreadPoolExecutor extends ThreadPoolExecutor{
public LogThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
protected void beforeExecute(Thread t, Runnable r) {
System.out.println(t.getName());
}
protected void afterExecute(Runnable r, Throwable t) {
}
}


本文来自编程入门网：http://www.bianceng.cn/Programming/Java/201109/29119_2.htm

测试结果如图：



当线程数达到5592是，只有在任务管理器Kill该进程了。

如何解决该问题呢，其实在刚才实例化时就看出来了，只需将

new LogThreadPoolExecutor(0, Integer.MAX_VALUE,
　　　　　 　　　60L, TimeUnit.SECONDS,
　　　　　　　　new SynchronousQueue<Runnable>());

Integer.MAX_VALUE 改为合适的大小。对于该参数的含义，涉及到线程池的 实现，将会在下个系列中指出。

当然，其他的解决方案就是控制Task的提交速率，避免超过其最大限制。


本文来自编程入门网：http://www.bianceng.cn/Programming/Java/201109/29119_3.htm
深入浅出多线程系列之五：一些同步构造(上篇)
1：Mutex

Mutex 就像一个C# lock一样，不同的是它可以跨进程.

进入和释放一个Mutex要花费几毫秒，大约比C#的lock慢50倍。

使用一个Mutex的实例，调用WaitOne方法来获取锁，ReleaseMutex方法来释放锁。

因为Mutex是跨进程的，所以我们可以使用Mutex来检测程序是否已经运行。

        public static void MainThread()
        {
            using (var mutex = new Mutex(false, "LoveJenny OneAtATimeDemo"))
            {
                if (!mutex.WaitOne(TimeSpan.FromSeconds(3), false))
                {
                    Console.WriteLine("只能运行一个应用程序!");
                    return;
                }

                RunProgram();
            }
        }
复制代码


2：Semaphore:

一个Semaphore就像一个酒吧一样，通过门卫来限制它的客人，一旦到达限制，没有人可以进入，

人们会在门外乖乖的排队，一旦有一个人离开酒吧，排队中的人就可以进入了一个了。

下面是个例子：

    class TheClub
    {
       //只能容纳三个人的酒吧
        static SemaphoreSlim _sem = new SemaphoreSlim(3);

        public static void MainThread()
        {
            for (int i = 1; i <= 5; i++)
                new Thread(Enter).Start(i); //有5个人向进入
        }
        static void Enter(object id)
        {
            Console.WriteLine(id + " 想要进入了");
            _sem.Wait();
            Console.WriteLine(id+" 已经进入了!");
            Thread.Sleep(1000 * (int)id);
            Console.WriteLine(id + " 离开了?");
            _sem.Release();
        }
    }
复制代码



3：AutoResetEvent

一个AutoResetEvent就像十字转门一样，插入一张票就让一个人通过，”Auto”代表门会自动的关上。

在十字门外面的人可以调用WaitOne方法来阻塞，等待。一旦有人插入了票(调用Set方法)，就可以让外面等待的人(调用WaitOne方法的线程)通过了。

创建AutoResetEvent有一个参数。

static EventWaitHandle _waitHandle = new AutoResetEvent(false);

其中false在msdn的解释是：初始状态为非终止，

按照我个人的理解false代表了十字转门非终止，所以可以正常的进入，等待。

而如果是true的话：初始状态为终止,也就是代表已经调用了Set了，

就是说十字转门已经停止了，所以接下来如果有人调用了WaitOne方法，这个调用WaitOne方法的人直接就可以进入了，不需要再插入票(不需要调用Set)了，之后的调用和false一致，这一点可以认为AutoResetEvent具有记忆功能，它记住了上次门是打开的状态。所以调用waitone方法可以进入。

class ThreadAutoResetEvent
    {
        static EventWaitHandle _waitHandle = new AutoResetEvent(false);

        public static void MainThread()
        {
            new Thread(Waiter).Start();
            Thread.Sleep(2000);
            _waitHandle.Set();
        }

        static void Waiter()
        {
            Console.WriteLine("Waiting...");
            _waitHandle.WaitOne();
            Console.WriteLine("Notified");
        }
}
复制代码
很简单，Waiter执行到Waiting…后，就开始调用WaitOne了，所以在门外排队等待。

而主线程在睡了两秒后，开始插入一张票(Set).所以Waiter就继续执行，所以打印Notified







接下来我们使用AutoResetEvent来模拟实现生产消费问题：



class ProducerConsumerQueue:IDisposable
    {
        EventWaitHandle _wh = new AutoResetEvent(false);
        Thread _worker;
        readonly object _locker = new object();
        Queue<string> _tasks = new Queue<string>();

        public ProducerConsumerQueue()
        {
            //创建并启动工作线程
            _worker = new Thread(Work);
            _worker.Start();
        }

        public void EnqueueTask(string task)
        {
            lock (_locker) _tasks.Enqueue(task);
            _wh.Set(); //一旦有任务了，唤醒等待的线程
        }

        public void Dispose()
        {
            EnqueueTask(null);
            _worker.Join(); //等待_worker线程执行结束
            _wh.Close();
        }

        void Work()
        {
            while (true)
            {
                string task = null;
                lock (_locker)
                {
                    if (_tasks.Count > 0)
                    {
                        task = _tasks.Dequeue();
                        if (task == null)
                            return;
                    }
                    if (task != null) //如果有任务的话，执行任务
                    {
                        Console.WriteLine("Performing task: " + task);
                        Thread.Sleep(1000);
                    }
                    else //否则阻塞,去睡觉吧
                    {
                        _wh.WaitOne();
                    }
                }
            }
        }
    }
复制代码
主线程调用如下：

        public static void Main()
        {
            using (ProducerConsumerQueue q = new ProducerConsumerQueue())
            {
                q.EnqueueTask("Hello");
                for (int i = 0; i < 10; i++) q.EnqueueTask("Say " + i);
                q.EnqueueTask("Goodbye!");
            }
        }
复制代码



4：ManualResetEvent：


一个ManualResetEvent就是一个普通门，

调用Set方法门就打开了，允许任意数量的人进入。

调用WaitOne方法就开始等待进入。

调用Reset方法门就关闭了。

在一个关闭的门上调用WaitOne方法就会被阻塞。

当门下次被打开的时候，所有等待的线程都可以进入了。

除了这些不同外，一个ManualResetEvent和AutoResetEvent类似。

在Framework4.0中ManualResetEvent提供了一个优化版本。ManualResetEventSlim。后面的版本速度更快，并且支持取消(CancellationToken).





参考资料：

http://www.albahari.com/threading/

CLR Via C# 3.0


1：CountdownEvent

Framework 4.0提供了一个CountdownEvent类，主要是让你等待多个线程。考虑下这样的场景：

有一个任务，3个人去做，你需要等这3个人都做完了才继续下一步操作。

下面就是：

class Thread15
    {
        static CountdownEvent _countdown = new CountdownEvent(3);

        public static void MainThread()
        {
            new Thread(SaySomething).Start("I'm thread 1");
            new Thread(SaySomething).Start("I'm thread 2");
            new Thread(SaySomething).Start("I'm thread 3");

            _countdown.Wait(); //阻塞，直到Signal被调用三次
            Console.WriteLine("All threads have finished speaking!");
        }

        static void SaySomething(object thing)
        {
            Thread.Sleep(1000);
            Console.WriteLine(thing);
            _countdown.Signal();
        }
    }
复制代码
注意在构造函数中我们传递了3.代表我们要等待3个线程都结束。



2：ThreadPool.RegisterWaitForSingleObject

如果你的应用程序中有大量的线程都在一个WaitHandle上花费了大量的时间的时候，

你可以通过线程池的ThreadPool.RegisterWaitForSingleObject方法来提高资源的利用率，这个方法接受一个委托，当这个WaitHandle调用signal方法后，

方法的委托就会被执行了。

class Thread16
    {
        static ManualResetEvent _starter = new ManualResetEvent(false);

        public static void MainThread()
        {
            RegisteredWaitHandle reg = ThreadPool.RegisterWaitForSingleObject
                (_starter, Go, "Some Data", -1, true);
　　　　　　//在_starter上等待执行Go方法，-1,代表永远不超时，true代表只执行一遍,”Some Data”是传递的参数

            Thread.Sleep(5000);
            Console.WriteLine("Signaling worker...");
            _starter.Set(); 　　　　//唤醒等待的线程
            Console.ReadLine();
            reg.Unregister(_starter);  //取消注册。
        }

        static void Go(object data, bool timeOut)
        {
            Console.Write("Started - " + data);
            Console.WriteLine("ThreadId:" + Thread.CurrentThread.ManagedThreadId);
        }

    }
复制代码


3：同步上下文：

通过继承ContextBoundObject类，并且加上Synchronization特性，CLR会自动的为你的操作加锁。

//继承自ContextBoundObject，且加上Synchronization特性修饰
    [Synchronization]
    public class AutoLock:ContextBoundObject
    {
        public void Demo()
        {
            Console.WriteLine("Start...");
            Thread.Sleep(1000);
            Console.WriteLine("end");
        }
    }
//主线程：
    public static void MainThread()
        {
            AutoLock safeInstance = new AutoLock();
            new Thread(safeInstance.Demo).Start();
            new Thread(safeInstance.Demo).Start();
            safeInstance.Demo();
        }
复制代码
输出为:

Start…

End

Start…

End

Start…

End



CLR会确保一次只有一个线程可以执行safeInstance里面的代码，它会自动的创建一个同步对象，然后

在每次调用方法或属性的时候都 lock它，从这个角度来讲safeInstance是一个同步上下文。

但是它是怎么样工作的，在Synchronization特性和System.Runtime.Remoting.Contexts命名空间中存在着线索。

一个ContextBoundObject被认为是一个远程(“remote”)对象.意味所有的方法调用都可以被介入。当我们实例化AutoLock的时候，CLR实际上返回了一个proxy对象，一个和AutoLock对象有着同样方法，同样属性的Proxy对象，在这里它扮演者中介的对象。这样就为自动加锁提供了介入的机会，在每一次方法调用上都会花费几微妙的时间来介入。



你可能会认为下面的代码会和上面的一样，是一样的输出结果：

//继承自ContextBoundObject，且加上Synchronization特性修饰
    [Synchronization]
    public class AutoLock:ContextBoundObject
    {
        public void Demo()
        {
            Console.WriteLine("Start...");
            Thread.Sleep(1000);
            Console.WriteLine("end");
        }

        public void Test()
        {
            new Thread(Demo).Start();
            new Thread(Demo).Start();
            new Thread(Demo).Start();
            Console.ReadLine();
        }

        public static void RunTest()
        {
            new AutoLock().Test();
        }
    }

public static void MainThread()
        {
            //AutoLock safeInstance = new AutoLock();
            //new Thread(safeInstance.Demo).Start();
            //new Thread(safeInstance.Demo).Start();
            //safeInstance.Demo();

            AutoLock.RunTest();
        }
复制代码
实际上这里我们会运行到Console.ReadLine方法这里，然后等待输入。

为什么？？

因为CLR会确保一次只有一个线程能够执行AutoLock的代码，所以当主线程执行到Console.ReadLine方法的时候，

就开始等待输入了。如果你按下Enter，代码就和上面的输出一样了。



注：还有一些同步构造将在以后讲到.