多线程详解五


1.线程池

线程池就像数据库连接池一样,是一个对象池。所有的对象池都有一个共同的目的,那就是为了提高对象的使用率,从而达到提高程序效率的目的。比如对于Servlet,它被设计为多线程的(如果它是单线程的,你就可以想象,当1000个人同时请求一个网页时,在第一个人获得请求结果之前,其它999个人都在郁闷地等待),如果为每个用户的每一次请求都创建一个新的线程对象来运行的话,系统就会在创建线程和销毁线程上耗费很大的开销,大大降低系统的效率。因此,Servlet多线程机制背后有一个线程池在支持,线程池在初始化初期就创建了一定数量的线程对象,通过提高对这些对象的利用率,避免高频率地创建对象,从而达到提高程序的效率的目的。

下面实现一个最简单的线程池,从中理解它的实现原理。为此我们定义了四个类,它们的用途及具体实现如下:

Task(任务):这是个代表任务的抽象类,其中定义了一个deal()方法,继承Task抽象类的子类需要实现这个方法,并把这个任务需要完成的具体工作在deal()方法编码实现。线程池中的线程之所以被创建,就是为了执行各种各样数量繁多的任务的,为了方便线程对任务的处理,我们需要用Task抽象类来保证任务的具体工作统一放在deal()方法里来完成,这样也使代码更加规范。

Task的定义如下:

public abstract class Task {
    public enum State {
        /* 新建 */NEW, /* 执行中 */RUNNING, /* 已完成 */FINISHED
    }

    // 任务状态
    private State state = State.NEW;

    public void setState(State state) {
        this.state = state;
    }

    public State getState() {
        return state;
    }

    public abstract void deal();

TaskQueue(任务队列):在同一时刻,可能有很多任务需要执行,而程序在同一时刻只能执行一定数量的任务,当需要执行的任务数超过了程序所能承受的任务数时怎么办呢?这就有了先执行哪些任务,后执行哪些任务的规则。TaskQueue类就定义了这些规则中的一种,它采用的是FIFO(先进先出,英文名是First In First Out)的方式,也就是按照任务到达的先后顺序执行。

TaskQueue类的定义如下:

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

public class TaskQueue {
    private List queue = new LinkedList();

    // 添加一项任务
    public synchronized void addTask(Task task) {
        if (task != null) {
            queue.add(task);
        }
    }

    // 完成任务后将它从任务队列中删除
    public synchronized void finishTask(Task task) {
        if (task != null) {
            task.setState(Task.State.FINISHED);
            queue.remove(task);
        }
    }

    // 取得一项待执行任务
    public synchronized Task getTask() {
        Iterator it = queue.iterator();
        Task task;
        while (it.hasNext()) {
            task = it.next();
            // 寻找一个新建的任务
            if (Task.State.NEW.equals(task.getState())) {
                // 把任务状态置为运行中
                task.setState(Task.State.RUNNING);
                return task;
            }
        }
        return null;
    }
}

				
				

addTask(Task task)方法用于当一个新的任务到达时,将它添加到任务队列中。这里使用了LinkedList类来保存任务到达的先后顺序。finishTask(Task task)方法用于任务被执行完毕时,将它从任务队列中清除出去。getTask()方法用于取得当前要执行的任务。

TaskThread(执行任务的线程):它继承自Thread类,专门用于执行任务队列中的待执行任务。

public class TaskThread extends Thread {
    // 该线程所属的线程池
    private ThreadPoolService service;

    public TaskThread(ThreadPoolService tps) {
        service = tps;
    }

    public void run() {
        // 在线程池运行的状态下执行任务队列中的任务
        while (service.isRunning()) {
            TaskQueue queue = service.getTaskQueue();
            Task task = queue.getTask();
            if (task != null) {
                task.deal();
            }
            queue.finishTask(task);
        }
    }
}

ThreadPoolService(线程池服务类):这是线程池最核心的一个类。它在被创建了时候就创建了几个线程对象,但是这些线程并没有启动运行,但调用了start()方法启动线程池服务时,它们才真正运行。stop()方法可以停止线程池服务,同时停止池中所有线程的运行。而runTask(Task task)方法是将一个新的待执行任务交与线程池来运行。

ThreadPoolService类的定义如下:

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

public class ThreadPoolService {
    // 线程数
    public static final int THREAD_COUNT = 5;

    // 线程池状态
    private Status status = Status.NEW;

    private TaskQueue queue = new TaskQueue();

    public enum Status {
        /* 新建 */NEW, /* 提供服务中 */RUNNING, /* 停止服务 */TERMINATED
    }

    private List threads = new ArrayList();

    public ThreadPoolService() {
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread t = new TaskThread(this);
            threads.add(t);
        }
    }

    // 启动服务
    public void start() {
        this.status = Status.RUNNING;
        for (int i = 0; i < THREAD_COUNT; i++) {
            threads.get(i).start();
        }
    }

    // 停止服务
    public void stop() {
        this.status = Status.TERMINATED;
    }

    // 是否正在运行
    public boolean isRunning() {
        return status == Status.RUNNING;
    }

    // 执行任务
    public void runTask(Task task) {
        queue.addTask(task);
    }

    protected TaskQueue getTaskQueue() {
        return queue;
    }
}

				
				

完成了上面四个类,我们就实现了一个简单的线程池。现在我们就可以使用它了,下面的代码做了一个简单的示例:

public class SimpleTaskTest extends Task {
    @Override
    public void deal() {
        // do something
    }

    public static void main(String[] args) throws InterruptedException {
        ThreadPoolService service = new ThreadPoolService();
        service.start();
        // 执行十次任务
        for (int i = 0; i < 10; i++) {
            service.runTask(new SimpleTaskTest());
        }
        // 睡眠1秒钟,等待所有任务执行完毕
        Thread.sleep(1000);
        service.stop();
    }
}

当然,我们实现的是最简单的,这里只是为了演示线程池的实现原理。在实际应用中,根据情况的不同,可以做很多优化。比如:

调整任务队列的规则,给任务设置优先级,级别高的任务优先执行。

动态维护线程池,当待执行任务数量较多时,增加线程的数量,加快任务的执行速度;当任务较少时,回收一部分长期闲置的线程,减少对系统资源的消耗。

事实上Java5.0及以上版本已经为我们提供了线程池功能,无需再重新实现。这些类位于java.util.concurrent包中。

Executors类提供了一组创建线程池对象的方法,常用的有一下几个:

public static ExecutorService newCachedThreadPool() {
    // other code
}

public static ExecutorService newFixedThreadPool(int nThreads) {
    // other code
}

public static ExecutorService newSingleThreadExecutor() {
    // other code
}

newCachedThreadPool()方法创建一个动态的线程池,其中线程的数量会根据实际需要来创建和回收,适合于执行大量短期任务的情况;newFixedThreadPool(int nThreads)方法创建一个包含固定数量线程对象的线程池,nThreads代表要创建的线程数,如果某个线程在运行的过程中因为异常而终止了,那么一个新的线程会被创建和启动来代替它;而newSingleThreadExecutor()方法则只在线程池中创建一个线程,来执行所有的任务。

这三个方法都返回了一个ExecutorService类型的对象。实际上,ExecutorService是一个接口,它的submit()方法负责接收任务并交与线程池中的线程去运行。submit()方法能够接受Callable和Runnable两种类型的对象。它们的用法和区别如下:

Runnable接口:继承Runnable接口的类要实现它的run()方法,并将执行任务的代码放入其中,run()方法没有返回值。适合于只做某种操作,不关心运行结果的情况。

Callable接口:继承Callable接口的类要实现它的call()方法,并将执行任务的代码放入其中,call()将任务的执行结果作为返回值。适合于执行某种操作后,需要知道执行结果的情况。

无论是接收Runnable型参数,还是接收Callable型参数的submit()方法,都会返回一个Future(也是一个接口)类型的对象。该对象中包含了任务的执行情况以及结果。调用Future的boolean isDone()方法可以获知任务是否执行完毕;调用Object get()方法可以获得任务执行后的返回结果,如果此时任务还没有执行完,get()方法会保持等待,直到相应的任务执行完毕后,才会将结果返回。

我们用下面的一个例子来演示Java5.0中线程池的使用:

import java.util.concurrent.*;

public class ExecutorTest {
        public static void main(String[] args) throws InterruptedException,
                        ExecutionException {
                ExecutorService es = Executors.newSingleThreadExecutor();
                Future fr = es.submit(new RunnableTest());// 提交任务

                Future fc = es.submit(new CallableTest());// 提交任务
                // 取得返回值并输出
                System.out.println((String) fc.get());

                // 检查任务是否执行完毕
                if (fr.isDone()) {
                        System.out.println("执行完毕-RunnableTest.run()");
                } else {
                        System.out.println("未执行完-RunnableTest.run()");
                }

                // 检查任务是否执行完毕
                if (fc.isDone()) {
                        System.out.println("执行完毕-CallableTest.run()");
                } else {
                        System.out.println("未执行完-CallableTest.run()");
                }

                // 停止线程池服务
                es.shutdown();
        }
}

class RunnableTest implements Runnable {
        public void run() {
                System.out.println("已经执行-RunnableTest.run()");
        }
}

class CallableTest implements Callable {
        public Object call() {
                System.out.println("已经执行-CallableTest.call()");
                return "返回值-CallableTest.call()";
        }
}

运行结果:

已经执行-RunnableTest.run()

已经执行-CallableTest.call()

返回值-CallableTest.call()

执行完毕-RunnableTest.run()

执行完毕-CallableTest.run()

使用完线程池之后,需要调用它的shutdown()方法停止服务,否则其中的所有线程都会保持运行,程序不会退出。



联系我们 | 友情链接