import java.io.Serializable;

import java.text.SimpleDateFormat;

import java.util.Date;/**

* 维持连接的消息对象。

*

* 创建时间：2010-7-18 上午12:22:09

* @author HouLei

* @since 1.0*/

public classKeepAlive implements Serializable{private static final long serialVersionUID = -2813120366138988480L;/*覆盖该方法，仅用于测试使用。

* @see java.lang.Object#toString()*/@OverridepublicString toString() {return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())+"\t维持连接包";

}

}

import java.io.IOException;

import java.io.InputStream;

import java.io.ObjectInputStream;

import java.io.ObjectOutputStream;

import java.net.Socket;

import java.net.UnknownHostException;

import java.util.concurrent.ConcurrentHashMap;/**

* C/S架构的客户端对象，持有该对象，可以随时向服务端发送消息。

*

* 创建时间：2010-7-18 上午12:17:25

* @author HouLei

* @since 1.0*/

public classClient {/**

* 处理服务端发回的对象，可实现该接口。*/

public static interfaceObjectAction{voiddoAction(Object obj,Client client);

}public static final classDefaultObjectAction implements ObjectAction{public voiddoAction(Object obj,Client client) {

System.out.println("处理：\t"+obj.toString());

}

}public static voidmain(String[] args) throws UnknownHostException, IOException {

String serverIp= "127.0.0.1";int port = 65432;

Client client= newClient(serverIp,port);

client.start();

}privateString serverIp;private intport;privateSocket socket;private boolean running=false;private longlastSendTime;private ConcurrentHashMap actionMapping = new ConcurrentHashMap();public Client(String serverIp, intport) {this.serverIp=serverIp;this.port=port;

}public voidstart() throws UnknownHostException, IOException {if(running)return;

socket= newSocket(serverIp,port);

System.out.println("本地端口："+socket.getLocalPort());

lastSendTime=System.currentTimeMillis();

running=true;new Thread(newKeepAliveWatchDog()).start();new Thread(newReceiveWatchDog()).start();

}public voidstop(){if(running)running=false;

}/**

* 添加接收对象的处理对象。

* @param cls 待处理的对象，其所属的类。

* @param action 处理过程对象。*/

public void addActionMap(Classcls,ObjectAction action){

actionMapping.put(cls, action);

}public voidsendObject(Object obj) throws IOException {

ObjectOutputStream oos= newObjectOutputStream(socket.getOutputStream());

oos.writeObject(obj);

System.out.println("发送：\t"+obj);

oos.flush();

}classKeepAliveWatchDog implements Runnable{long checkDelay = 10;long keepAliveDelay = 2000;public voidrun() {while(running){if(System.currentTimeMillis()-lastSendTime>keepAliveDelay){try{

Client.this.sendObject(newKeepAlive());

}catch(IOException e) {

e.printStackTrace();

Client.this.stop();

}

lastSendTime=System.currentTimeMillis();

}else{try{

Thread.sleep(checkDelay);

}catch(InterruptedException e) {

e.printStackTrace();

Client.this.stop();

}

}

}

}

}classReceiveWatchDog implements Runnable{public voidrun() {while(running){try{

InputStreamin =socket.getInputStream();if(in.available()>0){

ObjectInputStream ois= new ObjectInputStream(in);

Object obj=ois.readObject();

System.out.println("接收：\t"+obj);

ObjectAction oa= actionMapping.get(obj.getClass());

oa= oa==null?newDefaultObjectAction():oa;

oa.doAction(obj, Client.this);

}else{

Thread.sleep(10);

}

}catch(Exception e) {

e.printStackTrace();

Client.this.stop();

}

}

}

}

}

import java.io.IOException;

import java.io.InputStream;

import java.io.ObjectInputStream;

import java.io.ObjectOutputStream;

import java.net.ServerSocket;

import java.net.Socket;

import java.util.concurrent.ConcurrentHashMap;/**

* C/S架构的服务端对象。

*

* 创建时间：2010-7-18 上午12:17:37

* @author HouLei

* @since 1.0*/

public classServer {/**

* 要处理客户端发来的对象，并返回一个对象，可实现该接口。*/

public interfaceObjectAction{

Object doAction(Object rev);

}public static final classDefaultObjectAction implements ObjectAction{publicObject doAction(Object rev) {

System.out.println("处理并返回："+rev);returnrev;

}

}public static voidmain(String[] args) {int port = 65432;

Server server= newServer(port);

server.start();

}private intport;private volatile boolean running=false;private long receiveTimeDelay=3000;private ConcurrentHashMap actionMapping = new ConcurrentHashMap();privateThread connWatchDog;public Server(intport) {this.port =port;

}public voidstart(){if(running)return;

running=true;

connWatchDog= new Thread(newConnWatchDog());

connWatchDog.start();

}

@SuppressWarnings("deprecation")public voidstop(){if(running)running=false;if(connWatchDog!=null)connWatchDog.stop();

}public void addActionMap(Classcls,ObjectAction action){

actionMapping.put(cls, action);

}classConnWatchDog implements Runnable{public voidrun(){try{

ServerSocket ss= new ServerSocket(port,5);while(running){

Socket s=ss.accept();new Thread(newSocketAction(s)).start();

}

}catch(IOException e) {

e.printStackTrace();

Server.this.stop();

}

}

}classSocketAction implements Runnable{

Socket s;

boolean run=true;long lastReceiveTime =System.currentTimeMillis();publicSocketAction(Socket s) {this.s =s;

}public voidrun() {while(running &&run){if(System.currentTimeMillis()-lastReceiveTime>receiveTimeDelay){

overThis();

}else{try{

InputStreamin =s.getInputStream();if(in.available()>0){

ObjectInputStream ois= new ObjectInputStream(in);

Object obj=ois.readObject();

lastReceiveTime=System.currentTimeMillis();

System.out.println("接收：\t"+obj);

ObjectAction oa= actionMapping.get(obj.getClass());

oa= oa==null?newDefaultObjectAction():oa;

Objectout =oa.doAction(obj);if(out!=null){

ObjectOutputStream oos= newObjectOutputStream(s.getOutputStream());

oos.writeObject(out);

oos.flush();

}

}else{

Thread.sleep(10);

}

}catch(Exception e) {

e.printStackTrace();

overThis();

}

}

}

}private voidoverThis() {if(run)run=false;if(s!=null){try{

s.close();

}catch(IOException e) {

e.printStackTrace();

}

}

System.out.println("关闭："+s.getRemoteSocketAddress());

}

}

}