gatewaymonitor.java

java网关监控实例,解释了线程的应用方法

package com.cxt.platform.gateway;
import java.io.*;
import java.net.*;
import java.util.*;
import	java.sql.Time;

/**
 * Title:
 * Description:
 *		网关的监控,用于接受请求,返回相应的数据;如：线程监控、参数设定等
 * Copyright:    Copyright (c) 2001
 * Company:
 * @author
 * @version 1.0
 */

public class GatewayMonitor extends Thread
{
	private int 	m_port;	
	private boolean	m_run;
	private GatewayThreadGroup	m_gwthreadgroup;
	private	Gateway m_gateway;
	final int BUFFER_SIZE=10240;
	final int START=10;
	final int STOP=20;
	final int GET=30;
	final int SET=40;
	private	byte[] m_outbuf=new byte[BUFFER_SIZE];
	private byte[] m_inbuf=new byte[BUFFER_SIZE];
	
	public GatewayMonitor(GatewayThreadGroup tg,int port,Gateway gw)
	{
		m_port=port;
		m_run=true;
		m_gwthreadgroup=tg;
		m_gateway=gw;
	}

	//线程方法，子线程的入口
	public void run()
	{
		ServerSocket serv=null;
		
		try
		{
			serv=new ServerSocket(m_port);
		}
		catch (Exception e)
		{
			System.err.println("Can't make monitor server!");
			e.printStackTrace();
			System.exit(10);
		}
		 
		while(m_run)
		{
			Socket client=null;
			OutputStream clientSend=null;
			InputStream	clientReceive=null;
			int	out_len=0;
			int	pos=0;
			int	outpos=4;	//预留报文长度的两个字节和返回值的两个字节，共四个
			int iRetval=0;
			int	ret=0;
			int	len=0;
			
			try
			{
				client=serv.accept();
			}
			catch (Exception e)
			{
				System.err.println("Error when accept");
				e.printStackTrace();
				continue;
			}
			
			try 
			{
				clientSend = client.getOutputStream();
				clientReceive = client.getInputStream();
			
				//设定超时时间为5秒（5000毫秒）
				client.setSoTimeout(5000);
				
				//取整个报文的头两个字节，以便得到报文的长度
				ret=clientReceive.read(m_inbuf,0,2);
				if( ret<2 )	//取报文头两个字节失败！
				{
					System.err.println("Error when read length");
					throw new Exception("读报文出错！return code="+ret);
				}
				pos+=ret;
				//根据头两个字节，计算长度
				len=bytes2short(m_inbuf,0);
				if( len>BUFFER_SIZE-pos )	//长度超过了缓冲区的大小
				{
					System.err.println("Error that length exceed buffer size");
					throw new Exception("数据报文太长，超过缓冲区大小！length="+len);
				}
				//接收指定长度的数据
				ret=clientReceive.read(m_inbuf,pos,len);
				if( ret<len )
				{
					System.err.println("Error that length exceed buffer size");
					throw new Exception("读不到足够的数据，请查看报文！length="+ret);
				}
				
				switch(m_inbuf[pos])				//命令域的第一个字节代表命令类别
				{
					case	START:					//启动
					{
						switch(m_inbuf[pos+1])		//命令域的第二个字节代表具体命令
						{
							case	1:				//起动网关
								m_run=true;
								break;
							default	:				//无效的启动命令
								break;
						}
						break;
					}
					case	STOP:					//停止
					{
						switch(m_inbuf[pos+1])	
						{
							case	1:				//停止网关
								m_run=false;
								m_gateway.stop();
								//冒充client向网关发起连接，以解除网关ServerSocket.accept方法的堵塞
								/*try
								{
									Socket clientCheat = new Socket("localhost",m_gateway.getServerPort());
									//下列用法JDK1.4以上才支持
									//Socket clientCheat = new Socket();
									//clientCheat.connect(new InetSocketAddress("localhost",m_gateway.getServerPort()),5000);
									//
									clientCheat.close();
								}
								catch (Exception excp)
								{
									//无需处理
								}*/
								//
								break;
							default	:				//无效的停止命令
								break;
						}
						break;
					}
					case	GET:					//查询
					{
						switch(m_inbuf[pos+1])
						{
							case	1:		//线程监控
								
								GatewayHandler[] thread_list=new GatewayHandler[m_gwthreadgroup.activeCount()];
								ret=m_gwthreadgroup.enumerate(thread_list);
								outpos+=addThreadHeader(outpos);
								for( int i=0;i<ret;i++ )
								{
									outpos+=addThreadList(thread_list[i],outpos);
								}
									
								break;
							default	:		//无效的查询命令
								break;
						}
						break;
					}
					case	SET:			//设定
					{
						switch(m_inbuf[pos+1])
						{
							case	1:		//日志级别
								len=bytes2short(m_inbuf,pos+2);
								int logLevel=byteChar2Int(m_inbuf,pos+4,len);
								m_gateway.setLogLevel(logLevel);
								break;
							case	2:		//重新装载配置文件
								break;
							default	:		//无效的设置命令
								break;
						}
						break;
					}
					default	:				//无效的命令类型
						break;
				}
				
				short2bytes(m_outbuf,2,iRetval);	//返回值
				short2bytes(m_outbuf,0,outpos-2);	//返回报文的总长度
				clientSend.write(m_outbuf,0,outpos);
				clientSend.close();
				clientReceive.close();
				client.close();
			}
			catch (Exception e)
			{
				System.err.println("Error :"+e.toString());
				e.printStackTrace();
				try
				{
					clientSend.close();
					clientReceive.close();
					client.close();
				}
				catch (Exception e1)
				{
					e1.printStackTrace();
					System.err.println("Error when close socket,so how should i do???!!!");
				}
				continue;
			}
		}// end while
		
		try
		{
			serv.close();
		}
		catch (Exception e)
		{
			System.err.println("Error when close socket server,so how should i do???!!!");
			e.printStackTrace();
		}

		   
	}//end of run
	
	//从byte数组中的指定偏移取出相应的整形数据，高位在前，两个字节
	private int bytes2short(byte[] src,int pos)	
	{
		int	sh;

		sh=src[pos]<0 ? (256+src[pos])*256 : src[pos]*256;
		sh+=src[pos+1]<0 ? 256+src[pos+1] : src[pos+1];
		
		return	sh;
	}
	
	//将整形数据存储到指定的byte数组的指定偏移，高位在前，两个字节
	private void short2bytes(byte[] src,int pos,int value)	
	{
		src[pos]=(byte)(value/256);
		src[pos+1]=(byte)(value%256);
	}
	
	//将字符串型byte数组转成整形
	private int byteChar2Int(byte[] src,int pos,int len)
	{
		Integer i=new Integer(new String(src,pos,len));
		return i.intValue();
	}
	
	private int addThreadHeader(int	offset)
	{
		int	pos=0;
		byte[] byteArray;
		
///////////////////////////////////网关的静态属性
		//从网关中取出网关ID，并打到输出缓冲区；
		byteArray=m_gateway.getGatewayNO().getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关名，并打到输出缓冲区；
		byteArray=m_gateway.getGatewayName().getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关状态，并打到输出缓冲区；
		byteArray=String.valueOf(m_gateway.getGatewayStatus()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关接入服务类名，并打到输出缓冲区；
		byteArray=m_gateway.getReceiveServiceClassName().getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关监听端口，并打到输出缓冲区；
		byteArray=String.valueOf(m_gateway.getServerPort()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关最大允许连接数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gateway.getMaxClientCount()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关最大线程数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gateway.getMaxThreadCount()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关日志级别，并打到输出缓冲区；
		byteArray=String.valueOf(m_gateway.getLogLevel()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从网关中取出网关的起动时间，并打到输出缓冲区；
		byteArray=String.valueOf(new Time(m_gateway.getStartTime())).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		
/////////////////////////////////////////////////////网关线程的动态属性		
		//从线程组中取出接入的交易数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gwthreadgroup.getEntries()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程组中取出总服务的交易数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gwthreadgroup.getTotal()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程组中取出成功的交易数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gwthreadgroup.getSuccess()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程组中取出失败的交易数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gwthreadgroup.getFail()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程组中取出当前正活动的线程数，并打到输出缓冲区；
		byteArray=String.valueOf(m_gwthreadgroup.activeCount()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		/**********************************************
		/////////////////////////////////////////////////////////
		for(int i=0;i<pos;i++)
		{
			System.out.print(m_outbuf[offset+i]+" ");
		}
		System.out.println();
		/////////////////////////////////////////////////////////
		**********************************************/

		return	pos;
	}
	
	private int addThreadList(GatewayHandler thread_list,int offset)
	{
		int	pos=0;
		byte[] byteArray;
		
		//System.out.println("here:"+offset);
		//从线程对象中取出线程的名字，并打到输出缓冲区；
		byteArray=thread_list.getName().getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程对象中取出线程正服务的交易码，并打到输出缓冲区；
		short2bytes(m_outbuf,offset+pos,thread_list.getTrcd().length);
		pos+=2;
		System.arraycopy(thread_list.getTrcd(),0,m_outbuf,offset+pos,thread_list.getTrcd().length);
		pos+=thread_list.getTrcd().length;
		//从线程对象中取出线程正服务的交易的状态，并打到输出缓冲区；
		byteArray=String.valueOf(thread_list.getStatus()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		
		//从线程对象中取出线程正服务的交易的起始时间，并打到输出缓冲区；
		
		byteArray=String.valueOf(new Time(thread_list.getStartTm())).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		//从线程对象中取出线程正服务的交易的服务时间，并打到输出缓冲区；
		
		byteArray=String.valueOf(System.currentTimeMillis()-thread_list.getStartTm()).getBytes();
		short2bytes(m_outbuf,offset+pos,byteArray.length);
		pos+=2;
		System.arraycopy(byteArray,0,m_outbuf,offset+pos,byteArray.length);
		pos+=byteArray.length;
		
		/**********************************************
		/////////////////////////////////////////////////////////
		for(int i=0;i<pos;i++)
		{
			System.out.print(m_outbuf[offset+i]+" ");
		}
		System.out.println();
		/////////////////////////////////////////////////////////
		**********************************************/
		
		return	pos;
	}

}