Java 网络通信编程：服务器多任务连接与广播消息实现

//启动线程来实时读取服务器发来的消息
new Thread(new Runnable() {
    @Override
    public void run() {
        while (true) {
            String str;
            try {
                str = readMsg();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            System.out.println("server:" + str);
        }
    }
}).start();

//发送消息
while (true) {
    Scanner scanner = new Scanner(System.in);
    String line = scanner.nextLine();
    sendMsg(line);
}

整合后我们便可得到完整的自定义客户端代码。

import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Scanner;

public class MClient {
    public static void main(String[] args) {
        try {
            new MClient().startClient();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public InputStream is;
    public OutputStream os;

    public void startClient() throws Exception {
        Socket socket = new Socket("127.0.0.1", 2000);
        System.out.println("连接服务器");
        is = socket.getInputStream();
        os = socket.getOutputStream();

        //启动线程来实时读取服务器发来的消息
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    String str;
                    try {
                        str = readMsg();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println("server:" + str);
                }
            }
        }).start();

        //发送消息
        while (true) {
            Scanner scanner = new Scanner(System.in);
            String line = scanner.nextLine();
            sendMsg(line);
        }
    }

    //发送消息方法体
    public void sendMsg(String msg) throws Exception {
        String str = msg + "\r\n"; //拼接换行
        os.write(str.getBytes());
        os.flush();
    }

    //读取消息方法体
    public String readMsg() throws Exception {
        byte[] bytes = new byte[1024]; //保存消息的缓存区
        is.read(bytes);
        String msg = new String(bytes); //将 byte 数组转为字符串
        return msg.trim(); //去除空位
    }
}

2. 自定义服务端

2.1 服务端框架 (MServer)

此处服务器担任'消息中转站'的角色，它要与多个客户端建立实时连接。对于实时性，结合之前搭建客户端的经验，我们容易想到利用线程。

服务端对于每个连接过来的客户端，都会创建对应的 Socket 对象，为了实现消息广播功能，我们可以建立一个 ArrayList 集合用来存储 Socket 对象，就像一个'在线客户端名单'。接着循环用 accept() 方法阻塞监听每一个连接过来的客户端，将其加入 ArrayList 集合。accept() 方法是阻塞的，服务器在连接到新客户端之前不会向下执行代码，为了不影响每个客户端，我们需要在下面启动该客户端独立、全新的线程来保持通信。

import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;

public class MServer {
    public static void main(String[] args) {
        try {
            new MServer().startServer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void startServer() throws Exception {
        ServerSocket server = new ServerSocket(2000);
        System.out.println("启动服务器");
        ArrayList<Socket> list = new ArrayList<>();

        //重复监听连接过来的客户端
        while (true) {
            //阻塞监听连接过来的客户端
            Socket socket = server.accept();
            list.add(socket);

            //启动线程，保持通信
            ServerThread serverThread = new ServerThread(socket, list);
            new Thread(serverThread).start();
        }
    }
}

2.2 通信线程 (ServerThread)

搭建好服务器框架后我们来完善线程的具体内容。

收发消息的方法体我们可以基本套用客户端的，但客户端的 sendMsg() 只需要操作唯一的一个输出流（客户端只连一个服务器，只有一个和服务器通信的输出流）；而服务端的 sendMsg() 必须针对不同客户端的输出流（每个客户端对应一个独立的 Socket，每个 Socket 有专属的输出流），通过指定不同的 os 参数来区分要发送的目标客户端。

//发送消息方法体
public void sendMsg(OutputStream os, String msg) throws Exception {
    String str = msg + "\r\n"; //拼接换行
    os.write(str.getBytes());
    os.flush();
}

通信线程对每个客户端都是独立的，因此构造函数中需要传入对应的 Socket 对象。同时为了实现向其他客户端发送消息的功能，需要其他客户端的 Socket 对象，因而传入 ArrayList 集合，所有线程共享同一个 ArrayList 集合。

public ServerThread(Socket socket, ArrayList<Socket> list) {
    this.socket = socket;
    this.list = list;
    //获取客户端输入输出流
    try {
        is = socket.getInputStream();
        os = socket.getOutputStream();
        sendMsg(os, "客户端连接成功");
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

最后重写 run() 方法。对于广播功能，我们应该遍历 ArrayList，对除去客户端本身的其他客户端，将消息写入输出流。

public void run() {
    while (true) {
        try {
            //接收信息
            String msg = readMsg();
            System.out.println("client:" + msg);

            //广播功能
            for (Socket s : list) {
                if (s != socket) { //排除自己客户端
                    OutputStream os = s.getOutputStream();
                    sendMsg(os, msg);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

3. 结果演示

如此我们便实现了服务器多任务连接 + 广播消息。最后演示一下运行结果。

启动服务器和多个客户端，均连接成功。服务器会向客户端发送'客户端连接成功'的提示。

在任意客户端发送消息，服务器会接收到并将消息中转至其他客户端。

（两个不同客户端分别发送'abc123'和'你好'）