java5线程框架Executor的用法举例

java.util.concurrent 与java线程框架有关的类主要都在这个包下。
java.util.concurrent.atomic
包含了不用加锁情况下就能改变值的原子变量。

java.util.concurrent.locks
包包含锁定的工具

Executor 是 java5 下的一个多任务并发执行框架(Doug
Lea),可以建立一个类似数据库连接池的线程池来执行任务。这个框架主要由三个接口和其相应的具体类组成。Executor、
ExecutorService 和 ScheduledExecutorService 。

1
、 Executor 接口：是用来执行 Runnable 任务的；它只定义一个方法- execute(Runnable command)；执行 Ruannable 类型的任务。

2
、 ExecutorService 接口： 继承Executor接口，提供了执行Callable任务和中止任务执行的服务。

3
、 ScheduledExecutorService 接口：继承 ExecutorService 接口，提供了按排程执行任务的服务。

4
、 Executors 类：为了方便使用, 建议使用 Executors的工具类来得到 Executor 接口的具体对象。

Executors 类有几个重要的方法，在这里简明一下：

1
、 callable(Runnable task)： 将 Runnable 的任务转化成 Callable 的任务

2
、 newSingleThreadExecutor()： 产生一个 ExecutorService 对象，这个对象只有一个线程可用来执行任务，若任务多于一个，任务将按先后顺序执行。

3
、 newCachedThreadPool()： 产生一个 ExecutorService 对象，这个对象带有一个线程池，线程池的大小会根据需要调整，线程执行完任务后返回线程池，供执行下一次任务使用。

4
、 newFixedThreadPool(int poolSize)： 产生一个 ExecutorService 对象，这个对象带有一个大小为 poolSize 的线程池，若任务数量大于 poolSize ，任务会被放在一个 queue 里顺序执行。

5
、 newSingleThreadScheduledExecutor()： 产生一个 ScheduledExecutorService 对象，这个对象的线程池大小为 1 ，若任务多于一个，任务将按先后顺序执行。

6
、 newScheduledThreadPool(int poolSize)： 产生一个 ScheduledExecutorService 对象，这个对象的线程池大小为 poolSize ，若任务数量大于 poolSize ，任务会在一个 queue 里等待执行 。

有关Executor框架其它类的说明请参看JAVA 5 的 API文档

下面是几个简单的例子，用以示例Executors中几个主要方法的使用。

1、 Task.java 任务

2、 SingleThreadExecutorTest.java 单线程执行程序的测试

3、 CachedThreadPoolTest.java 线程池线程执行程序的测试

4、 FixedThreadPoolTest.java 线程池线程执行程序的测试(线程数固定)

5、 DaemonThreadFactory.java 守护线程生成工厂

6、 MaxPriorityThreadFactory.java 大优先级线程生成工厂

7、 MinPriorityThreadFactory.java 小优先级线程生成工厂

8、 ThreadFactoryExecutorTest.java 在自定义线程生成工厂下的测试

=============== 1、 Task.java

package Executor;

//可执行任务

public class Task implements Runnable {

// 中断信号

volatile boolean stop = false;

// 该任务执行的次数

private int runCount = 0;

// 任务标识

private int taskId;

public Task(int taskId) {

this.taskId = taskId;

System.out.println("Create Task-" + taskId);

}

// 执行任务

public void run() {

while (!stop) {

try {

Thread.sleep(10);

} catch (InterruptedException e) {

System.out.println("Task interrupted...");

}

// 线程运行3次后,中断信号置为true

if (++runCount == 3)

stop = true;

// 输出一些语句

System.out.println("" + Thread.currentThread().toString() + "/t/t/t/t execute Task-" + taskId + "'s " + runCount

+ "th run. ");

}

}

}

=============== 1 end

=============== 2、 SingleThreadExecutorTest.java

package Executor;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class SingleThreadExecutorTest {

public static void main(String[] args) {

try {

// 创建一个单线程执行程序

ExecutorService executorService = Executors.newSingleThreadExecutor();

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

executorService.execute(new Task(i));

}

executorService.shutdown();

} catch (Exception e) {}

}

}

=============== 2 end

=============== 3、 CachedThreadPoolTest.java

package Executor;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class CachedThreadPoolTest {

public static void main(String[] args) {

try {

// 建新线程的线程池，如果之前构造的线程可用则重用它们

ExecutorService executorService = Executors.newCachedThreadPool();

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

executorService.execute(new Task(i));

}

executorService.shutdown();

} catch (Exception e) {}

}

}

=============== 3 end

=============== 4、 FixedThreadPoolTest.java

package Executor;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class FixedThreadPoolTest {

public static void main(String[] args) {

try {

// 创建固定线程数的线程池，以共享的无界队列方式来运行这些线程

ExecutorService executorService = Executors.newFixedThreadPool(2);

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

executorService.execute(new Task(i));

}

executorService.shutdown();

} catch (Exception e) {}

}

}

=============== 4 end

=============== 5、 DaemonThreadFactory.java

package Executor;

import java.util.concurrent.ThreadFactory;

public class DaemonThreadFactory implements ThreadFactory {

//创建一个守护线程

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

t.setDaemon(true);

return t;

}

}

=============== 5 end

=============== 6、 MaxPriorityThreadFactory.java

package Executor;

import java.util.concurrent.ThreadFactory;

public class MaxPriorityThreadFactory implements ThreadFactory {

//创建一个最大优先级的线程

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

//优先级最大、意思是切换到这个线程的概率比其它的低一些

t.setPriority(Thread.MAX_PRIORITY);

return t;

}

}

=============== 6 end

=============== 7、 MinPriorityThreadFactory.java

package Executor;

import java.util.concurrent.ThreadFactory;

public class MinPriorityThreadFactory implements ThreadFactory {

//创建一个最小优先级的线程

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

//优先级最小、意思是切换到这个线程的概率比其它的低一些

t.setPriority(Thread.MIN_PRIORITY);

return t;

}

}

=============== 7 end

=============== 8、 ThreadFactoryExecutorTest.java

package Executor;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

public class ThreadFactoryExecutorTest {

public static void main(String[] args) {

try {

// 创建一个单线程执行程序

ExecutorService defaultExecutor = Executors.newCachedThreadPool();

ExecutorService daemonExec = Executors

.newCachedThreadPool(new DaemonThreadFactory());

ExecutorService maxPriorityExecutor = Executors

.newCachedThreadPool(new MaxPriorityThreadFactory());

ExecutorService minPriorityExecutor = Executors

.newCachedThreadPool(new MinPriorityThreadFactory());

//用守护线程执行任务

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

daemonExec.execute(new Task(i));

}

//用其它线程执行任务

for (int j = 10; j <= 20; j++){

if (j == 10)

maxPriorityExecutor.execute(new Task(j));

else if (j == 11)

minPriorityExecutor.execute(new Task(j));

else

defaultExecutor.execute(new Task(j));

}

} catch (Exception e) {}

}

}

=============== 8 end