pooledconnectionhandle.java

在机器上开一个端口,代理客户端的数据库请求,用在资源有限的终端访问大型数据库上使用,使用jdbc连接orcle, 客户端使用标准的socket即可

package athere.tp;

import java.io.*;
import java.net.*;
import java.util.*;
import com.behd.db.*;
import java.sql.*;
import java.util.Hashtable;
import java.util.Date;

//连接处理器
public class PooledConnectionHandle implements Runnable {

	// 存放所有请求的socket,使用链表模拟队列,注意1.5的新特性 List<Object>
	private static List pool = new LinkedList();

	// 当Update语句类的, 服务号和对应的mdb,没有提交的连接需要保留
	private static Hashtable Connlist = new Hashtable();

	private Socket socket = null; // 当前处理的客户端连接

	private String ClientServiceID = null; // 客户端请求的服务号

	private Connection conn = null; // 数据库连接

	private Statement statement = null; // Statemen

	private java.text.SimpleDateFormat dtFmt = new java.text.SimpleDateFormat(
			"yyyy-MM-dd HH:mm:ss");

	private java.text.SimpleDateFormat dFmt = new java.text.SimpleDateFormat(
			"yyyy-MM-dd");

	public PooledConnectionHandle() {
	}

	// 将请求的socket链入请求队列
	public static void processRequest(Socket requestSocket) {
		synchronized (pool) {
			pool.add(pool.size(), requestSocket);
			pool.notifyAll();
		}
	}

	// 主线程, 注意, 该种模式下注定了处理一个命令就需要连接一次服务器, 否则服务器的线程处于阻塞等待的状态
	public void run() {
		while (true) {
			synchronized (pool) {
				while (pool.isEmpty()) {
					try {
						pool.wait();
					} catch (Exception e) {
						return;
					}
				}
				// 当不是本线程对应的服务号,也需要阻塞
				socket = (Socket) pool.remove(0);
			}
			handleConnection();
		}
	}

	// 处理命令
	public void handleConnection() {
		String cmd = null;
		String readString = null;

		// 1.读客户端传来的命令
		try {
			BufferedReader inputReader = new BufferedReader(
					new InputStreamReader(socket.getInputStream()));
			readString = inputReader.readLine();
			System.out.println("handleConnection-客户端发送执行请求:" + readString);
		} catch (Exception e) {
			System.out.println("handleConnection-读取客户端发送的请求执行命令错误:"
					+ e.getMessage());
			return;
		}

		// 2.具体的命令具体处理
		cmd = readString.substring(0, 4);
		char auto_commit = '0'; // 是否提交，客户端传过来的
		if (cmd.equals("1001")) {
			int curr = 0; // 客户端请求的页号
			int rowsPrePage = 0; // 客户端请求的页包含记录数
			ClientServiceID = readString.substring(4, 8);
			curr = Integer.parseInt(readString.substring(9, 13), 16);
			rowsPrePage = Integer.parseInt(readString.substring(13, 17), 16);
			// 解释为查询语句进行处理
			execQuery(readString.substring(17), curr, rowsPrePage);
		} else if (cmd.equals("1002")) {
			ClientServiceID = readString.substring(4, 8);
			auto_commit = readString.substring(8, 9).charAt(0);
			System.out.println("客户端请求执行查询update:" + readString);
			// 解释为数据操作指令进行处理
			execUpdate(readString.substring(9), auto_commit);
		} else if (cmd.equals("1003")) {
			// 解释为调用存储过程程序包指令进行处理
			System.out.println("客户端请求执行查询procedure:" + readString);
		} else if (cmd.equals("1004")) {
			// 客户端要求断开连接
			try {

				try{
					socket.getInputStream().close();  // 关闭Socket输出流
				}catch(Exception e){}
				
				try{
					socket.getOutputStream().close(); // 关闭Socket输入流
				}catch(Exception e){}
				try{
					socket.close(); 				  // 关闭Socket
				}catch(Exception e){}
			} catch (Exception e) {
				System.out.println("客户端请求关闭socket错误:" + readString);
			}
		} else {
			// 原来实现的传文件操作
			sendFile(readString);
		}
	}

	private int writeString(OutputStream s, String str) throws IOException {
		/*
		 * 向socket写入信息,但是不写入信息的长度,为空和零长度字符串将不操作
		 */
		if (str == null) {
			return 0;
		}
		if (str.equals("")) {
			return 0;
		}
		byte[] tmpbytes = str.getBytes();
		s.write(tmpbytes);
		return tmpbytes.length;
	}

	public int writeString4ByteLen(OutputStream s, String str)
			throws IOException {
		/*
		 * 向socket写入信息,但是不写入信息的长度,为空和零长度字符串将不操作
		 */
		if (str == null) {
			return 0;
		}
		if (str.equals("")) {
			return 0;
		}
		byte[] tmpbytes = str.getBytes();
		if (tmpbytes.length <= 65535) {
			s.write(int24char(tmpbytes.length).getBytes());
			s.write(tmpbytes);
			return tmpbytes.length;
		} else {
			s.write("FFFF".getBytes());
			s.write(tmpbytes, 0, 65535);
			return 65535;
		}
	}

	public long writeString8ByteLen(OutputStream s, String str)
			throws IOException {
		/*
		 * 向socket写入信息,但是不写入信息的长度,为空和零长度字符串将不操作
		 */
		if (str == null) {
			return 0;
		}
		if (str.equals("")) {
			return 0;
		}
		byte[] tmpbytes = str.getBytes();
		if (tmpbytes.length <= 4294967295L) {
			s.write(int28char(tmpbytes.length).getBytes());
			s.write(tmpbytes);
			return tmpbytes.length;
		} else {
			s.write("FFFFFFFF".getBytes());
			s.write(tmpbytes, 0, 2147483647);
			s.write(tmpbytes, 2147483647, 2147483647);
			return 4294967295L;
		}
	}

	/**
	 * 执行查询语句,语句的格式 '1001' + 'XXXX' + '0'+ 'XXXX' + 'XXXX' + 'select ......'
	 * 'XXXX' 第一个，表示4位的服务号 'XXXX' 第二个，表示页号 'XXXX' 第三个，表示每页的记录数
	 * 
	 * @param sql
	 *            要执行的sql语句
	 * @param curr
	 * @param rowsprepage
	 */
	public void execQuery(String sql, int curr, int rowsprepage) {
		try {
			OutputStream o = socket.getOutputStream();

			System.out.println("execQuery-客户端要求执行查询:" + sql);
			System.out.println("execQuery-客户端要求当前页号:" + curr);
			System.out.println("execQuery-客户端要求每页的记录数:" + rowsprepage);
			System.out.println("execQuery-服务器正在调用execQuery进行中...");

			ResultSet rs = null;
			conn = getConn(o, '1');
			if (conn == null) {
				return;
			}

			this.writeString(o, "0000"); // (String)上位机向下的执行结果4位
			this.writeString(o, ClientServiceID); // (String)服务编号4位
			String ls_mess = null;
			boolean lb_sqlcodeRtn = false;
			try {
				rs = executeQueryForResultSet(sql);
				// 正确执行返回执行结果
				rs.last();
				if (rs.getRow() == 0) {// (Integer)执行结果sqlcode =100
					this.writeString(o, int24char(100));
					lb_sqlcodeRtn = true;
					this.writeString8ByteLen(o, "查询无记录(data not found)!");
				} else {// (Integer)执行结果sqlcode =0
					this.writeString(o, int24char(0));
					lb_sqlcodeRtn = true;
					ResultSetToString(o, rs, curr, rowsprepage);
				}
			} catch (SQLException e) {
				// 出现异常,返回异常信息,检查数据库配置等即可
				if (lb_sqlcodeRtn) {
					o.write('X'); // 将出现在数据类型的位置上
				} else {
					// -1不太好表示,直接使用0xffff,即65535
					this.writeString(o, "FFFF");
				}
				ls_mess = "execQuery-出现异常behdException:" + e.getMessage();
				this.writeString8ByteLen(o, ls_mess);
				System.out.println(ls_mess);
				e.printStackTrace();
			} catch (Exception e) {
				// 出现异常
				if (lb_sqlcodeRtn) {
					o.write('X');
				} else {
					this.writeString(o, "FFFF");
				}
				ls_mess = "execQuery-出现异常Exception:" + e.getMessage();
				this.writeString8ByteLen(o, ls_mess);
				System.out.println(ls_mess);
				e.printStackTrace();
			}
			o.flush();
		} catch (Exception e) {
			// 进行通知客户端出现异常
			System.out
					.println("Exception occured when handiling a client connection."
							+ e.getMessage());
		} finally {
			// 因该进行关闭printWriter的操作吧
		}
	}

	private ResultSet executeQueryForResultSet(String s) {
		ResultSet rs = null;
		try {
			statement = conn.createStatement(1004, 1008);
		} catch (SQLException eConn) {
			return null;
		}
		try {
			rs = statement.executeQuery(s);
		} catch (SQLException sqlexception) {
			System.out.println(s + " executeQuery: "
					+ sqlexception.getMessage());
		}
		return rs;
	}

	public static String int28char(int intVlaue) {
		/* 将intger 转化为0xFFFFFFFF八字节的16进制数据 */
		String ret = Integer.toHexString(intVlaue);
		int li_len = ret.length();
		if (li_len == 7) {
			return "0" + ret;
		} else if (li_len == 6) {
			return "00" + ret;
		} else if (li_len == 5) {
			return "000" + ret;
		} else if (li_len == 4) {
			return "0000" + ret;
		} else if (li_len == 3) {
			return "00000" + ret;
		} else if (li_len == 2) {
			return "000000" + ret;
		} else if (li_len == 1) {
			return "0000000" + ret;
		}
		return ret;
	}

	public static String int24char(int intVlaue) {
		/* 将intger 转化为0xFFFFFFFF八字节的16进制数据 */
		String ret = Integer.toHexString(intVlaue);
		int li_len = ret.length();
		if (li_len == 3) {
			return "0" + ret;
		} else if (li_len == 2) {
			return "00" + ret;