proxyserver.java

java HTTP代理多线程监听和处理连接 java HTTP代理多线程监听和处理连接

package proxy;


import java.io.*;
import java.net.*;
import java.util.*;

/**
 * 代理服务器
 * @author liwen
 *
 */

public class ProxyServer {
    static public boolean _DEBUG_ = true;
    static public final int SERVER_TIME_OUT = 800;
    static public final int CONNECT_RETRIES = 4;
    static public final int CONNECT_RETRIES_SLEEP = 2000;
    static public final int BUFER_SIZE = 1024;
    static public final int HTTP_PORT = 80;
    static public final int AUTHENTICATION_TYPE_HTTP = 1;
    static public boolean LOG_STREAM_DATA = false;
    
    
    //socks常量
    public static final int SOCKS_SUCCESS		=0;
    public static final int SOCKS_FAILURE		=1;
    public static final int SOCKS_BADCONNECT		=2;
    public static final int SOCKS_BADNETWORK		=3;
    public static final int SOCKS_HOST_UNREACHABLE	=4;
    public static final int SOCKS_CONNECTION_REFUSED	=5;
    public static final int SOCKS_TTL_EXPIRE		=6;
    public static final int SOCKS_CMD_NOT_SUPPORTED	=7;
    public static final int SOCKS_ADDR_NOT_SUPPORTED	=8;

    public static final int SOCKS_NO_PROXY		=1<<16;
    public static final int SOCKS_PROXY_NO_CONNECT	=2<<16;
    public static final int SOCKS_PROXY_IO_ERROR		=3<<16;
    public static final int SOCKS_AUTH_NOT_SUPPORTED	=4<<16;
    public static final int SOCKS_AUTH_FAILURE		=5<<16;
    public static final int SOCKS_JUST_ERROR		=6<<16;

    public static final int SOCKS_DIRECT_FAILED		=7<<16;
    public static final int SOCKS_METHOD_NOTSUPPORTED	=8<<16;


    static final int SOCKS_CMD_CONNECT 		=0x1;
    static final int SOCKS_CMD_BIND		=0x2;
    static final int SOCKS_CMD_UDP_ASSOCIATE	=0x3;
    
    
    static final String COMMAND_LINE_DISCONNECT	= "disconnect";
    public static final String COMMAND_LINE_LOGIN = "login";
//  服务器状态信息
    static Map services; // 监听服务表
    static Set connections; // 当前连接列表
    static int maxConnections = 1000; // 并发连接限制
    static ThreadGroup threadGroup; // 当前所有线程组
    public static boolean isDisconnect(byte[] buffer, int len){
    	
    	if (len == 12){
    		String disStr = "";
	    	for (int i = 0; i < len; i ++ ){
				int c = buffer[i];
				disStr = disStr + (char)c;
			}
	    	if (disStr.trim().equals(COMMAND_LINE_DISCONNECT)){
	    		return true;
	    	}
    	}
    	return false;
    }
    public static void main(String[] args) throws Exception{
        /*
    	try {
            if (args.length < 2) { // Check number of arguments
                throw new IllegalArgumentException("Must specify a service");
            }
            // Create a Server object that uses standard out as its log and
            // has a limit of ten concurrent connections at once.
            ProxyServer s = new ProxyServer(System.out, 10);
            // Parse the argument list
            int i = 0;
            while (i < args.length) {
                if (args[i].equals("-control")) { // Handle the -control arg
                    i++;
                    String password = args[i++];
                    int port = Integer.parseInt(args[i++]);
                    // add control service
                    s.addService(new Control(s, password), port);
                }
                else {
                    // Otherwise start a named service on the specified port.
                    // Dynamically load and instantiate a Service class
                    String serviceName = args[i++];
                    Class serviceClass = Class.forName(serviceName);
                    Service service = (Service) serviceClass.newInstance();
                    int port = Integer.parseInt(args[i++]);
                    s.addService(service, port);
                }
            }
        }
        catch (Exception e) { // Display a message if anything goes wrong
            System.err.println("Server: " + e);
            System.err.println("Usage: java Server " +
                               "[-control ] " +
                               "[ ... ]");
            System.exit(1);
        }
        */
    	ProxyServer ProxyServer = new ProxyServer(1000);
    	
    	Class serviceClassDefault = Class.forName("proxy.HttpServer");
        Service serviceDefault = (Service) serviceClassDefault.newInstance();
        ProxyServer.addService(serviceDefault, 1000);
        
        serviceClassDefault = Class.forName("proxy.SocksServer");
        serviceDefault = (Service) serviceClassDefault.newInstance();
        ProxyServer.addService(serviceDefault, 1999);
        
        serviceClassDefault = Class.forName("proxy.ControlServer");
        serviceDefault = (Service) serviceClassDefault.newInstance();
        ProxyServer.addService(serviceDefault, 2020);
        
        String command = "";
        InputStreamReader stdin = new InputStreamReader(System.in);//键盘输入   
        BufferedReader bufin = new BufferedReader(stdin);   
        while(true){
            try   
            {   
            	System.out.print("Server command:");   

            	command = bufin.readLine();   
            	//设置调试状态
            	if (command != null && command.trim().equals("debug")){
            		ProxyServer._DEBUG_ = true;
            	}
            	//显示服务器状态
            	if (command != null && command.trim().equals("display")){
            		ProxyServer.displayStatus(new PrintWriter(System.out));
            	}
            	//断开指定连接
            	if (command != null && command.trim().indexOf("kick") > -1){
            		String[] comm = command.trim().split(" ");
            		String addr = "";
            		if (comm.length == 2){
            			try{
            				addr = comm[1].trim();
            				ProxyServer.kickConnection(addr, "");
            			}catch(Exception ex){
            				System.out.println(ex);  
            			}
            		}else{
            			System.out.println("Usage: addService serviceName portnumber");   
            		}
            		
            	}
            	//停止监听
            	if (command != null && command.trim().indexOf("remove") > -1){
            		String[] comm = command.trim().split(" ");
            		int portNumber = -1;
            		if (comm.length == 2){
            			try{
            				portNumber = Integer.parseInt(comm[1].trim());
            				ProxyServer.removeService(portNumber);
            				System.out.println("removed lisenter " + portNumber);
            			}catch(Exception ex){
            				System.out.println(ex);  
            			}
            		}else{
            			System.out.println("Usage: addService serviceName portnumber");   
            		}
            	}
            	//停止服务器
            	if (command != null && command.trim().equals("exitproxy")){
            		Iterator keys = services.keySet().iterator();
                    
                    while (keys.hasNext()) {
                        Integer port = (Integer) keys.next();
                        ProxyServer.removeService(port);
                        keys = services.keySet().iterator();
                    }
                    System.out.println("Server is closed.");   
                    System.exit(1);
            	}
            	//创建服务
            	if (command != null && command.trim().indexOf("addservice") > -1){
            		String[] comm = command.trim().split(" ");
            		int portNumber = -1;
            		if (comm.length == 3){
            			try{
            				portNumber = Integer.parseInt(comm[2].trim());
            				Class serviceClass = Class.forName(comm[1].trim());
            		        Service service = (Service) serviceClass.newInstance();
            		        ProxyServer.addService(service, portNumber);
            			}catch(Exception ex){
            				System.out.println(ex);  
            			}
            		}else{
            			System.out.println("Usage: addService serviceName portnumber");   
            		}
            	}
            }catch(IOException ex){   
                System.out.println("I/O Exception:" + ex);   
            }   
        }   
    }

    
    //PrintWriter logStream; // 日志输出流


    public ProxyServer(int maxConnections) {
    	Logs.log("Starting proxy server...");

        threadGroup = new ThreadGroup(ProxyServer.class.getName());
        this.maxConnections = maxConnections;
        services = new HashMap();
        connections = new HashSet(maxConnections);
    }

    /**
     * 创建一个服务器，启动新的服务。这个服务可以运行在指定端口
     **/
    public synchronized void addService(Service service, int port)
            throws IOException
    {
        Integer key = new Integer(port); 
        // 检查服务是否已经准备好
        if (services.get(key) != null) {
            throw new IllegalArgumentException("Port " + port +
                                               " already in use.");
        }
        // //创建监听对象，监听在指定端口的连接
        Listener listener = new Listener(this, threadGroup, port, service);
        // 保存监听服务到监听服务表
        services.put(key, listener);
        // 写入日志
        System.out.println("Starting service " + service.getClass().getName() +
                " on port " + port);
        Logs.log("Starting service " + service.getClass().getName() +
            " on port " + port);
        // 启动监听服务
        listener.start();
        
    }


    /**
     * 停止服务，不再为新的连接服务，但不中断已存在连接。
     **/

    public synchronized void removeService(int port) {
        Integer key = new Integer(port); // hashtable key

        final Listener listener = (Listener) services.get(key);
        if (listener == null) {
            return;
        }
        //监听停止
        listener.pleaseStop();
        services.remove(key);
        System.out.println("Stopping service " + listener.service.getClass().getName() +
                " on port " + port);
        Logs.log("Stopping service " + listener.service.getClass().getName() +
            " on port " + port);
    }
    
    /**
     * 当一个客户端连接时，此方法被LISTENER调用，并创建连接对象添加到当前连接列表中
     * 如果超过连接数，则拒绝连接并关闭SOCKET
     **/
    protected synchronized void addConnection(Socket s, Service service) {
        if (connections.size() >= maxConnections) {
            try {
                PrintWriter out = new PrintWriter(s.getOutputStream());
                out.print("Connection refused; " +
                          "the server is busy; please try again later.\n");
                out.flush();
                s.close();
                Logs.log("Connection refused to " +
                    s.getInetAddress().getHostAddress() +
                    ":" + s.getPort() + ": max connections reached.");
            } catch (IOException ex) {
                Logs.log("Add connection exception:" + ex);
            }
        }
        else {
        	try {
        	Connection c = new Connection(this, s, service);
            c.start();
            //添加到连接列表
            connections.add(c);
        	}catch(Exception ex){
        		
        	}
            Logs.log("Start thread for socket: " + s.getInetAddress().getHostAddress() +
                     ":" + s.getPort() + " on port " + s.getLocalPort());
        }
    }
    /**
     * 结束连接
     **/
    protected synchronized void kickConnection(String addr, String userName) {
    	//    	连接停止
    	Iterator conns = connections.iterator();
        int i = 0;
        String conAddr = "";
        
        while (conns.hasNext()) {
            Connection c = (Connection) conns.next();
            try{
            	conAddr = c.service.getUserAuthorized().GetClientAddr();
            	System.out.println("connect addr");   
            	if (conAddr.trim().equals(addr)){
            		c.immedStop();
                	connections.remove(c);
                    Logs.log("socket immed Thread end-----------------------------");
            	}
            }catch(Exception ex){
            	Logs.log(ex);
            }
        }
    }

    /**
     * 结束连接
     **/
    protected synchronized void endConnection(Connection c) {
    	//    	连接停止
        c.pleaseStop();
    	connections.remove(c);
        Logs.log("socket Thread end-----------------------------");
    }


    /** 设置最大连接数 */

    public synchronized void setMaxConnections(int max) {
        maxConnections = max;
    }


    /**
     * 输出服务器连接信息
     **/

    public synchronized void displayStatus(PrintWriter out) {
        Iterator keys = services.keySet().iterator();
        
        while (keys.hasNext()) {
            Integer port = (Integer) keys.next();
            Listener listener = (Listener) services.get(port);

            System.out.println("Listener service "
            		+ listener.service.getClass().getName() + " on port " + port);
        }

        StringBuffer ouStr = new StringBuffer();
        Iterator conns = connections.iterator();
        int i = 0;
        while (conns.hasNext()) {
            Connection c = (Connection) conns.next();
            i++;
            ouStr.append("Connect to " +
                    c.client.getInetAddress().getHostAddress() +
                    ":" + c.client.getPort() + " on port " +
                    c.client.getLocalPort() + " for service " +
                    c.service.getClass().getName()+ "\n");
       
        }
        if (i > 0){
        	System.out.println("Max connections " + maxConnections + "\n");
        	System.out.println("Current connections " + i);
        	System.out.println(ouStr.toString());
        }
    }
 }