Java网络编程（一）-Socket编程

柔光的暖阳◎ 2022-06-13 04:46 288阅读 0赞

从本篇博客开始，后面几篇博客会着重介绍Java网络编程相关方面的知识，主要涉及Socket编程，Http协议编程。  
在网络通讯中，我们把主动发起通信请求的程序称为**客户端**，而在通讯中等待客户端发起请求建立连接的程序称为**服务端**。因而网络编程最重要的就是分别开发客户端程序和服务端程序。  
![这里写图片描述][SouthEast]

对于请求建立连接客户端，Java提供了Socket类用于客户端开发，主要完成以下四个基本操作：连接远程主机，发送数据，接收数据，关闭连接。  
对于接收连接的服务端，Java提供了ServerSocket类表示服务器Socket，主要完成以下三个基本操作：绑定端口，监听入站数据，在绑定端口上接受来自远程机器的连接。  
Java利用socket实现了客户端-服务端全双工及时通讯，即客户端和服务端可以同时发送和接收数据。

**一、客户端Socket**  
1、利用Socket构造器构造套接字  
2、Socket尝试连接服务器主机  
3、建立连接后，利用socket获得输入输出流，实现相互交互数据  
4、通信结束后，关闭输入输出流和Socket连接  
下面看一个具体的Client端示例：

package com.wygu.client;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    import java.io.OutputStream;
    import java.io.OutputStreamWriter;
    import java.io.Writer;
    import java.net.Socket;
    
    public class TestClient { 
    
        public static void main(String[] args) {
            Socket socket = null;
            OutputStream outputStream = null;
            InputStream inputStream = null;
            try {
                //创建Socket套接字，建立和服务端：127.0.0.1，端口号：8080的连接
                socket = new Socket("127.0.0.1",8080);
                //设置Socket响应超时时间，超时时间按照毫秒度量，下面设置为10秒
                //socket.setSoTimeout(100000);
    
                //返回输出流，向服务端写入数据
                outputStream = socket.getOutputStream();
                //包装输出流OutputStream为Writer，向服务端写入数据
                Writer writer = new OutputStreamWriter(outputStream);
                String sendData = "Hello,Server!!";
                System.out.println("Client send:"+sendData);    
                writer.write(sendData);
                //强制输出
                writer.flush();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }finally{
                try {
                    if(null!=socket){
                        //关闭socket连接
                        socket.close();
                    }
                    if(null!=outputStream){
                        //关闭输出流
                        outputStream.close();
                    }
                    if(null!=inputStream){
                        //关闭输入流
                        inputStream.close();
                    }       
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }           
            }
    
    
        }
    
    }

**二、服务端ServerSocket**  
1、利用ServerSocket()构造器在一个特定端口创建ServerSocket实例  
2、ServerSocket使用accept()方法监听特定端口的入站连接。其中，accept()会一直阻塞，直到一个客户端尝试建立连接后，accept()会返回一个客户端和服务端的Socket实例  
3、利用socket获得输入输出流，实现相互交互数据  
4、交互完毕后，关闭连接  
5、服务端返回到步骤2，等待下一次的连接  
下面看一个具体的Server端实例：

package com.wygu.client;
    
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.OutputStreamWriter;
    import java.io.Writer;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    public class TestServer { 
    
        public static void main(String[] args) {
            ServerSocket server = null;
            try {
                //创建端口号为8080的ServerSocket实例
                server = new ServerSocket(8080);
                while(true){
                    Socket connection = null;
                    InputStreamReader reader = null;
                    Writer writer = null;
                    try {
                        //阻塞式等待客户端连接
                        connection = server.accept();
                        //创建输入流，并读取客户端发送的数据
                        reader= new InputStreamReader(connection.getInputStream());
                        StringBuilder receiveData = new StringBuilder();
                        for(int c=reader.read();c!=-1;c=reader.read()){
                            receiveData.append((char)c);
                        }
                        System.out.println("Server receive:"+receiveData.toString());   
                        Thread.sleep(1000);
                    }catch (IOException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            if(null!=connection){
                                connection.close();
                            }
                            if(null!=reader){
                                reader.close();
                            }
                            if(null!=writer){
                                writer.close();
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
    
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if(null!=server){
                    try {
                        server.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }
    }

**三、Socket通信插件项目**  
项目中涉及到的功能点：长连接，多线程，消息队列。  
**1、客户端**  
客户端的启动可以通过main()方法启动，不管被启动多少次，保证只建立一条链路  
**1）Client**

package com.wygu.client;
    
    import java.io.BufferedReader;
    import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    import java.io.OutputStream;
    import java.io.OutputStreamWriter;
    import java.io.Writer;
    import java.net.InetSocketAddress;
    import java.net.Socket;
    import com.wygu.queue.MsgInQueue;
    
    public class Client implements Runnable{ 
    
        //将Socket连接Client设置为单例模式
        private volatile static Client clientInsance = null;
    
        public static Client getInstance(){
            if(null==clientInsance){
                synchronized (Client.class) {
                    if(null==clientInsance){
                        clientInsance = new Client();
                    }
                }
            }
            return clientInsance;
        }
    
        private String serverHost = null;
        private String serverPort = null;
        private Socket socketClient = null;
        private InputStream inputStream = null;
        private OutputStream outputStream = null;
    
    
        private long systemCurTime;
        //设置线程沉睡时间
        private long minSleepTime = 2;
        private long maxSleepTime = 20;
        private long sleepTime;
        private long heartBeatTime = 60;
    
        private boolean running = true;
    
        public void startClient(String host,String port){
            initSocket(host, port);
            Thread thread = new Thread(this);
            thread.setDaemon(true);
            thread.start();
            try {
                Thread.sleep(3 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        private void initSocket(String host, String port){
            this.serverHost = host;
            this.serverPort = port;
        }
    
        private void connect() throws NumberFormatException, IOException{
            this.socketClient = new Socket();
            this.socketClient.connect(new InetSocketAddress(serverHost, Integer.valueOf(serverPort)));
            this.socketClient.setSoTimeout(10000);
            this.inputStream = new DataInputStream(socketClient.getInputStream());
            this.outputStream = new DataOutputStream(socketClient.getOutputStream());
            this.sleepTime = this.minSleepTime;
        }
    
        @Override
        public void run() {
            try {
                connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
            setActiveTime();
            while(running){
                try {
                    Thread.sleep(sleepTime);
                    //从发送队列中取出消息，然后发送
                    if(sendMessage()) {                     
                        setActiveTime();
                    }
                    //从接收队列中取出消息，然后转换
                    if(inputStream.available() > 0){
                        onMessage();
                        setActiveTime();
                        sleepTime = this.minSleepTime;
                    }
                    else{
                        sleepTime = this.maxSleepTime;
                    }
                    if((this.systemCurTime + this.heartBeatTime * 1000L) < System.currentTimeMillis()){
                        activeTest();
                        setActiveTime();
                    }
                } catch (InterruptedException e) {                  
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (Exception e) {
                    e.printStackTrace();
                }   
            }
    
        }
    
        private void onMessage() throws IOException {       
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
            String receiveData = bufferedReader.readLine(); 
            System.out.println("Client receive:"+receiveData);  
        }
    
        private boolean sendMessage() {
            try {
                String mString = MsgInQueue.getInstance().getMsg();         
                if(null!=mString){
                    System.out.println("Client send:"+mString); 
                    Writer writer = new OutputStreamWriter(outputStream);
                    writer.write(mString+'\n');//加入换行符，按行读取
                    writer.flush();
                }
    
            } catch (IOException e) {
                e.printStackTrace();
            }
    
            return false;
        }
    
        //心跳包测试
        private void activeTest() throws IOException {
            Writer writer = new OutputStreamWriter(outputStream);
            String heatBeatStr = "00"+'\n';//加入换行符，按行读取
            writer.write(heatBeatStr);
            writer.flush();
        }
    
        private void setActiveTime() {
            this.systemCurTime = System.currentTimeMillis();
    
        }
    
    }

**2）消息队列**

package com.wygu.queue;
    
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    public class MsgInQueue { 
    
        private volatile static MsgInQueue msgInQueue = null;
        //单例模式，保证整个机制中只有一个消息队列实例 
        public static MsgInQueue getInstance(){
            if(null==msgInQueue){
                synchronized (MsgInQueue.class) {
                    if(null==msgInQueue){
                        msgInQueue = new MsgInQueue();
                    }
                }
            }
            return msgInQueue;
        }
    
        private ConcurrentLinkedQueue<String> msgQueue = new ConcurrentLinkedQueue<String>();
    
        public String getMsg(){
            return msgQueue.poll();
        }
    
        public void putMsg(String inputMsg){
            msgQueue.add(inputMsg);
        }
    
    }

**3）Main程序**

package com.wygu.client;
    
    import java.util.Scanner;
    
    import com.wygu.queue.MsgInQueue;
    
    public class Main {
    
        @SuppressWarnings("resource")
        public static void main(String[] args) {
            String clientIp = "127.0.0.1";
            String clientport = "10800";
            Client.getInstance().startClient(clientIp, clientport);
            Scanner in=new Scanner(System.in);
            String inputStr = null;
            //只建立一条链路，实现多线程客户端访问服务端
            while(!"quit".equals(inputStr=in.nextLine())){
                MsgInQueue.getInstance().putMsg(inputStr);
            }
    
        }
    
    }

代码可以实现客户端和服务端之间只需要建立一条链路，通过消息队列机制实现多线程同时发送消息，为保证不同的线程收到属于自身的服务端应答，可以加入消息ID机制实现。

**2、服务端**  
服务端接收到一个客户端连接后，会创建一个线程处理该连接的消息机制  
**1）Server端**

package com.wygu.server;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    public class Server implements Runnable{ 
    
        private String serverPort;
    
        private Server(String port){
            this.setServerPort(port);
        }
    
        public static void startServer(String port){
            Thread thread = new Thread(new Server(port));
            thread.start();
        }
    
        @Override
        public void run() {
            try {
                ServerSocket _server = new ServerSocket(Integer.valueOf(this.getServerPort()));
                System.out.println(String.format("Server [%s] startup success!", _server.getLocalSocketAddress()));
                while (true) {
                    try {
                        Socket socket = _server.accept();
                        Connection connection = new Connection(socket);
                        System.out.println("Accept socket from " + socket.getInetAddress().getHostAddress() + " " + socket.getPort());
                        new Thread(connection).start();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    
        public String getServerPort() {
            return serverPort;
        }
    
        public void setServerPort(String serverPort) {
            this.serverPort = serverPort;
        }
    
    }

**2）消息处理线程**

package com.wygu.server;
    
    import java.io.BufferedReader;
    import java.io.DataInputStream;
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    import java.io.OutputStream;
    import java.io.OutputStreamWriter;
    import java.io.Writer;
    import java.net.Socket;
    
    public class Connection implements Runnable{ 
    
        private Socket socket;
        private InputStream inputStream;
        private OutputStream outputStream;
    
        public Connection(Socket socket) {
            this.socket = socket;
        }
    
        private void init() throws IOException{
            this.inputStream = socket.getInputStream();
            this.outputStream = socket.getOutputStream();
        }
    
    
        @Override
        public void run() {
            try {
                init();
                Writer writer = new OutputStreamWriter(outputStream);
                while (true) {
                    try {
                        String receiveData = null;
                        if (inputStream.available() > 0) {
                            receiveData = readMsg(inputStream);
                            if(null!=receiveData){
                                System.out.println("Server receive:"+receiveData);  
                                String sendData = "I have received!"+'\n';   //服务端给予客户端应答
                                System.out.println("Server send:"+sendData);    
                                writer.write(sendData);
                                writer.flush();
                            }
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    Thread.sleep(1000);
                }   
            } catch (Exception e) {
                e.printStackTrace();
            }       
        }
    
        private String readMsg(InputStream in) throws IOException {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
            String receiveData = bufferedReader.readLine();
            //检查心跳包
            if ("00".equals(receiveData)) {
                System.out.println("Server heart beat!");
                return null;
            }
            return receiveData;
        }
    }

**3）Main程序**

package com.wygu.server;
    
    public class Main {
    
        public static void main(String[] args) {
            String serverPort = "10800";
            Server.startServer(serverPort);
            while(true){
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

[SouthEast]: /images/20220613/9f738c047ffc49b2a74066505d200f30.png