socketserver.java

一个基于PlaceLab的室内和室外的智能导航系统

package org.placelab.proxy;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Enumeration;
import java.util.List;
import java.util.Vector;

public abstract class SocketServer implements Runnable {

	private int port;
	private ServerSocket ss = null;
	private Thread listener = null;
	private volatile boolean listen = false;
	private Vector workers = new Vector();
	
	private long startTime = 0;
	private long numConnections = 0;
	private int threadCount = 1;
	private String label = "SocketServer";
	
	public SocketServer (int port, int threadCount) {
		this.port = port;
		this.threadCount = threadCount;
	}
	
	public SocketServer (int port, int threadCount, String label) {
		this.port = port;
		this.threadCount = threadCount;
		this.label = label;
	}
	
	public void startServer () throws Exception {
		if (listener != null) 
			throw new Exception("Already started.");
					
		ss = new ServerSocket(port);
		listen = true;
		listener = new Thread(this);
		listener.start();
		
//		System.err.println(label + " - Started on port " + port + ".");
		startTime = System.currentTimeMillis();
		numConnections = 0;
		
		for (int i=0;i<threadCount;i++) 
			new SocketWorker(workers).start();
	}
	
	public void stopServer () {
		
		if (!listen)
			return;
		
		listen = false;
		
		try {
			ss.close();
			listener.join();
			listener = null;
			ss = null;
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		// Wait for worker threads to die.
		while (workers.size() > 0) {
//			System.err.println(label +" - Waiting on " + workers.size() + " worker threads to die.");
			((SocketWorker) workers.get(0)).kill();
		}
		
//		System.err.println(label + " - Stopped.");
		startTime = 0;
	}
	
	public SocketWorker getWorker() {
		
		// it will only ever take two passes to find an available thread.
		for (int t=0;t<2;t++) {	
			for (Enumeration e = workers.elements(); e.hasMoreElements();) {
				SocketWorker sw = (SocketWorker) e.nextElement();
				if (!sw.isBusy())
					return sw;
			}
			
			// wait for a new thread to become available.
			synchronized(this) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					
				}
			}
		}

		return null;
	}
	
	public void run () {
		while(listen) {
			try {
				Socket s = ss.accept();
				
				numConnections++;
				getWorker().setSocket(s);
				
				
			} catch (IOException e) {
				// occurs on stopServer
			}
		}
	}
	
	public abstract void serviceSocket (Socket s);
	
	public long getNumberOfConnections () {
		return numConnections;
	}
	
	public long getStartTimeMillis () {
		return startTime;
	}
	
	public int getPort () {
		return port;
	}
	
	class SocketWorker extends Thread {
		Socket s;
		List pool;
		volatile boolean kill = false;
		volatile boolean isBusy = false;
		
		public SocketWorker (List pool) {
			this.pool = pool;
			pool.add(this);
		}
		
		public boolean isBusy() {
			return isBusy;
		}
		
		public void setSocket (Socket s) {
			this.s = s;
			isBusy = true;
			synchronized(this) {
				// notify our run() that we have a new socket
				this.notify();
			}
 		}
		
		public void kill () {
			
			kill = true;
			
			// push thread out of its wait() condition
			synchronized(this) {
				this.notify();
			}
			pool.remove(this);
		}
		
		public void run () {
			
			while (!kill) {
				synchronized(this) {
					try {
						this.wait();
						if (kill)
							return;
					} catch (InterruptedException e) {
						
					}
				}
				
				try {
					s.setTcpNoDelay(true);
					serviceSocket(s);
					s.shutdownInput();
					s.shutdownOutput();
					s.close();
				} catch (Exception e) {
					
				}
				
				isBusy = false;
				
				// notify when a new thread becomes available
				synchronized(SocketServer.this) {
					SocketServer.this.notify();
				}
			}
			
		}
	
	}
}