Java多线程

 

一.进程和线程

进程：每个进程都有独立的代码和数据空间（进程上下文），进程间的切换会有较大的开销，一个进程包含1--n个线程。（进程是资源分配的最小单位）

 

线程：同一类线程共享代码和数据空间，每个线程有独立的运行栈和程序计数器(PC)，线程切换开销小。（线程是cpu调度的最小单位）

 

线程和进程一样分为五个阶段：创建、就绪、运行、阻塞、终止。

 

多进程是指操作系统能同时运行多个任务（程序）。

 

多线程是指在同一程序中有多个顺序流在执行

 

二.线程的创建

Java中线程的创建常见有如三种基本形式

1.继承Thread类，重写该类的run()方法

 

 

class MyThread extends Thread {
     
     private int i = 0;
 
     @Override
     public void run() {
         for (i = 0; i < 100; i++) {
             System.out.println(Thread.currentThread().getName() + " " + i);
         }
     }
 }

Thread myThread1 = new MyThread();

 myThread1.start();

 

2.实现Runnable接口，并重写该接口的run()方法。

该run()方法同样是线程执行体，创建Runnable实现类的实例，并以此实例作为Thread类的target来创建Thread对象，该Thread对象才是真正的线程对象。

 1 class MyRunnable implements Runnable {
 2     private int i = 0;
 3 
 4     @Override
 5     public void run() {
 6         for (i = 0; i < 100; i++) {
 7             System.out.println(Thread.currentThread().getName() + " " + i);
 8         }
 9     }
10 }

new Thread(new MyRunnable()).start();

 

3.使用Callable和Future接口创建线程。

具体是创建Callable接口的实现类，并实现clall()方法。并使用FutureTask类来包装Callable实现类的对象，且以此FutureTask对象作为Thread对象的target来创建线程。

 

 1 public class ThreadTest {
 2 
 3     public static void main(String[] args) {
 4 
 5         Callable<Integer> myCallable = new MyCallable();    // 创建MyCallable对象
 6         FutureTask<Integer> ft = new FutureTask<Integer>(myCallable); //使用FutureTask来包装MyCallable对象
 7 
 8         for (int i = 0; i < 100; i++) {
 9             System.out.println(Thread.currentThread().getName() + " " + i);
10             if (i == 30) {
11                 Thread thread = new Thread(ft);   //FutureTask对象作为Thread对象的target创建新的线程
12                 thread.start();                      //线程进入到就绪状态
13             }
14         }
15 
16         System.out.println("主线程for循环执行完毕..");
17         
18         try {
19             int sum = ft.get();            //取得新创建的新线程中的call()方法返回的结果
20             System.out.println("sum = " + sum);
21         } catch (InterruptedException e) {
22             e.printStackTrace();
23         } catch (ExecutionException e) {
24             e.printStackTrace();
25         }
26 
27     }
28 }
29 
30 
31 class MyCallable implements Callable<Integer> {
32     private int i = 0;
33 
34     // 与run()方法不同的是，call()方法具有返回值
35     @Override
36     public Integer call() {
37         int sum = 0;
38         for (; i < 100; i++) {
39             System.out.println(Thread.currentThread().getName() + " " + i);
40             sum += i;
41         }
42         return sum;
43     }
44 
45 }

 

注意：对于线程的启动而言，都是调用线程对象的start()方法，但是不能对同一线程对象两次调用start()方法。

 

三.线程的生命周期和状态转换

 

 

 

Java线程具有五中基本状态

新建状态（New）：当线程对象对创建后，即进入了新建状态，如：Thread t = new MyThread();

就绪状态（Runnable）：当调用线程对象的start()方法（t.start();），线程即进入就绪状态。处于就绪状态的线程，只是说明此线程已经做好了准备，随时等待CPU调度执行，并不是说执行了t.start()此线程立即就会执行；

运行状态（Running）：当CPU开始调度处于就绪状态的线程时，此时线程才得以真正执行，即进入到运行状态。注：就     绪状态是进入到运行状态的唯一入口，也就是说，线程要想进入运行状态执行，首先必须处于就绪状态中；

阻塞状态（Blocked）：处于运行状态中的线程由于某种原因，暂时放弃对CPU的使用权，停止执行，此时进入阻塞状态，直到其进入到就绪状态，才 有机会再次被CPU调用以进入到运行状态。根据阻塞产生的原因不同，阻塞状态又可以分为三种：

1.等待阻塞：运行状态中的线程执行wait()方法，使本线程进入到等待阻塞状态；

2.同步阻塞 -- 线程在获取synchronized同步锁失败(因为锁被其它线程所占用)，它会进入同步阻塞状态；

3.其他阻塞 -- 通过调用线程的sleep()或join()或发出了I/O请求时，线程会进入到阻塞状态。当sleep()状态超时、join()等待线程终止或者超时、或者I/O处理完毕时，线程重新转入就绪状态。

 

死亡状态（Dead）：线程执行完了或者因异常退出了run()方法，该线程结束生命周期。

 

四.线程的调度

1、调整线程优先级：Java线程有优先级，优先级高的线程会获得较多的运行机会。

 

Java线程的优先级用整数表示，取值范围是1~10，Thread类有以下三个静态常量：

static int MAX_PRIORITY

          线程可以具有的最高优先级，取值为10。

static int MIN_PRIORITY

          线程可以具有的最低优先级，取值为1。

static int NORM_PRIORITY

          分配给线程的默认优先级，取值为5。

 

Thread类的setPriority()和getPriority()方法分别用来设置和获取线程的优先级。

 

每个线程都有默认的优先级。主线程的默认优先级为Thread.NORM_PRIORITY。

线程的优先级有继承关系，比如A线程中创建了B线程，那么B将和A具有相同的优先级。

JVM提供了10个线程优先级，但与常见的操作系统都不能很好的映射。如果希望程序能移植到各个操作系统中，应该仅仅使用Thread类有以下三个静态常量作为优先级，这样能保证同样的优先级采用了同样的调度方式。

 

2、线程睡眠：Thread.sleep(long millis)方法，使线程转到阻塞状态。millis参数设定睡眠的时间，以毫秒为单位。当睡眠结束后，就转为就绪（Runnable）状态。sleep()平台移植性好。

 

3、线程等待：Object类中的wait()方法，导致当前的线程等待，直到其他线程调用此对象的 notify() 方法或 notifyAll() 唤醒方法。这个两个唤醒方法也是Object类中的方法，行为等价于调用 wait(0) 一样。

 

4、线程让步：Thread.yield() 方法，暂停当前正在执行的线程对象，把执行机会让给相同或者更高优先级的线程。

 

5、线程加入：join()方法，等待其他线程终止。在当前线程中调用另一个线程的join()方法，则当前线程转入阻塞状态，直到另一个进程运行结束，当前线程再由阻塞转为就绪状态。

 

6、线程唤醒：Object类中的notify()方法，唤醒在此对象监视器上等待的单个线程。如果所有线程都在此对象上等待，则会选择唤醒其中一个线程。选择是任意性的，并在对实现做出决定时发生。线程通过调用其中一个 wait 方法，在对象的监视器上等待。 直到当前的线程放弃此对象上的锁定，才能继续执行被唤醒的线程。被唤醒的线程将以常规方式与在该对象上主动同步的其他所有线程进行竞争；例如，唤醒的线程在作为锁定此对象的下一个线程方面没有可靠的特权或劣势。类似的方法还有一个notifyAll()，唤醒在此对象监视器上等待的所有线程。

 

五.线程同步

1、同步方法

　　即有synchronized关键字修饰的方法。 由于java的每个对象都有一个内置锁，当用此关键字修饰方法时， 内置锁会保护整个方法。在调用该方法前，需要获得内置锁，否则就处于阻塞状态。

注： synchronized关键字也可以修饰静态方法，此时如果调用该静态方法，将会锁住整个类。

 

2、同步代码块

        即有synchronized关键字修饰的语句块。 被该关键字修饰的语句块会自动被加上内置锁，从而实现同步

        代码如： synchronized(object){ }

注：同步是一种高开销的操作，因此应该尽量减少同步的内容。通常没有必要同步整个方法，使用synchronized代码块同步关键代码即可。

 

同步方法:给一个方法增加synchronized修饰符之后就可以使它成为同步方法，这个方法可以是静态方法和非静态方法，但是不能是抽象类的抽象方法，也不能是接口中的接口方法。

同步块：同步块是通过锁定一个指定的对象，来对同步块中包含的代码进行同步；

 

3、wait与notify

wait():使一个线程处于等待状态，并且释放所持有的对象的lock。

 

sleep():使一个正在运行的线程处于睡眠状态，是一个静态方法，调用此方法要捕捉InterruptedException异常。

notify():唤醒一个处于等待状态的线程，注意的是在调用此方法的时候，并不能确切的唤醒某一个等待状态的线程，而是由JVM确定唤醒哪个线程，而且不是按优先级。

Allnotity():唤醒所有处入等待状态的线程，注意并不是给所有唤醒线程一个对象的锁，而是让它们竞争。

 

4、使用特殊域变量(volatile)实现线程同步

    a.volatile关键字为域变量的访问提供了一种免锁机制

    b.使用volatile修饰域相当于告诉虚拟机该域可能会被其他线程更新

    c.因此每次使用该域就要重新计算，而不是使用寄存器中的值 

    d.volatile不会提供任何原子操作，它也不能用来修饰final类型的变量 

 

5、使用重入锁实现线程同步

    ReentrantLock类是可重入、互斥、实现了Lock接口的锁，它与使用synchronized方法和快具有相同的基本行为和语义，并且扩展了其能力。

 ReenreantLock类的常用方法有：

ReentrantLock() : 创建一个ReentrantLock实例 

lock() : 获得锁 

unlock() : 释放锁 

注：ReentrantLock()还有一个可以创建公平锁的构造方法，但由于能大幅度降低程序运行效率，不推荐使用 

 

注：关于Lock对象和synchronized关键字的选择： 

        a.最好两个都不用，使用一种java.util.concurrent包提供的机制，能够帮助用户处理所有与锁相关的代码。 

        b.如果synchronized关键字能满足用户的需求，就用synchronized，因为它能简化代码 

        c.如果需要更高级的功能，就用ReentrantLock类，此时要注意及时释放锁，否则会出现死锁，通常在finally代码释放锁 

 

6、使用局部变量实现线程同步

    如果使用ThreadLocal管理变量，则每一个使用该变量的线程都获得该变量的副本，副本之间相互独立，这样每一个线程都可以随意修改自己的变量副本，而不会对其他线程产生影响。

     ThreadLocal 类的常用方法

ThreadLocal() : 创建一个线程本地变量 

get() : 返回此线程局部变量的当前线程副本中的值 

initialValue() : 返回此线程局部变量的当前线程的"初始值" 

set(T value) : 将此线程局部变量的当前线程副本中的值设置为value

 

    注：ThreadLocal与同步机制 

        a.ThreadLocal与同步机制都是为了解决多线程中相同变量的访问冲突问题。 

        b.前者采用以"空间换时间"的方法，后者采用以"时间换空间"的方式

 

7、使用阻塞队列实现线程同步

前面5种同步方式都是在底层实现的线程同步，但是我们在实际开发当中，应当尽量远离底层结构。 使用javaSE5.0版本中新增的java.util.concurrent包将有助于简化开发。 本小节主要是使用LinkedBlockingQueue<E>来实现线程的同步 LinkedBlockingQueue<E>是一个基于已连接节点的，范围任意的blocking queue。 队列是先进先出的顺序（FIFO），关于队列以后会详细讲解~LinkedBlockingQueue 类常用方法 LinkedBlockingQueue() : 创建一个容量为Integer.MAX_VALUE的LinkedBlockingQueue put(E e) : 在队尾添加一个元素，如果队列满则阻塞 size() : 返回队列中的元素个数 take() : 移除并返回队头元素，如果队列空则阻塞代码实例： 实现商家生产商品和买卖商品的同步

 

注：BlockingQueue<E>定义了阻塞队列的常用方法，尤其是三种添加元素的方法，我们要多加注意，当队列满时：

　　add()方法会抛出异常

　　offer()方法返回false

　　put()方法会阻塞

 

8、使用原子变量实现线程同步

需要使用线程同步的根本原因在于对普通变量的操作不是原子的。

 

那么什么是原子操作呢？原子操作就是指将读取变量值、修改变量值、保存变量值看成一个整体来操作即-这几种行为要么同时完成，要么都不完成。在java的util.concurrent.atomic包中提供了创建了原子类型变量的工具类，使用该类可以简化线程同步。其中AtomicInteger 表可以用原子方式更新int的值，可用在应用程序中(如以原子方式增加的计数器)，但不能用于替换Integer；可扩展Number，允许那些处理机遇数字类的工具和实用工具进行统一访问。

 

AtomicInteger类常用方法：

AtomicInteger(int initialValue) : 创建具有给定初始值的新的

AtomicIntegeraddAddGet(int dalta) : 以原子方式将给定值与当前值相加

get() : 获取当前值

 

补充--原子操作主要有：　　

对于引用变量和大多数原始变量(long和double除外)的读写操作；　　

对于所有使用volatile修饰的变量(包括long和double)的读写操作。

 

六.线程数据传递

1、通过构造方法传递数据 
在创建线程时，必须要建立一个Thread类的或其子类的实例。因此，我们不难想到在调用start方法之前通过线程类的构造方法将数据传入线程。并将传入的数据使用类变量保存起来，以便线程使用(其实就是在run方法中使用)。下面的代码演示了如何通过构造方法来传递数据： 

代码如下:

package mythread; 
public class MyThread1 extends Thread 
{ 
private String name; 
public MyThread1(String name) 
{ 
this.name = name; 
} 
public void run() 
{ 
System.out.println("hello " + name); 
} 
public static void main(String[] args) 
{ 
Thread thread = new MyThread1("world"); 
thread.start(); 
} 
} 

由于这种方法是在创建线程对象的同时传递数据的，因此，在线程运行之前这些数据就就已经到位了，这样就不会造成数据在线程运行后才传入的现象。如果要传递更复杂的数据，可以使用集合、类等数据结构。使用构造方法来传递数据虽然比较安全，但如果要传递的数据比较多时，就会造成很多不便。由于Java没有默认参数，要想实现类似默认参数的效果，就得使用重载，这样不但使构造方法本身过于复杂，又会使构造方法在数量上大增。因此，要想避免这种情况，就得通过类方法或类变量来传递数据。 

2、通过变量和方法传递数据 
向对象中传入数据一般有两次机会，第一次机会是在建立对象时通过构造方法将数据传入，另外一次机会就是在类中定义一系列的public的方法或变量（也可称之为字段）。然后在建立完对象后，通过对象实例逐个赋值。下面的代码是对MyThread1类的改版，使用了一个setName方法来设置 name变量： 

代码如下:

package mythread; 
public class MyThread2 implements Runnable 
{ 
private String name; 
public void setName(String name) 
{ 
this.name = name; 
} 
public void run() 
{ 
System.out.println("hello " + name); 
} 
public static void main(String[] args) 
{ 
MyThread2 myThread = new MyThread2(); 
myThread.setName("world"); 
Thread thread = new Thread(myThread); 
thread.start(); 
} 
} 

3、通过回调函数传递数据 
上面讨论的两种向线程中传递数据的方法是最常用的。但这两种方法都是main方法中主动将数据传入线程类的。这对于线程来说，是被动接收这些数据的。然而，在有些应用中需要在线程运行的过程中动态地获取数据，如在下面代码的run方法中产生了3个随机数，然后通过Work类的process方法求这三个随机数的和，并通过Data类的value将结果返回。从这个例子可以看出，在返回value之前，必须要得到三个随机数。也就是说，这个 value是无法事先就传入线程类的。 

代码如下:

package mythread; 
class Data 
{ 
public int value = 0; 
} 
class Work 
{ 
public void process(Data data, Integer numbers) 
{ 
for (int n : numbers) 
{ 
data.value += n; 
} 
} 
} 
public class MyThread3 extends Thread 
{ 
private Work work; 
public MyThread3(Work work) 
{ 
this.work = work; 
} 
public void run() 
{ 
java.util.Random random = new java.util.Random(); 
Data data = new Data(); 
int n1 = random.nextInt(1000); 
int n2 = random.nextInt(2000); 
int n3 = random.nextInt(3000); 
work.process(data, n1, n2, n3); // 使用回调函数 
System.out.println(String.valueOf(n1) + "+" + String.valueOf(n2) + "+" 
+ String.valueOf(n3) + "=" + data.value); 
} 
public static void main(String[] args) 
{ 
Thread thread = new MyThread3(new Work()); 
thread.start(); 
} 
} 

在上面代码中的process方法被称为回调函数。从本质上说，回调函数就是事件函数。在Windows API中常使用回调函数和调用API的程序之间进行数据交互。因此，调用回调函数的过程就是最原始的引发事件的过程。在这个例子中调用了process方法来获得数据也就相当于在run方法中引发了一个事件。

 

七.线程池

1.创建线程池基本方法：

(1)定义线程类

class Handler implements Runnable{}

 

(2)建立ExecutorService线程池

ExecutorService executorService = Executors.newCachedThreadPool();

或者

int cpuNums = Runtime.getRuntime().availableProcessors();  //获取当前系统的CPU 数目

ExecutorService executorService =Executors.newFixedThreadPool(cpuNums * POOL_SIZE); //ExecutorService通常根据系统资源情况灵活定义线程池大小

 

(3)调用线程池操作

循环操作，把新实例放入Executor池中

while(true){

        executorService.execute(new Handler(socket)); 

           // class Handler implements Runnable{

        或者

        executorService.execute(createTask(i));

            //private static Runnable createTask(final int taskID)

 

      }

 

2.ExecutorService线程池对象

 

1.newCachedThreadPool()	-缓存型池子，先查看池中有没有以前建立的线程，如果有，就reuse.如果没有，就建一个新的线程加入池中 -缓存型池子通常用于执行一些生存期很短的异步型任务  因此在一些面向连接的daemon型SERVER中用得不多。 -能reuse的线程，必须是timeout IDLE内的池中线程，缺省timeout是60s,超过这个IDLE时长，线程实例将被终止及移出池。   注意，放入CachedThreadPool的线程不必担心其结束，超过TIMEOUT不活动，其会自动被终止。
2.newFixedThreadPool	-newFixedThreadPool与cacheThreadPool差不多，也是能reuse就用，但不能随时建新的线程 -其独特之处:任意时间点，最多只能有固定数目的活动线程存在，此时如果有新的线程要建立，只能放在另外的队列中等待，直到当前的线程中某个线程终止直接被移出池子 -和cacheThreadPool不同，FixedThreadPool没有IDLE机制（可能也有，但既然文档没提，肯定非常长，类似依赖上层的TCP或UDP IDLE机制之类的），所以FixedThreadPool多数针对一些很稳定很固定的正规并发线程，多用于服务器 -从方法的源代码看，cache池和fixed 池调用的是同一个底层池，只不过参数不同: fixed池线程数固定，并且是0秒IDLE（无IDLE） cache池线程数支持0-Integer.MAX_VALUE(显然完全没考虑主机的资源承受能力），60秒IDLE
3.ScheduledThreadPool	-调度型线程池 -这个池子里的线程可以按schedule依次delay执行，或周期执行
4.SingleThreadExecutor	-单例线程，任意时间池中只能有一个线程 -用的是和cache池和fixed池相同的底层池，但线程数目是1-1,0秒IDLE（无IDLE）

 

实例：

class Handle implements Runnable {

private String name;

public Handle(String name) {

this.name = "thread"+name;

}

@Override

public void run() {

System.out.println( name +" Start. Time = "+new Date());

        processCommand();

        System.out.println( name +" End. Time = "+new Date());

}

private void processCommand() {

        try {

            Thread.sleep(1000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }

@Override

    public String toString(){

        return this.name;

    }

}

 

public static void testCachedThreadPool() {

System.out.println("Main: Starting at: "+ new Date());  

ExecutorService exec = Executors.newCachedThreadPool();   //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE

     for(int i = 0; i < 10; i++) {   

            exec.execute(new Handle(String.valueOf(i)));   

     }   

     exec.shutdown();  //执行到此处并不会马上关闭线程池,但之后不能再往线程池中加线程，否则会报错

     System.out.println("Main: Finished all threads at"+ new Date());

}

 

public static void testFixThreadPool() {

System.out.println("Main Thread: Starting at: "+ new Date());  

ExecutorService exec = Executors.newFixedThreadPool(5);   

     for(int i = 0; i < 10; i++) {   

            exec.execute(new Handle(String.valueOf(i)));   

     }   

     exec.shutdown();  //执行到此处并不会马上关闭线程池

     System.out.println("Main Thread: Finished at:"+ new Date());

}

 

public static void testSingleThreadPool() {

System.out.println("Main Thread: Starting at: "+ new Date());  

ExecutorService exec = Executors.newSingleThreadExecutor();   //创建大小为1的固定线程池

     for(int i = 0; i < 10; i++) {   

            exec.execute(new Handle(String.valueOf(i)));   

     }   

     exec.shutdown();  //执行到此处并不会马上关闭线程池

     System.out.println("Main Thread: Finished at:"+ new Date());

}

 

public static void testScheduledThreadPool() {

System.out.println("Main Thread: Starting at: "+ new Date());  

ScheduledThreadPoolExecutor  exec = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(10);   //创建大小为10的线程池

     for(int i = 0; i < 10; i++) {   

            exec.schedule(new Handle(String.valueOf(i)), 10, TimeUnit.SECONDS);//延迟10秒执行

     }   

     exec.shutdown();  //执行到此处并不会马上关闭线程池

     while(!exec.isTerminated()){

            //wait for all tasks to finish

     }

     System.out.println("Main Thread: Finished at:"+ new Date());

}

 

3.线程池的常用方法

(1)submit()

 

       将线程放入线程池中，除了使用execute，也可以使用submit，它们两个的区别是一个使用有返回值，一个没有返回值。submit的方法很适应于生产者-消费者模式，通过和Future结合一起使用，可以起到如果线程没有返回结果，就阻塞当前线程等待线程 池结果返回。

 

一般用第一种比较多

如下实例。注意，submit中的线程要实现接口Callable

 

package com.func.axc.executors;

 

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

 

/**

 * 功能概要：缓冲线程池实例-submit运行

 * 

 */

class TaskWithResult implements Callable<String> { 

    private int id; 

 

    public TaskWithResult(int id) { 

            this.id = id; 

    } 

 

    /** 

     * 任务的具体过程，一旦任务传给ExecutorService的submit方法，则该方法自动在一个线程上执行。 

     * 

     * @return 

     * @throws Exception 

     */

    public String call() throws Exception { 

            System.out.println("call()方法被自动调用,干活！！！             " + Thread.currentThread().getName()); 

            //一个模拟耗时的操作

            for (int i = 999999; i > 0; i--) ; 

            return"call()方法被自动调用，任务的结果是：" + id + "    " + Thread.currentThread().getName(); 

    } 

}

 

public class ThreadPool2 {

  public static void main(String[] args) { 

          ExecutorService executorService = Executors.newCachedThreadPool(); 

          List<Future<String>> resultList = new ArrayList<Future<String>>(); 

 

          //创建10个任务并执行

          for (int i = 0; i < 10; i++) { 

                  //使用ExecutorService执行Callable类型的任务，并将结果保存在future变量中

                  Future<String> future = executorService.submit(new TaskWithResult(i)); 

                  //将任务执行结果存储到List中

                  resultList.add(future); 

          } 

        //启动一次顺序关闭，执行以前提交的任务，但不接受新任务。如果已经关闭，则调用没有其他作用。

          executorService.shutdown(); 

          

          //遍历任务的结果

          for (Future<String> fs : resultList) { 

                  try { 

                          System.out.println(fs.get());     //打印各个线程（任务）执行的结果

                  } catch (InterruptedException e) { 

                          e.printStackTrace(); 

                  } catch (ExecutionException e) { 

                          e.printStackTrace(); 

                  } finally { 

                          

                  } 

          } 

  } 

}

结果如下：

 

从上面可以看到，输出结果的依次的。说明每次get都 阻塞了的。

 

看了下它的源码，其实它最终还是调用 了execute方法

 

    public <T> Future<T> submit(Callable<T> task) {

        if (task == null) throw new NullPointerException();

        RunnableFuture<T> ftask = newTaskFor(task);

        execute(ftask);

        return ftask;

    }

 

 

(2) execute()

表示往线程池添加线程，有可能会立即运行，也有可能不会。无法预知线程何时开始，何时线束。

 

主要源码如下：

 

    public void execute(Runnable command) {

        if (command == null)

            throw new NullPointerException();

        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

            if (runState == RUNNING && workQueue.offer(command)) {

                if (runState != RUNNING || poolSize == 0)

                    ensureQueuedTaskHandled(command);

            }

            else if (!addIfUnderMaximumPoolSize(command))

                reject(command); // is shutdown or saturated

        }

    }

 

(3)shutdown()

 

通常放在execute后面。如果调用 了这个方法，一方面，表明当前线程池已不再接收新添加的线程，新添加的线程会被拒绝执行。另一方面，表明当所有线程执行完毕时，回收线程池的资源。注意，它不会马上关闭线程池！

 

(4)shutdownNow()

 

不管当前有没有线程在执行，马上关闭线程池！这个方法要小心使用，要不可能会引起系统数据异常！

 

4.ThreadPoolExecutor技术内幕

 

经过上面的过程，基本上可以掌握线程池的一些基本用法。下面再来看看JAVA中线程池的源码实现。

首先是其继承关系如下：

通过观察上面四种线程池的源码：

 

如：newFixedThreadPool

 

    public static ExecutorService newFixedThreadPool(int nThreads) {

        return new ThreadPoolExecutor(nThreads, nThreads,

                                      0L, TimeUnit.MILLISECONDS,

                                      new LinkedBlockingQueue<Runnable>());

    }

 

如：newCachedThreadPool

    public static ExecutorService newCachedThreadPool() {

        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,

                                      60L, TimeUnit.SECONDS,

                                      new SynchronousQueue<Runnable>());

    }

 

如：newSingleThreadExecutor

    public static ExecutorService newSingleThreadExecutor() {

        return new FinalizableDelegatedExecutorService

            (new ThreadPoolExecutor(1, 1,

                                    0L, TimeUnit.MILLISECONDS,

                                    new LinkedBlockingQueue<Runnable>()));

    }

 

可以发现，其实它们调用的都是同一个接口ThreadPoolExecutor方法，只不过传入参数不一样而已。

 

从源码中可以看到其实最终都是调用了以下的方法：

 

    public ThreadPoolExecutor(int corePoolSize,

                              int maximumPoolSize,

                              long keepAliveTime,

                              TimeUnit unit,

                              BlockingQueue<Runnable> workQueue,

                              ThreadFactory threadFactory,

                              RejectedExecutionHandler handler) {

        if (corePoolSize < 0 ||

            maximumPoolSize <= 0 ||

            maximumPoolSize < corePoolSize ||

            keepAliveTime < 0)

            throw new IllegalArgumentException();

        if (workQueue == null || threadFactory == null || handler == null)

            throw new NullPointerException();

        this.corePoolSize = corePoolSize;

        this.maximumPoolSize = maximumPoolSize;

        this.workQueue = workQueue;

        this.keepAliveTime = unit.toNanos(keepAliveTime);

        this.threadFactory = threadFactory;

        this.handler = handler;

 

    }

 

 

总结：

       ThreadPoolExecutor中，包含了一个任务缓存队列和若干个执行线程，任务缓存队列是一个大小固定的缓冲区队列，用来缓存待执行的任务，执行线程用来处理待执行的任务。每个待执行的任务，都必须实现Runnable接口，执行线程调用其run()方法，完成相应任务。

ThreadPoolExecutor对象初始化时，不创建任何执行线程，当有新任务进来时，才会创建执行线程。

构造ThreadPoolExecutor对象时，需要配置该对象的核心线程池大小和最大线程池大小：

当目前执行线程的总数小于核心线程大小时，所有新加入的任务，都在新线程中处理

当目前执行线程的总数大于或等于核心线程时，所有新加入的任务，都放入任务缓存队列中

当目前执行线程的总数大于或等于核心线程，并且缓存队列已满，同时此时线程总数小于线程池的最大大小，那么创建新线程，加入线程池中，协助处理新的任务。

当所有线程都在执行，线程池大小已经达到上限，并且缓存队列已满时，就rejectHandler拒绝新的任务

 

 

 

5.自定义线程池

再来看看它的方法

    public ThreadPoolExecutor(int corePoolSize,//核心线程大小

                              int maximumPoolSize,//最大线程大小 

                              long keepAliveTime,//线程缓存时间

                              TimeUnit unit,//前面keepAlive

                              BlockingQueue<Runnable> workQueue,//缓存队列

                              ThreadFactory threadFactory,//线程工大

                              RejectedExecutionHandler handler)//拒绝策略

 

block queue有以下几种实现：

ArrayBlockingQueue : 有界的数组队列

 LinkedBlockingQueue : 可支持有界/无界的队列，使用链表实现

 PriorityBlockingQueue : 优先队列，可以针对任务排序

SynchronousQueue : 队列长度为1的队列，和Array有点区别就是：client thread提交到block queue会是一个阻塞过程，直到有一个worker thread连接上来poll task。当线

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略

 

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。

ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）

ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

 

比如定义如下一个线程池：

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardOldestPolicy());

 

这里核心线程数为2，最大线程数为4，线程缓存时间为3秒，缓冲队列的容量设置为3。线程工厂设置为默认

 

下面是一个具体实例：

 

package com.func.axc.executors;

 

import java.io.Serializable;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.Executors;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

 

 

public class MyThreadPoolTest {

 

private static int produceTaskSleepTime = 2;

private static int produceTaskMaxNumber = 10;

 

public static void main(String[] args) {

// 构造一个线程池

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardOldestPolicy());

 

for (int i = 1; i <= produceTaskMaxNumber; i++) {

try {

// 产生一个任务，并将其加入到线程池

String task = "task@ " + i;

System.out.println("put " + task);

threadPool.execute(new ThreadPoolTask(task));

// 便于观察，等待一段时间

Thread.sleep(produceTaskSleepTime);

} catch (Exception e) {

e.printStackTrace();

}

}

}

}

 

/**

 * 线程池执行的任务

 */

class ThreadPoolTask implements Runnable, Serializable {

private static final long serialVersionUID = 0;

private static int consumeTaskSleepTime = 2000;

// 保存任务所需要的数据

private Object threadPoolTaskData;

 

ThreadPoolTask(Object tasks) {

this.threadPoolTaskData = tasks;

}

 

public void run() {

// 处理一个任务，这里的处理方式太简单了，仅仅是一个打印语句

System.out.println(Thread.currentThread().getName());

System.out.println("start .." + threadPoolTaskData);

 

try {

// //便于观察，等待一段时间

Thread.sleep(consumeTaskSleepTime);

} catch (Exception e) {

e.printStackTrace();

}

threadPoolTaskData = null;

}

 

public Object getTask() {

return this.threadPoolTaskData;

}

}