一个项目看java TCP/IP Socket编程(1.3版)[转]

前一段时间刚做了个java程序和网络上多台机器的c程序通讯的项目，遵循的是TCP/IP协议，用到了java的Socket编程。网络通讯是java的强项，用TCP/IP协议可以方便的和网络上的其他程序互通消息。

先来介绍下网络协议：
    TCP/IP
        Transmission Control Protocol 传输控制协议
        Internet Protocol 互联网协议
    UDP
        User Datagram Protocol 用户数据协议

连接协议：
    分为：
    面向连接协议: Connection Oriented Protocol
    非连接协议: Connectionless Protocol

    1).面向连接协议是指两台电脑在传输数据前，先会建立一个专属的连接。就如电信局的交换机会为打电话双方提供专属连接一样。
    Internet上的面向连接协议就是TCP/IP
    特点：确认回应；分组序号；流量控制。
    TCP/IP属于可靠性传输，适合不容许有传输错误的网络程序设计使用

    2).非连接协议：无专属连接，无分组，容错，距离短，可同时对多台电脑进行数据传输
    Internet上的非连接协议就是UDP

    TCP在网络通信上有极强的生命力，例如远程连接（Telnet）和文件传输（FTP）都需要不定长度的数据被可靠地传输。相比之下UDP操作简单，而且仅需要较少的监护，因此通常用于局域网高可靠性的分散系统中client/server应用程序。

Socket 是程序与网络间的一种接口，大部分网络应用程序都是点对点的，所谓点就是服务器端和客户端所执行的程序。Socket是用来接收和传送分组的一个端点。

java的Socket编程要用到java.net包，最常用的是net包下的6个类：InetAddress(互联网协议 (IP) 地址)类，Socket(套接字)类，ServerSocket(套接字服务器)类，DatagramSocket(发送和接收数据报包的套接字)类，DatagramPacket(数据报包)类，MulticastSocket(多播数据报套接字类用于发送和接收 IP 多播包)类，其中InetAddress、Socket、ServerSocket类是属于TCP面向连接协议，DatagramSocket、DatagramPacket和MulticastSocket类则属于UDP非连接协议的传送类。

本项目因为使用TCP/IP协议，主要用到Socket和ServerSocket类

项目代码如下

Java代码 
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;  
    }  
  
}  

以上是Socket编程，ServerSocket在项目里没有用到，但是我也写了个包装类供参考

Java代码 
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;  
            }  
        }  
    }  
}  

Socket编程就是运用Socket或者ServerSocket类搭配线程来使用(由于TCP/IP属于可靠性传输，不会丢包)。可能会因为在发送请求或者接受消息时Socket阻塞而导致主线程挂死，因此发送请求、接收消息的方法要放进子线程里处理；对于同一目标ip和端口，在同一个子线程里只能new一个Socket，也就是说，要对同一地址建立多条连接，就要开启多个线程。而且注意连接可能会因作用不同分长连接和短连接，要分别处理，本项目中发送请求和接受message就分别属于短连接和长连接，因此分别开发了RequestSocketThread和GetMessageSocketThread两个子线程区分对待。可以同时开发个Message类来封装打包和解包消息的方法(项目中开发MessageFactory.java)，方便调用

发表于 2010-07-14 22:10 DZM.528 阅读(1350) 评论(0) 编辑收藏所属分类: 网络编程

导航

统计

随笔分类(70)

收藏夹(4)

左邻右舍

我的链接

最新评论

阅读排行榜