转载java多线程设计模式（二）

http://xuefeng.javaeedev.com

---

Guarded Suspention

Guarded Suspention模式主要思想是：

当条件不满足时，线程等待，直到条件满足时，等待该条件的线程被唤醒。

我们设计一个客户端线程和一个服务器线程，客户端线程不断发送请求给服务器线程，服务器线程不断处理请求。当请求队列为空时，服务器线程就必须等待，直到客户端发送了请求。

先定义一个请求队列：Queue

java 代码

package com.javaeedev.thread;

import java.util.*;

public class Queue {
    private List queue = new LinkedList();
    public synchronized Request getRequest() {
        while(queue.size()==0) {
            try {
                this.wait();
            }
            catch(InterruptedException ie) {
                return null;
            }
        }
        return (Request)queue.remove(0);
    }

    public synchronized void putRequest(Request request) {
        queue.add(request);
        this.notifyAll();
    }

}

 

蓝色部分就是服务器线程的等待条件，而客户端线程在放入了一个request后，就使服务器线程等待条件满足，于是唤醒服务器线程。

客户端线程：ClientThread

java 代码

package com.crackj2ee.thread;

public class ClientThread extends Thread {
    private Queue queue;
    private String clientName;

    public ClientThread(Queue queue, String clientName) {
        this.queue = queue;
        this.clientName = clientName;
    }

    public String toString() {
        return "[ClientThread-" + clientName + "]";
    }

    public void run() {
        for(int i=0; i<100; i++) {
            Request request = new Request("" + (long)(Math.random()*10000));
            System.out.println(this + " send request: " + request);
            queue.putRequest(request);
            try {
                Thread.sleep((long)(Math.random() * 10000 + 1000));
            }
            catch(InterruptedException ie) {
            }
        }
        System.out.println(this + " shutdown.");
    }
}

 

服务器线程：ServerThread

java 代码

package com.crackj2ee.thread;
public class ServerThread extends Thread {
    private boolean stop = false;
    private Queue queue;

    public ServerThread(Queue queue) {
        this.queue = queue;
    }

    public void shutdown() {
        stop = true;
        this.interrupt();
        try {
            this.join();
        }
        catch(InterruptedException ie) {}
    }

    public void run() {
        while(!stop) {
            Request request = queue.getRequest();
            System.out.println("[ServerThread] handle request: " + request);
            try {
                Thread.sleep(2000);
            }
            catch(InterruptedException ie) {}
        }
        System.out.println("[ServerThread] shutdown.");
    }
}

 

服务器线程在红色部分可能会阻塞，也就是说，Queue.getRequest是一个阻塞方法。这和java标准库的许多IO方法类似。

最后，写一个Main来启动他们：

java 代码

package com.crackj2ee.thread;

public class Main {

    public static void main(String[] args) {
        Queue queue = new Queue();
        ServerThread server = new ServerThread(queue);
        server.start();
        ClientThread[] clients = new ClientThread[5];
        for(int i=0; i             clients[i] = new ClientThread(queue, ""+i);
            clients[i].start();
        }
        try {
            Thread.sleep(100000);
        }
        catch(InterruptedException ie) {}
        server.shutdown();
    }
}

 

我们启动了5个客户端线程和一个服务器线程，运行结果如下：

[ClientThread-0] send request: Request-4984
[ServerThread] handle request: Request-4984
[ClientThread-1] send request: Request-2020
[ClientThread-2] send request: Request-8980
[ClientThread-3] send request: Request-5044
[ClientThread-4] send request: Request-548
[ClientThread-4] send request: Request-6832
[ServerThread] handle request: Request-2020
[ServerThread] handle request: Request-8980
[ServerThread] handle request: Request-5044
[ServerThread] handle request: Request-548
[ClientThread-4] send request: Request-1681
[ClientThread-0] send request: Request-7859
[ClientThread-3] send request: Request-3926
[ServerThread] handle request: Request-6832
[ClientThread-2] send request: Request-9906
......

可以观察到ServerThread处理来自不同客户端的请求。

思考

Q: 服务器线程的wait条件while(queue.size()==0)能否换成if(queue.size()==0)?

A: 在这个例子中可以，因为服务器线程只有一个。但是，如果服务器线程有多个（例如Web应用程序有多个线程处理并发请求，这非常普遍），就会造成严重问题。

Q: 能否用sleep(1000)代替wait()?

A: 绝对不可以。sleep()不会释放锁，因此sleep期间别的线程根本没有办法调用getRequest()和putRequest()，导致所有相关线程都被阻塞。

Q: (Request)queue.remove(0)可以放到synchronized() {}块外面吗？

A: 不可以。因为while()是测试queue，remove()是使用queue，两者是一个原子操作，不能放在synchronized外面。

总结

多线程设计看似简单，实际上必须非常仔细地考虑各种锁定/同步的条件，稍不小心，就可能出错。并且，当线程较少时，很可能发现不了问题，一旦问题出现又难以调试。

所幸的是，已有一些被验证过的模式可以供我们使用，我们会继续介绍一些常用的多线程设计模式。

源代码地址：http://javaeedev.googlecode.com/svn/trunk/GuardedSuspension/

Zip包下载：http://javaeedev.googlecode.com/files/GuardedSuspension.zip

---

Worker Pattern

前面谈了多线程应用程序能极大地改善用户相应。例如对于一个Web应用程序，每当一个用户请求服务器连接时，服务器就可以启动一个新线程为用户服务。

然而，创建和销毁线程本身就有一定的开销，如果频繁创建和销毁线程，CPU和内存开销就不可忽略，垃圾收集器还必须负担更多的工作。因此，线程池就是为了避免频繁创建和销毁线程。

每当服务器接受了一个新的请求后，服务器就从线程池中挑选一个等待的线程并执行请求处理。处理完毕后，线程并不结束，而是转为阻塞状态再次被放入线程池中。这样就避免了频繁创建和销毁线程。

Worker Pattern实现了类似线程池的功能。首先定义Task接口：

java 代码

package com.crackj2ee.thread;
public interface Task {
    void execute();
}

线程将负责执行execute()方法。注意到任务是由子类通过实现execute()方法实现的，线程本身并不知道自己执行的任务。它只负责运行一个耗时的execute()方法。

 

具体任务由子类实现，我们定义了一个CalculateTask和一个TimerTask：

java 代码

// CalculateTask.java
package com.crackj2ee.thread;
public class CalculateTask implements Task {
    private static int count = 0;
    private int num = count;
    public CalculateTask() {
        count++;
    }
    public void execute() {
        System.out.println("[CalculateTask " + num + "] start...");
        try {
            Thread.sleep(3000);
        }
        catch(InterruptedException ie) {}
        System.out.println("[CalculateTask " + num + "] done.");
    }
}

// TimerTask.java
package com.crackj2ee.thread;
public class TimerTask implements Task {
    private static int count = 0;
    private int num = count;
    public TimerTask() {
        count++;
    }
    public void execute() {
        System.out.println("[TimerTask " + num + "] start...");
        try {
            Thread.sleep(2000);
        }
        catch(InterruptedException ie) {}
        System.out.println("[TimerTask " + num + "] done.");
    }
}

以上任务均简单的sleep若干秒。

TaskQueue实现了一个队列，客户端可以将请求放入队列，服务器线程可以从队列中取出任务：

java 代码

package com.crackj2ee.thread;
import java.util.*;
public class TaskQueue {
    private List queue = new LinkedList();
    public synchronized Task getTask() {
        while(queue.size()==0) {
            try {
                this.wait();
            }
            catch(InterruptedException ie) {
                return null;
            }
        }
        return (Task)queue.remove(0);
    }
    public synchronized void putTask(Task task) {
        queue.add(task);
        this.notifyAll();
    }
}

 

终于到了真正的WorkerThread，这是真正执行任务的服务器线程：

java 代码

package com.crackj2ee.thread;
public class WorkerThread extends Thread {
    private static int count = 0;
    private boolean busy = false;
    private boolean stop = false;
    private TaskQueue queue;
    public WorkerThread(ThreadGroup group, TaskQueue queue) {
        super(group, "worker-" + count);
        count++;
        this.queue = queue;
    }
    public void shutdown() {
        stop = true;
        this.interrupt();
        try {
            this.join();
        }
        catch(InterruptedException ie) {}
    }
    public boolean isIdle() {
        return !busy;
    }
    public void run() {
        System.out.println(getName() + " start.");
        while(!stop) {
            Task task = queue.getTask();
            if(task!=null) {
                busy = true;
                task.execute();
                busy = false;
            }
        }
        System.out.println(getName() + " end.");
    }
}

前面已经讲过，queue.getTask()是一个阻塞方法，服务器线程可能在此wait()一段时间。此外，WorkerThread还有一个shutdown方法，用于安全结束线程。

 

最后是ThreadPool，负责管理所有的服务器线程，还可以动态增加和减少线程数：

java 代码

package com.crackj2ee.thread;
import java.util.*;
public class ThreadPool extends ThreadGroup {
    private List threads = new LinkedList();
    private TaskQueue queue;
    public ThreadPool(TaskQueue queue) {
        super("Thread-Pool");
        this.queue = queue;
    }
    public synchronized void addWorkerThread() {
        Thread t = new WorkerThread(this, queue);
        threads.add(t);
        t.start();
    }
    public synchronized void removeWorkerThread() {
        if(threads.size()>0) {
            WorkerThread t = (WorkerThread)threads.remove(0);
            t.shutdown();
        }
    }
    public synchronized void currentStatus() {
        System.out.println("-----------------------------------------------");
        System.out.println("Thread count = " + threads.size());
        Iterator it = threads.iterator();
        while(it.hasNext()) {
            WorkerThread t = (WorkerThread)it.next();
            System.out.println(t.getName() + ": " + (t.isIdle() ? "idle" : "busy"));
        }
        System.out.println("-----------------------------------------------");
    }
}

currentStatus()方法是为了方便调试，打印出所有线程的当前状态。

 

最后，Main负责完成main()方法：

java 代码

package com.crackj2ee.thread;
public class Main {
    public static void main(String[] args) {
        TaskQueue queue = new TaskQueue();
        ThreadPool pool = new ThreadPool(queue);
        for(int i=0; i<10; i++) {
            queue.putTask(new CalculateTask());
            queue.putTask(new TimerTask());
        }
        pool.addWorkerThread();
        pool.addWorkerThread();
        doSleep(8000);
        pool.currentStatus();
        pool.addWorkerThread();
        pool.addWorkerThread();
        pool.addWorkerThread();
        pool.addWorkerThread();
        pool.addWorkerThread();
        doSleep(5000);
        pool.currentStatus();
    }
    private static void doSleep(long ms) {
        try {
            Thread.sleep(ms);
        }
        catch(InterruptedException ie) {}
    }
}

 

main()一开始放入了20个Task，然后动态添加了一些服务线程，并定期打印线程状态，运行结果如下：

worker-0 start.
[CalculateTask 0] start...
worker-1 start.
[TimerTask 0] start...
[TimerTask 0] done.
[CalculateTask 1] start...
[CalculateTask 0] done.
[TimerTask 1] start...
[CalculateTask 1] done.
[CalculateTask 2] start...
[TimerTask 1] done.
[TimerTask 2] start...
[TimerTask 2] done.
[CalculateTask 3] start...
-----------------------------------------------
Thread count = 2
worker-0: busy
worker-1: busy
-----------------------------------------------
[CalculateTask 2] done.
[TimerTask 3] start...
worker-2 start.
[CalculateTask 4] start...
worker-3 start.
[TimerTask 4] start...
worker-4 start.
[CalculateTask 5] start...
worker-5 start.
[TimerTask 5] start...
worker-6 start.
[CalculateTask 6] start...
[CalculateTask 3] done.
[TimerTask 6] start...
[TimerTask 3] done.
[CalculateTask 7] start...
[TimerTask 4] done.
[TimerTask 7] start...
[TimerTask 5] done.
[CalculateTask 8] start...
[CalculateTask 4] done.
[TimerTask 8] start...
[CalculateTask 5] done.
[CalculateTask 9] start...
[CalculateTask 6] done.
[TimerTask 9] start...
[TimerTask 6] done.
[TimerTask 7] done.
-----------------------------------------------
Thread count = 7
worker-0: idle
worker-1: busy
worker-2: busy
worker-3: idle
worker-4: busy
worker-5: busy
worker-6: busy
-----------------------------------------------
[CalculateTask 7] done.
[CalculateTask 8] done.
[TimerTask 8] done.
[TimerTask 9] done.
[CalculateTask 9] done.

仔细观察：一开始只有两个服务器线程，因此线程状态都是忙，后来线程数增多，7个线程中的两个状态变成idle，说明处于wait()状态。

思考：本例的线程调度算法其实根本没有，因为这个应用是围绕TaskQueue设计的，不是以Thread Pool为中心设计的。因此，Task调度取决于TaskQueue的getTask()方法，你可以改进这个方法，例如使用优先队列，使优先级高的任务先被执行。

如果所有的服务器线程都处于busy状态，则说明任务繁忙，TaskQueue的队列越来越长，最终会导致服务器内存耗尽。因此，可以限制 TaskQueue的等待任务数，超过最大长度就拒绝处理。许多Web服务器在用户请求繁忙时就会拒绝用户：HTTP 503 SERVICE UNAVAILABLE

从JDK 5开始，java.util.concurrent包已经内置了Worker线程模式，无需我们手动编写上述代码。不过，理解Worker模式的原理非常重要。

---