分类： Java
2009-05-14 17:04:50
package com.sse.monitor.serv;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.BufferedOutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import com.sse.monitor.bean.Message;
import com.sse.monitor.bean.MessageHead;
import com.sse.monitor.bean.ResponseMessage;
import com.sse.monitor.form.ListenerInvoke;
import com.sse.monitor.form.MainForm;
import com.sse.monitor.util.SwingUtils;
/**
* Socket套接字工厂，对外接口是静态方法 SocketFactory.request(String, String, String, int)
* Copyright: Copyright (c) 2008
* Company: conserv
* @author cuishen
* @version 1.3
*/
public class SocketFactory {
private Socket socket = null;
private String targetIpAddress = null;
private int targetPort = 0;
private static SocketFactory sf = new SocketFactory();
public SocketFactory() {
}
/**
* 建立一条TCP/IP连接
* @param targetIpAddress String 目标ip地址
* @param targetPort String 目标端口
* @throws IOException
*/
private void connect(String targetIpAddress, int targetPort) throws IOException {
setTargetIpAddress(targetIpAddress);
setTargetPort(targetPort);
if(socket == null)
socket = new Socket(targetIpAddress, targetPort);
}
/**
* 这是对外接口。发送命令，接收反馈和接收message放两个线程，
* 发送命令并接收反馈是短连接，所以每次执行成功后，将销毁socket并终止线程，
* 接收message是长连接，所以可能会new出n个线程，建议对接收message的线程做缓存
* @param commandType String 命令类型
* @param commandContent String 命令内容
* @param targetIP String 目标ip
* @param targetPort int 目标端口
*/
public static void request(String commandType, String commandContent, String targetIP, int targetPort) {
if (commandType.equalsIgnoreCase(MessageFactory.SCAN_COMMAND)) {
sf.new GetMessageSocketThread(commandType, commandContent, targetIP, targetPort);
} else {
sf.new RequestSocketThread(commandType, commandContent, targetIP, targetPort);
}
}
/**
* 发送请求
* @param commandType String 命令类型
* @param commandContent String 命令内容
* @param targetIp String 目标ip
*/
private void sendRequest(String commandType, String commandContent, String targetIp) {
OutputStream os = null;
BufferedOutputStream bs = null;
try {
os = socket.getOutputStream();
bs = new BufferedOutputStream(os);
char[] message = MessageFactory.makeRequestMessage(targetIp, commandType, commandContent, MessageFactory.COMMAND_TRADE_CODE, MessageFactory.RIGHT_COMMAND, MessageFactory.MESSAGE_END_FLAG);
for (int i = 0; i < message.length; i++)
bs.write(new String(message).getBytes(), i, 1);
bs.flush();
SwingUtils.appendLog(MainForm.jTextArea, "发送请求：'" + commandType + "' '" + commandContent + "' '" + targetIp + "'", ReadConfig.commandStateShowLineCount);
} catch (IOException e) {
SwingUtils.appendLog(MainForm.jTextArea, "Error!!! 发送请求：'" + commandType + "' '" + commandContent + "' '" + targetIp + "'失败！！ " + e.getMessage(), ReadConfig.commandStateShowLineCount);
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
} finally {
}
}
/**
* 获得反馈
*
* @return 如果成功获得反馈，则返回true；否则返回false
*/
private boolean getResponse() {
InputStream is = null;
DataInputStream di = null;
boolean returnFlag = false;
try {
is = socket.getInputStream();
di = new DataInputStream(is);
byte[] temp = new byte[1];
int flag = 0;
ArrayList tempByteList = new ArrayList();
int i = 0;
while (flag != -1) {
i++;
flag = di.read(temp = new byte[1]);
if (flag != -1)
tempByteList.add(temp);
if (i == 38)
break;
}
if (i == 1) {
SwingUtils.Error("未收到response!!!");
return false;
}
MessageHead messageHead = MessageFactory.readHead(tempByteList);
SwingUtils.appendLog(MainForm.jTextArea, "收到 response", ReadConfig.commandStateShowLineCount);
tempByteList = new ArrayList();
i = 0;
while (flag != -1) {
i++;
flag = di.read(temp = new byte[1]);
if (flag != -1)
tempByteList.add(temp);
if (i == 26)
break;
}
byte[] length = new byte[4];
di.read(length);
int len = Integer.parseInt(new String(length, MessageFactory.DEFAULT_CHAR_SET).trim());
flag = 0;
for (int j = 0; j < (len + 37); j++) {
flag = di.read(temp = new byte[1]);
if (flag == -1)
break;
tempByteList.add(temp);
}
ResponseMessage rm = MessageFactory.readResponseMessage(tempByteList, len);
if (messageHead.getErrorCode().equals(MessageFactory.SUCCESS))
returnFlag = true;
else
SwingUtils.Error("errorCode: " + messageHead.getErrorCode() + "; content: " + rm.getCommandContent());
} catch (IOException e) {
e.printStackTrace();
} finally {
}
return returnFlag;
}
/**
* 分发消息的方法,将消息按进程名发送到对应的消息缓存
*
* 消息缓存ListenerInvoke.messageMap，key = machineName +
* '|' + programName + '|' + processId, value = messageList
* 存放messageMap里面的键名的List -- ListenerInvoke.messageMapKeyList
* 进程状态缓存ListenerInvoke.processStateMap, key = machineName + '|' +
* programName + '|' + processId, value = String
*
* @param message Message
*/
private void distributeMess(Message message) {
String machineName = message.getMachineName();
String programName = message.getProgramName();
String processId = message.getProcessId();
String content = message.getContent();
String key = machineName + '|' + programName + '|' + processId;
key = key.trim();
ArrayList messageList = null;
if (ListenerInvoke.messageMap.get(key) == null) {
synchronized (ListenerInvoke.messageMap) {
if (ListenerInvoke.messageMap.get(key) == null) {
messageList = new ArrayList();
messageList.add(content);
ListenerInvoke.messageMap.put(key, messageList);
}
}
} else {
messageList = (ArrayList) ListenerInvoke.messageMap.get(key);
synchronized (messageList) {
if (ListenerInvoke.messageMap.get(key) != null) {
messageList.add(content);
if (!ReadConfig.threadDeleteMessCacheOrFIFO
&& messageList.size() >= ReadConfig.messageCacheSizeLimit)
messageList.remove(0);
}
}
}
if (!ListenerInvoke.messageMapKeyList.contains(key)) {
synchronized (ListenerInvoke.messageMapKeyList) {
if (!ListenerInvoke.messageMapKeyList.contains(key))
ListenerInvoke.messageMapKeyList.add(key);
}
}
}
/**
* 接收message
* @return Message
*/
private boolean getMessage() {
InputStream is = null;
DataInputStream di = null;
Message message = null;
try {
if (this.socket == null) return false;
is = this.socket.getInputStream();
if (is == null) return false;
di = new DataInputStream(is);
byte[] temp = new byte[1];
int flag = 0;
ArrayList tempByteList = new ArrayList();
int i = 0;
while (flag != -1) {
i++;
flag = di.read(temp = new byte[1]);
if (flag != -1)
tempByteList.add(temp);
if (i == 38)
break;
}
if (i == 1) return false;
tempByteList = new ArrayList();
i = 0;
while (flag != -1) {
i++;
flag = di.read(temp = new byte[1]);
if (flag != -1)
tempByteList.add(temp);
if (i == 74)
break;
}
byte[] length = new byte[4];
di.read(length);
int len = Integer.parseInt(new String(length,
MessageFactory.DEFAULT_CHAR_SET).trim());
flag = 0;
for (int j = 0; j < len; j++) {
flag = di.read(temp = new byte[1]);
if (flag == -1)
break;
tempByteList.add(temp);
}
message = MessageFactory.readMessage(tempByteList, len);
SwingUtils.appendLog(MainForm.jTextArea, "收到新 Message",
ReadConfig.commandStateShowLineCount);
distributeMess(message);// 分发message
} catch (IOException e) {
e.printStackTrace();
} finally {
}
return true;
}
/**
* 负责发送请求接收反馈的内部线程类，每new一个RequestSocketThread线程，
* 就new一个socket，建立一条专属连接，成功接收反馈后将销毁socket，终止线程。
* 将发送请求，接收反馈放进内部线程处理，是为了防止套接字阻塞造成主线程挂死。
* @author cuishen
* @version 1.2
*/
class RequestSocketThread implements Runnable {
private SocketFactory socketFactory;
private String commandType = null;
private String commandContent = null;
private String targetIP = null;
Thread t;
public RequestSocketThread(String commandType, String commandContent, String targetIP, int targetPort) {
this.socketFactory = new SocketFactory();
try {
this.socketFactory.connect(ReadConfig.targetIpAddress, ReadConfig.targetPort);
} catch (UnknownHostException e) {
SwingUtils.Error("主机 IP 地址无法确定,无法建立连接！ targetIP=" + ReadConfig.targetIpAddress + ", targetPort=" + ReadConfig.targetPort);
e.printStackTrace();
} catch (IOException e) {
SwingUtils.Error("访问被拒绝，无法建立连接，请检查网络！ targetIP=" + ReadConfig.targetIpAddress + ", targetPort=" + ReadConfig.targetPort);
e.printStackTrace();
}
this.commandType = commandType;
this.commandContent = commandContent;
this.targetIP = targetIP;
t = new Thread(this);
t.start();
}
public void run() {
this.socketFactory.sendRequest(commandType, commandContent, targetIP);
this.socketFactory.getResponse();
stopThread();
}
public void stopThread() {
try {
this.commandType = null;
this.commandContent = null;
this.targetIP = null;
socketFactory.closeSocket();
socketFactory = null;
this.t.join(100);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
t = null;
}
}
}
/**
* 负责接收message的内部线程类，每new一个GetMessageSocketThread线程，
* 就new一个socket，建立一条专属TCP/IP连接，getMessage是长连接，所以建议
* 将该线程放入缓存方便管理
* @author cuishen
* @version 1.2
*/
class GetMessageSocketThread implements Runnable {
private SocketFactory socketFactory;
private String commandType = null;
private String commandContent = null;
private String targetIP = null;
Thread t;
private boolean flag = false;
private boolean ifGetResponse = true;
private boolean ifGetMessage = true;
private boolean ifSendRequest = true;
private boolean ifCycle = true;
public GetMessageSocketThread(String commandType, String commandContent, String targetIP, int targetPort) {
this.socketFactory = new SocketFactory();
try {
this.socketFactory.connect(ReadConfig.targetIpAddress, ReadConfig.targetPort);
} catch (UnknownHostException e) {
SwingUtils.Error("主机 IP 地址无法确定,无法建立连接！ targetIP="
+ ReadConfig.targetIpAddress + ", targetPort="
+ ReadConfig.targetPort);
e.printStackTrace();
} catch (IOException e) {
SwingUtils.Error("访问被拒绝，无法建立连接，请检查网络！ targetIP="
+ ReadConfig.targetIpAddress + ", targetPort="
+ ReadConfig.targetPort);
e.printStackTrace();
}
this.commandType = commandType;
this.commandContent = commandContent;
this.targetIP = targetIP;
t = new Thread(this);
t.start();
}
public void run() {
while (ifCycle) {
if (ifSendRequest) {
this.socketFactory.sendRequest(commandType, commandContent, targetIP);
ifSendRequest = false;
}
if (ifGetResponse) {
flag = socketFactory.getResponse();
ifGetResponse = false;
}
if (flag && ifGetMessage && socketFactory.socket != null) {
if (!socketFactory.getMessage()) {
try {
Thread.sleep(ReadConfig.getMessageThreadSleep);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
public void stopThread() {
try {
this.commandType = null;
this.commandContent = null;
this.targetIP = null;
ifGetMessage = false;
ifCycle = false;
socketFactory.closeSocket();
socketFactory = null;
this.t.join(100);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
t = null;
}
}
}
/**
* 关闭套接字
*/
private void closeSocket() {
try {
if (!socket.isClosed())
socket.close();
socket = null;
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* @return the targetIpAddress
*/
public String getTargetIpAddress() {
return targetIpAddress;
}
/**
* @param targetIpAddress
* the targetIpAddress to set
*/
public void setTargetIpAddress(String targetIpAddress) {
this.targetIpAddress = targetIpAddress;
}
/**
* @return the targetPort
*/
public int getTargetPort() {
return targetPort;
}
/**
* @param targetPort
* the targetPort to set
*/
public void setTargetPort(int targetPort) {
this.targetPort = targetPort;
}
}
package com.sse.monitor.serv;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
/**
* 服务器套接字工厂
* Copyright: Copyright (c) 2008
* @author cuishen
* @version 1.0
*/
public class ServerSocketFactory {
private static ServerSocket server;
private static Socket client;
private boolean ifRunServer = true;
public void runServer(int port) throws IOException {
//本地建立一个套接字服务器，等待其他机器访问
server = new ServerSocket(port);
System.out.println("Socket Server Start...");
new ServerThread();
}
class ServerThread implements Runnable {
Thread t;
public ServerThread() {
t = new Thread(this);
t.start();
}
public void run() {
try {
while(ifRunServer) {
if(client == null) client = server.accept();
if(client != null) //getMessage();
Thread.sleep(ReadConfig.serverThreadSleep);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
catch (IOException e) {
e.printStackTrace();
}
}
public void stopThread() {
try {
ifRunServer = false;
this.t.join(100);
} catch (InterruptedException ex) {
System.out.println("socket服务器线程终止异常！！！");
} finally {
t = null;
}
}
}
}