基于Java實(shí)現(xiàn)Socket編程入門

更新時(shí)間：2022年03月10日 11:43:37 作者：Lumin

Java最初是作為網(wǎng)絡(luò)編程語言出現(xiàn)的，使得客戶端和服務(wù)器的溝通變成了現(xiàn)實(shí)，而在網(wǎng)絡(luò)編程中，使用最多的就是Socket，本文就來介紹一下基于Java實(shí)現(xiàn)Socket編程入門，感興趣的可以來了解一下

認(rèn)識(shí)Socket

socket，又稱套接字，是在不同的進(jìn)程間進(jìn)行網(wǎng)絡(luò)通訊的一種協(xié)議、約定或者說是規(guī)范。

對(duì)于socket編程，它更多的時(shí)候像是基于TCP/UDP等協(xié)議做的一層封裝或者說抽象，是一套系統(tǒng)所提供的用于進(jìn)行網(wǎng)絡(luò)通信相關(guān)編程的接口。

建立socket的基本流程

我們以linux操作系統(tǒng)提供的基本api為例，了解建立一個(gè)socket通信的基本流程：

可以看到本質(zhì)上，socket是對(duì)tcp連接（當(dāng)然也有可能是udp等其他連接）協(xié)議，在編程層面上的簡化和抽象。

1.最基本的Socket示范

1.1 單向通信

首先，我們從只發(fā)送和接收一次消息的socket基礎(chǔ)代碼開始：

服務(wù)端:

package com.marklux.socket.base;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * The very basic socket server that only listen one single message.
 */

public class BaseSocketServer {

    private ServerSocket server;
    private Socket socket;
    private int port;
    private InputStream inputStream;
    private static final int MAX_BUFFER_SIZE = 1024;

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public BaseSocketServer(int port) {
        this.port = port;
    }

    public void runServerSingle() throws IOException {
        this.server = new ServerSocket(this.port);

        System.out.println("base socket server started.");
        // the code will block here till the request come.
        this.socket = server.accept();

        this.inputStream = this.socket.getInputStream();

        byte[] readBytes = new byte[MAX_BUFFER_SIZE];

        int msgLen;
        StringBuilder stringBuilder = new StringBuilder();

        while ((msgLen = inputStream.read(readBytes)) != -1) {
            stringBuilder.append(new String(readBytes,0,msgLen,"UTF-8"));
        }

        System.out.println("get message from client: " + stringBuilder);

        inputStream.close();
        socket.close();
        server.close();
    }

    public static void main(String[] args) {
        BaseSocketServer bs = new BaseSocketServer(9799);
        try {
            bs.runServerSingle();
        }catch (IOException e) {
            e.printStackTrace();
        }

    }
}

客戶端:

package com.marklux.socket.base;

import java.io.IOException;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.Socket;

/**
 * The very basic socket client that only send one single message.
 */

public class BaseSocketClient {
    private String serverHost;
    private int serverPort;
    private Socket socket;
    private OutputStream outputStream;

    public BaseSocketClient(String host, int port) {
        this.serverHost = host;
        this.serverPort = port;
    }

    public void connetServer() throws IOException {
        this.socket = new Socket(this.serverHost, this.serverPort);
        this.outputStream = socket.getOutputStream();
        // why the output stream?
    }

    public void sendSingle(String message) throws IOException {
        try {
            this.outputStream.write(message.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            System.out.println(e.getMessage());
        }
        this.outputStream.close();
        this.socket.close();
    }

    public static void main(String[] args) {
        BaseSocketClient bc = new BaseSocketClient("127.0.0.1",9799);
        try {
            bc.connetServer();
            bc.sendSingle("Hi from mark.");
        }catch (IOException e) {
            e.printStackTrace();
        }
    }
}

先運(yùn)行服務(wù)端，再運(yùn)行客戶端，就可以看到效果。

注意這里的IO操作實(shí)現(xiàn)，我們使用了一個(gè)大小為MAX_BUFFER_SIZE的byte數(shù)組作為緩沖區(qū)，然后從輸入流中取出字節(jié)放置到緩沖區(qū)，再從緩沖區(qū)中取出字節(jié)構(gòu)建到字符串中去，這在輸入流文件很大時(shí)非常有用，事實(shí)上，后面要講到的NIO也是基于這種思路實(shí)現(xiàn)的。

1.2 雙向通信

上面的例子只實(shí)現(xiàn)了一次單向的通信，這顯然有點(diǎn)浪費(fèi)通道。socket連接支持全雙工的雙向通信（底層是tcp），下面的例子中，服務(wù)端在收到客戶端的消息后，將返回給客戶端一個(gè)回執(zhí)。

并且我們使用了一些java.io包裝好的方法，來簡化整個(gè)通信的流程（因?yàn)橄㈤L度不大，不再使用緩沖區(qū)）。

服務(wù)端:

public void runServer() throws IOException {
        this.serverSocket = new ServerSocket(port);
        this.socket = serverSocket.accept();
        this.inputStream = socket.getInputStream();

        String message = new String(inputStream.readAllBytes(), "UTF-8");

        System.out.println("received message: " + message);

        this.socket.shutdownInput(); // 告訴客戶端接收已經(jīng)完畢，之后只能發(fā)送

        // write the receipt.

        this.outputStream = this.socket.getOutputStream();
        String receipt = "We received your message: " + message;
        outputStream.write(receipt.getBytes("UTF-8"));

        this.outputStream.close();
        this.socket.close();
    }

客戶端:

public void sendMessage(String message) throws IOException {
        this.socket = new Socket(host,port);
        this.outputStream = socket.getOutputStream();
        this.outputStream.write(message.getBytes("UTF-8"));
        this.socket.shutdownOutput(); // 告訴服務(wù)器，所有的發(fā)送動(dòng)作已經(jīng)結(jié)束，之后只能接收
        this.inputStream = socket.getInputStream();
        String receipt = new String(inputStream.readAllBytes(), "UTF-8");
        System.out.println("got receipt: " + receipt);
        this.inputStream.close();
        this.socket.close();
    }

注意這里我們?cè)诜?wù)端接受到消息以及客戶端發(fā)送消息后，分別調(diào)用了shutdownInput()和shutdownOutput()而不是直接close對(duì)應(yīng)的stream，這是因?yàn)樵陉P(guān)閉任何一個(gè)stream，都會(huì)直接導(dǎo)致socket的關(guān)閉，也就無法進(jìn)行后面回執(zhí)的發(fā)送了。
但是注意，調(diào)用shutdownInput()和shutdownOutput()之后，對(duì)應(yīng)的流也會(huì)被關(guān)閉，不能再次向socket發(fā)送/寫入了。

2. 發(fā)送更多的消息：結(jié)束的界定

剛才的兩個(gè)例子中，每次打開流，都只能進(jìn)行一次寫入/讀取操作，結(jié)束后對(duì)應(yīng)流被關(guān)閉，就無法再次寫入/讀取了。

在這種情況下，如果要發(fā)送兩次消息，就不得不建立兩個(gè)socket，既耗資源又麻煩。其實(shí)我們完全可以不關(guān)閉對(duì)應(yīng)的流，只要分次寫入消息就可以了。

但是這樣的話，我們就必須面對(duì)另一個(gè)問題：如何判斷一次消息發(fā)送的結(jié)束呢？

2.1 使用特殊符號(hào)

最簡單的辦法是使用一些特殊的符號(hào)來標(biāo)記一次發(fā)送完成，服務(wù)端只要讀到對(duì)應(yīng)的符號(hào)就可以完成一次讀取，然后進(jìn)行相關(guān)的處理操作。

下面的例子中我們使用換行符\n來標(biāo)記一次發(fā)送的結(jié)束，服務(wù)端每接收到一個(gè)消息，就打印一次，并且使用了Scanner來簡化操作:

服務(wù)端:

public void runServer() throws IOException {
        this.server = new ServerSocket(this.port);

        System.out.println("base socket server started.");

        this.socket = server.accept();
        // the code will block here till the request come.

        this.inputStream = this.socket.getInputStream();
        Scanner sc = new Scanner(this.inputStream);
        while (sc.hasNextLine()) {
            System.out.println("get info from client: " + sc.nextLine());
        } // 循環(huán)接收并輸出消息內(nèi)容
        this.inputStream.close();
        socket.close();
    }

客戶端:

public void connetServer() throws IOException {
        this.socket = new Socket(this.serverHost, this.serverPort);
        this.outputStream = socket.getOutputStream();
    }

public void send(String message) throws IOException {
        String sendMsg = message + "\n"; // we mark \n as a end of line.
        try {
            this.outputStream.write(sendMsg.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            System.out.println(e.getMessage());
        }
//        this.outputStream.close();
//        this.socket.shutdownOutput();
    }

 public static void main(String[] args) {
        CycleSocketClient cc = new CycleSocketClient("127.0.0.1", 9799);
        try {
            cc.connetServer();
            Scanner sc = new Scanner(System.in);
            while (sc.hasNext()) {
                String line = sc.nextLine();
                cc.send(line);
            }
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

運(yùn)行后效果是，客戶端每輸入一行文字按下回車后，服務(wù)端就會(huì)打印出對(duì)應(yīng)的消息讀取記錄。

2.2 根據(jù)長度界定

回到原點(diǎn)，我們之所以不好定位消息什么時(shí)候結(jié)束，是因?yàn)槲覀儾荒軌虼_定每次消息的長度。

那么其實(shí)可以先將消息的長度發(fā)送出去，當(dāng)服務(wù)端知道消息的長度后，就能夠完成一次消息的接收了。

總的來說，發(fā)送一次消息變成了兩個(gè)步驟

發(fā)送消息的長度
發(fā)送消息

最后的問題就是，“發(fā)送消息的長度”這一步驟所發(fā)送的字節(jié)量必須是固定的，否則我們?nèi)匀粫?huì)陷入僵局。

一般來說，我們可以使用固定的字節(jié)數(shù)來保存消息的長度，比如規(guī)定前2個(gè)字節(jié)就是消息的長度，不過這樣我們能夠傳送的消息最大長度也就被固定死了，以2個(gè)字節(jié)為例，我們發(fā)送的消息最大長度不超過2^16個(gè)字節(jié)即64K。

如果你了解一些字符的編碼，就會(huì)知道，其實(shí)我們可以使用變長的空間來儲(chǔ)存消息的長度，比如：

第一個(gè)字節(jié)首位為0：即0XXXXXXX，表示長度就一個(gè)字節(jié)，最大128，表示128B
第一個(gè)字節(jié)首位為110，那么附帶后面一個(gè)字節(jié)表示長度：即110XXXXX 10XXXXXX，最大2048，表示2K
第一個(gè)字節(jié)首位為1110，那么附帶后面二個(gè)字節(jié)表示長度：即110XXXXX 10XXXXXX 10XXXXXX，最大131072，表示128K
依次類推

當(dāng)然這樣實(shí)現(xiàn)起來會(huì)麻煩一些，因此下面的例子里我們?nèi)匀皇褂霉潭ǖ膬蓚€(gè)字節(jié)來記錄消息的長度。

服務(wù)端:

public void runServer() throws IOException {
        this.serverSocket = new ServerSocket(this.port);
        this.socket = serverSocket.accept();
        this.inputStream = socket.getInputStream();
        byte[] bytes;
        while (true) {
            // 先讀第一個(gè)字節(jié)
            int first = inputStream.read();
            if (first == -1) {
                // 如果是-1，說明輸入流已經(jīng)被關(guān)閉了，也就不需要繼續(xù)監(jiān)聽了
                this.socket.close();
                break;
            }
            // 讀取第二個(gè)字節(jié)
            int second = inputStream.read();

            int length = (first << 8) + second; // 用位運(yùn)算將兩個(gè)字節(jié)拼起來成為真正的長度

            bytes = new byte[length]; // 構(gòu)建指定長度的字節(jié)大小來儲(chǔ)存消息即可

            inputStream.read(bytes);

            System.out.println("receive message: " + new String(bytes,"UTF-8"));
        }
    }

客戶端:

public void connetServer() throws IOException {
        this.socket = new Socket(host,port);
        this.outputStream = socket.getOutputStream();
    }

public void sendMessage(String message) throws IOException {
        // 首先要把message轉(zhuǎn)換成bytes以便處理
        byte[] bytes = message.getBytes("UTF-8");
        // 接下來傳輸兩個(gè)字節(jié)的長度，依然使用移位實(shí)現(xiàn)
        int length = bytes.length;
        this.outputStream.write(length >> 8); // write默認(rèn)一次只傳輸一個(gè)字節(jié)
        this.outputStream.write(length);
        // 傳輸完長度后，再正式傳送消息
        this.outputStream.write(bytes);
    }

public static void main(String[] args) {
        LengthSocketClient lc = new LengthSocketClient("127.0.0.1",9799);
        try {
            lc.connetServer();
            Scanner sc = new Scanner(System.in);
            while (sc.hasNextLine()) {
                lc.sendMessage(sc.nextLine());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

3. 處理更多的連接：多線程

3.1 同時(shí)實(shí)現(xiàn)消息的發(fā)送與接收

在考慮服務(wù)端處理多連接之前，我們先考慮使用多線程改造一下原有的一對(duì)一對(duì)話實(shí)例。

在原有的例子中，消息的接收方并不能主動(dòng)地向?qū)Ψ桨l(fā)送消息，換句話說我們并沒有實(shí)現(xiàn)真正的互相對(duì)話，這主要是因?yàn)橄⒌陌l(fā)送和接收這兩個(gè)動(dòng)作并不能同時(shí)進(jìn)行，因此我們需要使用兩個(gè)線程，其中一個(gè)用于監(jiān)聽鍵盤輸入并將其寫入socket，另一個(gè)則負(fù)責(zé)監(jiān)聽socket并將接受到的消息顯示。

出于簡單考慮，我們直接讓主線程負(fù)責(zé)鍵盤監(jiān)聽和消息發(fā)送，同時(shí)另外開啟一個(gè)線程用于拉取消息并顯示。

消息拉取線程 ListenThread.java

public class ListenThread implements Runnable {
    private Socket socket;
    private InputStream inputStream;

    public ListenThread(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() throws RuntimeException{
        try {
            this.inputStream = socket.getInputStream();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }

        while (true) {
            try {
                int first = this.inputStream.read();
                if (first == -1) {
                    // 輸入流已經(jīng)被關(guān)閉，無需繼續(xù)讀取
                    throw new RuntimeException("disconnected.");
                }
                int second = this.inputStream.read();
                int msgLength = (first<<8) + second;
                byte[] readBuffer = new byte[msgLength];
                this.inputStream.read(readBuffer);

                System.out.println("message from [" + socket.getInetAddress() + "]: " + new String(readBuffer,"UTF-8"));
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e.getMessage());
            }
        }
    }
}

主線程，啟動(dòng)時(shí)由用戶選擇是作為server還是client：

public class ChatSocket {
    private String host;
    private int port;
    private Socket socket;
    private ServerSocket serverSocket;
    private OutputStream outputStream;

    // 以服務(wù)端形式啟動(dòng)，創(chuàng)建會(huì)話
    public void runAsServer(int port) throws IOException {
        this.serverSocket = new ServerSocket(port);
        System.out.println("[log] server started at port " + port);
        // 等待客戶端的加入
        this.socket = serverSocket.accept();
        System.out.println("[log] successful connected with " + socket.getInetAddress());
        // 啟動(dòng)監(jiān)聽線程
        Thread listenThread = new Thread(new ListenThread(this.socket));
        listenThread.start();
        waitAndSend();
    }

    // 以客戶端形式啟動(dòng)，加入會(huì)話
    public void runAsClient(String host, int port) throws IOException {
        this.socket = new Socket(host, port);
        System.out.println("[log] successful connected to server " + socket.getInetAddress());
        Thread listenThread = new Thread(new ListenThread(this.socket));
        listenThread.start();
        waitAndSend();
    }

    public void waitAndSend() throws IOException {
        this.outputStream = this.socket.getOutputStream();
        Scanner sc = new Scanner(System.in);
        while (sc.hasNextLine()) {
            this.sendMessage(sc.nextLine());
        }
    }

    public void sendMessage(String message) throws IOException {
        byte[] msgBytes = message.getBytes("UTF-8");
        int length = msgBytes.length;
        outputStream.write(length>>8);
        outputStream.write(length);
        outputStream.write(msgBytes);
    }

    public static void main(String[] args) {
        Scanner scanner = new Scanner(System.in);
        ChatSocket chatSocket = new ChatSocket();
        System.out.println("select connect type: 1 for server and 2 for client");
        int type = Integer.parseInt(scanner.nextLine().toString());
        if (type == 1) {
            System.out.print("input server port: ");
            int port = scanner.nextInt();
            try {
                chatSocket.runAsServer(port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }else if (type == 2) {
            System.out.print("input server host: ");
            String host = scanner.nextLine();
            System.out.print("input server port: ");
            int port = scanner.nextInt();
            try {
                chatSocket.runAsClient(host, port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

3.2 使用線程池優(yōu)化服務(wù)端并發(fā)能力

作為服務(wù)端，如果一次只跟一個(gè)客戶端建立socket連接，未免顯得太過浪費(fèi)資源，因此我們完全可以讓服務(wù)端和多個(gè)客戶端建立多個(gè)socket。

那么既然要處理多個(gè)連接，就不得不面對(duì)并發(fā)問題了（當(dāng)然，你也可以寫循環(huán)輪流處理）。我們可以使用多線程來處理并發(fā)，不過線程的創(chuàng)建和銷毀都會(huì)消耗大量的資源和時(shí)間，所以最好一步到位，用一個(gè)線程池來實(shí)現(xiàn)。

下面給出一個(gè)示范性質(zhì)的服務(wù)端代碼：

public class SocketServer {
  public static void main(String args[]) throws Exception {
    // 監(jiān)聽指定的端口
    int port = 55533;
    ServerSocket server = new ServerSocket(port);
    // server將一直等待連接的到來
    System.out.println("server將一直等待連接的到來");

    //如果使用多線程，那就需要線程池，防止并發(fā)過高時(shí)創(chuàng)建過多線程耗盡資源
    ExecutorService threadPool = Executors.newFixedThreadPool(100);
    
    while (true) {
      Socket socket = server.accept();
      
      Runnable runnable=()->{
        try {
          // 建立好連接后，從socket中獲取輸入流，并建立緩沖區(qū)進(jìn)行讀取
          InputStream inputStream = socket.getInputStream();
          byte[] bytes = new byte[1024];
          int len;
          StringBuilder sb = new StringBuilder();
          while ((len = inputStream.read(bytes)) != -1) {
            // 注意指定編碼格式，發(fā)送方和接收方一定要統(tǒng)一，建議使用UTF-8
            sb.append(new String(bytes, 0, len, "UTF-8"));
          }
          System.out.println("get message from client: " + sb);
          inputStream.close();
          socket.close();
        } catch (Exception e) {
          e.printStackTrace();
        }
      };
      threadPool.submit(runnable);
    }

  }
}

4. 連接?；?/h2>
我想你不難發(fā)現(xiàn)一個(gè)問題，那就是當(dāng)socket連接成功建立后，如果中途發(fā)生異常導(dǎo)致其中一方斷開連接，此時(shí)另一方是無法發(fā)現(xiàn)的，只有在再次嘗試發(fā)送/接收消息才會(huì)因?yàn)閽伋霎惓６顺觥?/p>
簡單的說，就是我們維持的socket連接，是一個(gè)長連接，但我們沒有保證它的時(shí)效性，上一秒它可能還是可以用的，但是下一秒就不一定了。

4.1 使用心跳包

保證連接隨時(shí)可用的最常見方法就是定時(shí)發(fā)送心跳包，來檢測(cè)連接是否正常。這對(duì)于實(shí)時(shí)性要求很高的服務(wù)而言，還是非常重要的（比如消息推送）。

大體的方案如下：

雙方約定好心跳包的格式，要能夠區(qū)別于普通的消息。
客戶端每隔一定時(shí)間，就向服務(wù)端發(fā)送一個(gè)心跳包
服務(wù)端每接收到心跳包時(shí)，將其拋棄
如果客戶端的某個(gè)心跳包發(fā)送失敗，就可以判斷連接已經(jīng)斷開
如果對(duì)實(shí)時(shí)性要求很高，服務(wù)端也可以定時(shí)檢查客戶端發(fā)送心跳包的頻率，如果超過一定時(shí)間沒有發(fā)送可以認(rèn)為連接已經(jīng)斷開