Java 5.0 多线程编程实践

　}
} 

　　这里使用了Java 5的另外一个新特性泛型，声明TimeConsumingTask的时候使用了String做为类型参数。必须实现Callable接口的call函数，其作用类似与Runnable中的run函数，在call函数里写入要执行的代码，其返回值类型等同于在类声明中传入的类型值。在这段程序中，我们提交了一个Callable的任务，然后程序不会堵塞，而是继续执行dos.write("let’s do soemthing other".getBytes());当程序执行到String result = future.get()时如果call函数已经执行完毕，则取得返回值，如果还在执行，则等待其执行完毕。

服务器端的完整实现

　　服务器端的完整实现代码如下：

package com.andrew;

import Java.io.DataOutputStream;
import Java.io.IOException;
import Java.io.Serializable;
import Java.net.ServerSocket;
import Java.net.Socket;
import Java.util.concurrent.ArrayBlockingQueue;
import Java.util.concurrent.BlockingQueue;
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;
import Java.util.concurrent.RejectedExecutionHandler;
import Java.util.concurrent.ThreadPoolExecutor;
import Java.util.concurrent.TimeUnit;
import Java.util.concurrent.locks.ReentrantLock;

public class Server {
　private static int produceTaskSleepTime = 100;
　private static int consumeTaskSleepTime = 1200;
　private static int produceTaskMaxNumber = 100;
　private static final int CORE_POOL_SIZE = 2;
　private static final int MAX_POOL_SIZE = 100;
　private static final int KEEPALIVE_TIME = 3;
　private static final int QUEUE_CAPACITY = (CORE_POOL_SIZE + MAX_POOL_SIZE) / 2;
　private static final TimeUnit TIME_UNIT = TimeUnit.SECONDS;
　private static final String HOST = "127.0.0.1";
　private static final int PORT = 19527;
　private BlockingQueueworkQueue = new ArrayBlockingQueue(QUEUE_CAPACITY);
　//private ThreadPoolExecutor serverThreadPool = null;
　private ExecutorService pool = null;
　private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.DiscardOldestPolicy();
　private ServerSocket serverListenSocket = null;
　private int times = 5;
　public void start() {
　　// You can also init thread pool in this way.
　　/*serverThreadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,
　　MAX_POOL_SIZE, KEEPALIVE_TIME, TIME_UNIT, workQueue,
　　rejectedExecutionHandler);*/
　　pool = Executors.newFixedThreadPool(10);
　　try {
　　　serverListenSocket = new ServerSocket(PORT);
　　　serverListenSocket.setReuseAddress(true);

　　　System.out.println("I’m listening");
　　　while (times-- >0) {
　　　　Socket socket = serverListenSocket.accept();
　　　　String welcomeString = "hello";
　　　　//serverThreadPool.execute(new ServiceThread(socket, welcomeString));
　　　　pool.execute(new ServiceThread(socket));
　　　}
　　} catch (IOException e) {
　　　// TODO Auto-generated catch block
　　　e.printStackTrace();
　　}
　　cleanup();
　}

　public void cleanup() {
　　if (null != serverListenSocket) {
　　　try {
　　　　serverListenSocket.close();
　　　} catch (IOException e) {
　　　　// TODO Auto-generated catch block
　　　　e.printStackTrace();
　　　}
　　}
　　//serverThreadPool.shutdown();
　　pool.shutdown();
　}

　public static void main(String args[]) {
　　Server server = new Server();
　　server.start();
　}
}

class ServiceThread implements Runnable, Serializable {
　private static final long serialVersionUID = 0;
　private Socket connectedSocket = null;
　private String helloString = null;
　private static int count = 0;
　private static ReentrantLock lock = new ReentrantLock();