serverthread.java

实现多人即时聊天

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

import javax.swing.Timer;
import java.awt.event.ActionListener;
import java.awt.event.ActionEvent;

class ServerThread{
	
	private Server server;
	private SwitchThread switchthread;
	private UserInfo yourself;
	private Socket Csocket;
	//流
	private ObjectOutputStream StreamOut;
	private ObjectInputStream StreamIn;

	private boolean invoked=false;
	private boolean shutdown=false;

	private LinkedList ONingQueue;
	private LinkedList OFFingQueue;
	private LinkedList RealTimeMsgQueue;
	private LinkedList NewPubMsgQueue;
	

	private Timer OutPutTimer;//在OutputThread中创建

	public ServerThread(){}
	
	public ServerThread(Server server,
			    SwitchThread switchthread,
			    UserInfo yourself,
			    Socket Csocket,
			    ObjectOutputStream StreamOut,
			    ObjectInputStream StreamIn){

			this.server=server;
			this.switchthread=switchthread;
			this.yourself=yourself;
			this.Csocket=Csocket;
			this.StreamOut=StreamOut;
			this.StreamIn=StreamIn;

			ONingQueue=new LinkedList();
			OFFingQueue=new LinkedList();
			RealTimeMsgQueue=new LinkedList();
			NewPubMsgQueue=new LinkedList();
			
			new Thread(new InputThread()).start();
			new Thread(new OutputThread()).start();

	}//public ServerThread()
	
public void IOF(Timer timer){

System.out.println("In IOF()");

	if(invoked) return;

	invoked=true;
	timer.stop(); 

	server.OffUserReceiver(yourself.getUser());
	terminate();

//server.OffUserReceiver(yourself)后terminate()
//确保先将CSthread中条目移除后再关闭OutputThread退出

}//public void IOF(Timer timer)

public void terminate(){

System.out.println("In terminate()");

	shutdown=true;

System.out.print("In terminate() shutdown=");
System.out.println(shutdown);

}//public void terminate()


/********************下面四个方法由SwitchThread调用*****************************/

public void OnUserReceiver(User Onuser){
	synchronized(ONingQueue){ONingQueue.add(Onuser);}
}//public void OnUserReceiver(User Onuser)

public void OffUserReceiver(User Offuser){
	synchronized(OFFingQueue){OFFingQueue.add(Offuser);}
}//public void OffUserReceiver(User Offuser)

public void RealTimeMsgReceiver(Message msg){

System.out.println("in ServerThread RealTimeMsgReceiver\n即时信息正在交付对应ServerThread");

	synchronized(RealTimeMsgQueue){RealTimeMsgQueue.add(msg);}

System.out.println("in ServerThread RealTimeMsgReceiver\n即时信息已经成功添加到\nServerThread RealTimeMsgQueue");

}//public void RealTimeMsgReceiver(Message)

public void PubMsgReceiver(PublicMessage pubmsg){
	synchronized(NewPubMsgQueue){NewPubMsgQueue.add(pubmsg);}
}//public void PubMsgReceiver(PublicMessage pubmsg)

/********************上面四个方法由SwitchThread调用*****************************/	



public static void main(String args[]){
	new ServerThread();
}//main()
	

class InputThread implements Runnable{
	Message tempmsg;
 	User tempUser;
	char tempConChar;
public void run(){
System.out.println("in InputThread");
     try{
	
	while(!shutdown){
		tempConChar=StreamIn.readChar();
		switch(tempConChar){
			case 'F':{//即时信息

System.out.println("In InputThread case'F'--收到即时信息");

				tempmsg=(Message)StreamIn.readObject();

System.out.println("即时信息成功读取,并准备交给switchthread.RealTimeMsgReceiver()");

				switchthread.RealTimeMsgReceiver(tempmsg);
				break;
			}//case 'F':
			case 'M':{//留言
				tempmsg=(Message)StreamIn.readObject();
				server.MsgReceiver(tempmsg);
				break;
			}//case 'M':
			case 'N':{//用户修改信息

				tempUser=(User)StreamIn.readObject();

/******************************以下测试*************************************************************************/
System.out.println("in InputThread，收到修改后的用户信息 tempUser：");
System.out.println("in InputThread 用户名："+tempUser.getUserName());
System.out.println("in InputThread 职位："+tempUser.getJob());
System.out.println("in InputThread 密码："+tempUser.getPassword());
System.out.println("in InputThread 邮箱："+tempUser.getEMail());
System.out.println("in InputThread 电话："+tempUser.getTel());

/******************************以上测试*************************************************************************/

				server.UserInfoModifyReceiver(tempUser);
				break;
			}//case 'N':
			case 'D':{//当前用户下线
				IOF(OutPutTimer);
				break;
			}//case 'D':
		}//switch(tempConChar)
	}//while(!shutdown)
     }catch(Exception e){

System.out.println("in InputThread run()异常");
System.out.println(e);
System.out.println(((OptionalDataException)e).eof);
System.out.println(((OptionalDataException)e).length);
System.out.println("准备调用IOF(Timer)");

	IOF(OutPutTimer);
     }//catch(Exception e)


}//public void run()	
}//class InputThread implements Runnable

class OutputThread implements Runnable{
		//User tempUser;
		//Message tempmsg;
		//PublicMessage tempPubMsg;
	public void run(){

System.out.println("in OutputThread");

	OutPutTimer=new Timer(2000,new ActionListener(){
			User tempUser;
			Message tempmsg;
			PublicMessage tempPubMsg;
		public void actionPerformed(ActionEvent evt){
//System.out.println("in OutputThread OutPutTimer");
		    try{

			if(shutdown){
System.out.println("in OutputThread 准备发送结束控制字符'G'");
				StreamOut.writeChar('G');
				StreamOut.flush();
				return;	
			}

			synchronized(ONingQueue){
				for(;ONingQueue.size()>0;){
					tempUser=(User)ONingQueue.poll();
					StreamOut.writeChar('U');
					StreamOut.writeObject(tempUser);
					StreamOut.flush();
				}
			}//synchronized(ONingQueue)


			synchronized(OFFingQueue){
				for(;OFFingQueue.size()>0;){
					tempUser=(User)OFFingQueue.poll();
					StreamOut.writeChar('D');
					StreamOut.writeObject(tempUser);
					StreamOut.flush();
				}
			}//synchronized(OFFingQueue)


			synchronized(RealTimeMsgQueue){
				for(;RealTimeMsgQueue.size()>0;){

System.out.println("in OutPutTimer正准备向客户端发送一个即时信息");

					tempmsg=(Message)RealTimeMsgQueue.poll();
					StreamOut.writeChar('F');
					StreamOut.writeObject(tempmsg);
					StreamOut.flush();

System.out.println("in OutPutTimer已经成功向客户端发送了一个即时信息");

				}
			}//synchronized(RealTimeMsgQueue)


			synchronized(NewPubMsgQueue){
				for(;NewPubMsgQueue.size()>0;){
					tempPubMsg=(PublicMessage)NewPubMsgQueue.poll();
					StreamOut.writeChar('C');
					StreamOut.writeObject(tempPubMsg);
					StreamOut.flush();
				}			
			}//synchronized(NewPubMsgQueue)	
		    }catch(Exception e){

System.out.println("in OutPutTimer异常");
System.out.println(e);
System.out.println("准备调用IOF(Timer)");

			IOF(OutPutTimer);
          	    }//catch(Exception e)




		}//public void actionPerformed(ActionEvent evt)
	});
	OutPutTimer.start();


	    //try{
	    //   	while(!shutdown){
				
	//	}//while(!shutdown)
			
			

            //}catch(Exception e){
	//	IOF(OutPutTimer);
           // }//catch(Exception e)
	}//public void run()
}//class OutputThread implements Runnable

	
}//class ServerThread