反应堆模式

package tutorial.pattern;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;

public class ReactorPattern {

	DemultiPlexer demultiPlexer = new DemultiPlexer();

	public static void main(String args[]) throws InterruptedException, IOException {
		ReactorPattern reactorPattern = new ReactorPattern();
		ServerSocket server = new ServerSocket(1220);
		Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(server));

		long s = System.currentTimeMillis();
		for (;;) {
			if (System.currentTimeMillis() - s > 30000)
				break;
			Socket socket = server.accept();
			int i = socket.getInputStream().read();
			if (i != -1) {
				reactorPattern.getDemultiPlexer().accept(i);
			}			 
		}
		server.close();
	}
	public DemultiPlexer getDemultiPlexer() {
		return demultiPlexer;
	}
	
}

class Dispatcher {
	private Integer requestId;
	private Integer resourceId;
	private DemultiPlexer demultiPlexer;

	public Dispatcher(DemultiPlexer demultiPlexer, Integer requestId, Integer resourceId) {
		super();
		this.requestId = requestId;
		this.resourceId = resourceId;
		this.demultiPlexer = demultiPlexer;
	}

	/**
	 * create a request handler when demultiPlexer has an ideal resource.
	 * 
	 * @return RequestHandler
	 */
	public RequestHandler createRequestHandler() {
		return new RequestHandler(this, requestId, resourceId);
	}

	/**
	 * when request handler complete task, free the resource assigned.
	 */
	public synchronized void freeResource(Integer i) {
		demultiPlexer.returnResource(i);
	}

	public Integer getRequestId() {
		return requestId;
	}

	public void setRequestId(Integer requestId) {
		this.requestId = requestId;
	}

	public Integer getResourceId() {
		return resourceId;
	}

	public void setResourceId(Integer resourceId) {
		this.resourceId = resourceId;
	}

	public DemultiPlexer getDemultiPlexer() {
		return demultiPlexer;
	}

	public void setDemultiPlexer(DemultiPlexer demultiPlexer) {
		this.demultiPlexer = demultiPlexer;
	}

}

class RequestHandler extends Thread {
	private Integer tid;
	private Integer rid;
	private Dispatcher dispatcher;

	public RequestHandler(Dispatcher dispatcher, Integer tid, Integer rid) {
		this.tid = tid;
		this.rid = rid;
		this.dispatcher = dispatcher;
	}

	@Override
	public void run() {
		System.out.println("the request No.[" + tid + "] is handling the requesting with resource [" + rid + "]");
		try {
			sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("the request No.[" + tid + "] processed the requesting with resource [" + rid + "]");

		// free resource after request was handled
		dispatcher.freeResource(rid);

		System.out.println("return resource: " + rid);
	}
}

class DemultiPlexer {	
	private ArrayBlockingQueue<Integer> resources = new ArrayBlockingQueue<Integer>(2);
	{
		resources.add(1);
		resources.add(2);
	}

	private ArrayList<Integer> requests = new ArrayList<Integer>(5);

	/**
	 * 
	 * @param s
	 *            the starting time of get resource action
	 * @param timeout
	 *            , limit the request resource action within a time period in
	 *            second
	 * @return a integer represent a resource requesting
	 */
	public synchronized Integer getResource(long s, int timeout) {
		for (;;) {
			if ((System.currentTimeMillis() - s) / 1000 > timeout) {
				throw new RuntimeException("time out to get resource");
			} else if (resources.size() > 0) {
				return resources.poll();
			}
			try {
				Thread.sleep(500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public void returnResource(Integer i) {
		synchronized (resources) {
			resources.add(i);
		}
	}

	/**
	 * accept a request and create a new dispatcher to assign the resource to a
	 * request handler.
	 * 
	 * @param Request
	 */
	public void accept(Integer requestId) {
		requests.add(requestId);
		Integer rid = getResource(System.currentTimeMillis(), 5);
		Dispatcher d = new Dispatcher(this, requestId, rid);
		d.createRequestHandler().start();
	}
}

class Requester extends Thread {

	private String address = "localhost";

	private int port = 1220;

	private int requestID;

	@Override
	public void run() {
		try {
			Socket socket = new Socket(address, port);
			socket.getOutputStream().write(requestID);
			sleep(5000);
			socket.close();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("No." + requestID + " was sent...");
	}

	public static void main(String args[]) {
		for (int i=0; i< 10 ; i++){
			new Requester(i).start();
		}
	}

	public void setRequestID(int requestID) {
		this.requestID = requestID;
	}

	public Requester(int requestID) {
		this.requestID = requestID;
	}

	public int getRequestID() {
		return requestID;
	}
}

class ShutdownHookThread extends Thread {
	private ServerSocket server;

	public ShutdownHookThread(ServerSocket server) {
		this.server = server;
	}

	@Override
	public void run() {
		try {
			server.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			server = null;
		}
	}

}