多线程常用模式总结二

6.Thread-Per-Message模式
这个模式在swing开发中随处可见。比如我们希望点击一个按钮就启动文件系统扫描程序，一般的做法就是

okButton.addActionListener(new ActionListener(){
			public void actionPerformed(ActionEvent e) {
				new Thread(new Runnable(){
                            public void run() {
				doFileSystemScan();
			}}).start();
			}});

简单的理解就是来一个请求，开启一个线程去处理请求。这样的应用场景也见于Servlet容器处理http请求，socket处理文件读写操作等。这个设计主要是用来解决一些耗时的操作，提高响应性，但是同时也会产生额外的线程的启动开销和无法得到响应结果。创建和销毁线程本身就有一定的开销，如果频繁创建和销毁线程，CPU和内存开销就不可忽略，垃圾收集器还必须负担更多的工作。为解决线程开销的问题，有了worker线程模式；为解决响应结果的问题，有了Future模式。

7.Worker线程模式
先初始化一个线程池，所有线程处于阻塞状态。接受到一个新的请求后，就从线程池中挑选一个等待的线程并执行请求处理。处理完毕后，线程并不结束，而是转为阻塞状态再次被放入线程池中，这样就避免了频繁创建和销毁线程。以一个简单的例子。
首先是请求类Request

public class Request {
    private final String name;
    private final int number;  
    private static final Random random = new Random();
    public Request(String name, int number) {
        this.name = name;
        this.number = number;
    }
    public void execute() {
        System.out.println(Thread.currentThread().getName() + " executes " + this);
        try {
            Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException ignore) {
        }
    }
    public String toString() {
        return "[ Request from " + name + " No." + number + " ]";
    }
}

再是请求处理器，它会接收所有请求，队列缓冲起来，交给workerThread执行。

public class RequestProcesser {
    private static final int MAX_REQUEST = 100;
    private final Request[] requestQueue;
    private int tail;
    private int head;
    private int count; // Request的数量

    private final WorkerThread[] threadPool;

    public RequestProcesser(int threads) {
        this.requestQueue = new Request[MAX_REQUEST];
        this.head = 0;
        this.tail = 0;
        this.count = 0;

        threadPool = new WorkerThread[threads];
        for (int i = 0; i < threadPool.length; i++) {
            threadPool[i] = new WorkerThread("Worker-" + i, this);
        }
    }
    
    /**
     * 启动所有工作线程
     */
    public void startWorkers() {
        for (int i = 0; i < threadPool.length; i++) {
            threadPool[i].start();
        }
    }
    
    /**
     * 放入请求到工作队列
     * @param request
     */
    public synchronized void putRequest(Request request) {
        while (count >= requestQueue.length) {
            try {
                wait();
            } catch (InterruptedException ignore) {
            }
        }
        requestQueue[tail] = request;
        tail = (tail + 1) % requestQueue.length;
        count++;
        notifyAll();
    }
    
    /**
     * 得到最近的请求
     * @return
     */
    public synchronized Request takeRequest() {
        while (count <= 0) {
            try {
                wait();
            } catch (InterruptedException ignore) {
            }
        }
        Request request = requestQueue[head];
        head = (head + 1) % requestQueue.length;
        count--;
        notifyAll();
        return request;
    }
}

接下来是真正的workerThread.

public class WorkerThread extends Thread {
    private final RequestProcesser queue;
    public WorkerThread(String name, RequestProcesser queue) {
        super(name);
        this.queue = queue;
    }
    public void run() {
        while (true) {
            Request request = queue.takeRequest();
            request.execute();
        }
    }
}

下面是模拟并发请求的clientThread

public class ClientThread extends Thread {
    private final RequestProcesser processer;
    private static final Random random = new Random();
    public ClientThread(String name, RequestProcesser processer) {
        super(name);
        this.processer = processer;
    }
    public void run() {
        try {
            for (int i = 0; true; i++) {
                Request request = new Request(getName(), i);
                processer.putRequest(request);
                Thread.sleep(random.nextInt(1000));
            }
        } catch (InterruptedException e) {
        }
    }
}

最后用个main类来跑下这个小程序。

public class Main {
    public static void main(String[] args) {
        RequestProcesser processer = new RequestProcesser(5);
        processer.startWorkers();
        new ClientThread("hankzhang", processer).start();
        new ClientThread("yashironan", processer).start();
        new ClientThread("yunyunzei", processer).start();
    }
}

在这个小程序里面，当队列满，采用的方式是client线程阻塞。对于一般的web服务器在用户请求繁忙时就会拒绝用户：HTTP 503 SERVICE UNAVAILABLE。

8.Future模式
Thread-Per-Message模式引申出2个问题，worker Thread解决了一个，Future就是用来解决另一个的。我个人理解就是，在请求发出后，马上可以获取返回值Future，但是这个返回值Future并不是真正的值，因为线程还在跑呢，跑完之前不会有结果出来的。Future中有个getContent方法就是获取真正的返回值的，但是这个方法是阻塞性的，结果返回之前是不会跳出来的。
首先是Future接口。

public interface Future {
    public abstract String getContent();
}

然后是它的实现类。

public class FutureTask implements Future {
    private RealData realdata = null;
    private boolean ready = false;
    public synchronized void setRealData(RealData 
realdata) {
        if (ready) {                        
            return;
        }
        this.realdata = realdata;
        this.ready = true;
        notifyAll();
    }
    public synchronized String getContent() {
        while (!ready) {
            try {
                wait();
            } catch (InterruptedException e) {
            }
        }
        return realdata.getContent();
    }
}

再是真实的返回数据类。

public class RealData implements Future {
    private final String content;
    public RealData(int count, char c) {
        System.out.println(" making RealData(" + count + ", " + c + ") BEGIN");
        char[] buffer = new char[count];
        for (int i = 0; i < count; i++) {
            buffer[i] = c;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
            }
        }
        System.out.println(" making RealData(" + count + ", " + c + ") END");
        this.content = new String(buffer);
    }
    public String getContent() {
        return content;
    }
}

用以模拟请求处理的Host。

public class Host {
    public Future request(final int count, final char c) {
        System.out.println(" request(" + count + ", " + c + ") BEGIN");
        //建立FutureData
        final FutureTask future = new FutureTask();
        //建立RealData，启动新线程
        new Thread() {                                      
            public void run() {                             
                RealData realdata = new RealData(count, c);
                future.setRealData(realdata);
            }                                               
        }.start();                                          
        System.out.println(" request(" + count + ", " + c + ") END");
        //取回FutureData，作为传回值
        return future;
    }
}

用个Main程序跑起来

public class Main {
    public static void main(String[] args) {
        System.out.println("main BEGIN");
        Host host = new Host();
        Future data1 = host.request(10, 'A');
        Future data2 = host.request(20, 'B');
        Future data3 = host.request(30, 'C');

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
        }

        System.out.println("data1 = " + data1.getContent());
        System.out.println("data2 = " + data2.getContent());
        System.out.println("data3 = " + data3.getContent());
        System.out.println("main END");
    }
}

jdk1.5后，有了真正的Future接口以及它的实现类FutureTask，实现的更为精巧，详细。