Socket多线程实现服务端与多个客户端通信以及客户端之间的通信

使用Socket实现服务端与客户端通信

Socket

socket一般指套接字，将TCP/IP协议封装为几个简单的接口，应用层调用接口就能实现进程间的通信。通信的两个进程各自持有一个socket，双方通过socket提供的接口进行通信，socket是成对出现的。

socket通信实现过程

服务端创建ServerSocket对象，调用accept()方法监听请求，当接收到请求时，返回一个socket对象。

ServerSocket serverSocket = new ServerSocket(8888);//创建ServerSocket对象时需要绑定一个端口Socket accept = serverSocket.accept();

客户端创建Socket对象，连接服务端，socket的构造函数包含两个参数，第一个是服务端的ip地址，因为服务端在本机，所以使用localhost，第二个参数是服务端运行的端口，就是在创建ServerSocket对象时绑定的端口。

socket = new Socket("localhost",8888);

客户端和服务端分别通过各自持有的socket对象获取输入流和输出流。

InputStream is = socket.getInputStream();OutputStream os = socket.getOutputStream();

客户端和服务端可以通过输入流的read()方法发送消息和输出流的write()方法接收消息。

输入流和输出流的简单封装

虽然使用ObjectInputStream和ObjectOutputStream更加方便，但是为了更好的了解通信过程和方便开发，选择对InputSream和OutputStream进行封装，本文只封装了int和String类型。

读取一个int数据，read方法返回一个0-255范围的int数值，即8个bit，int有32个bit，所以需要读取4次，通过位运算整合为一个int变量。

public static int readInt(InputStream is) {int[] values = new int[4];try {for (int i = 0; i < 4; i++) {values[i] = is.read();}} catch (IOException e) {e.printStackTrace();}int value = values[0]<<24 | values[1]<<16 | values[2]<<8 | values[3]<<0;return value;}

发送一个int数据，需要将一个int数据拆分为4个部分，分4次发送。

public static void writeInt(OutputStream os,int value) {int[] values = new int[4];values[0] = (value>>24)&0xFF;values[1] = (value>>16)&0xFF;values[2] = (value>>8)&0xFF;values[3] = (value>>0)&0xFF;try{for (int i = 0; i < 4; i++) {os.write(values[i]);}}catch (IOException e){e.printStackTrace();}}

读取一个String字符串，字符串长度不是固定的，read()方法虽然使用-1作为输入结束的标识，但是只有在发送方将输出流关闭时，read()才会接受到-1，否则将一直阻塞，所以需要在发送字符串前发送字符串的长度，并且因为是以byte数组的形式发送的，发送的长度也需要转换为byte数组的长度。接收时使用byte数组接收，并且String提供了将byte数组作为参数的构造方法。

public static String readString(InputStream is) {int len = readInt(is);byte[] sByte = new byte[len];try {is.read(sByte);} catch (IOException e) {e.printStackTrace();}String s = new String(sByte);return s;}

发送一个String。

public static void writeString(OutputStream os,String s) {byte[] bytes = s.getBytes();int len = bytes.length;writeInt(os,len);try {os.write(bytes);} catch (IOException e) {e.printStackTrace();}}

多线程实现服务端与多个客户端同时通信

为了能持续收发消息，客户端与服务端都需要使用死循环，但是没有收发消息的时候程序就会阻塞，因此需要使用多线程，将消息的监听和收发交给一个线程来控制，服务端一个线程用来和一个客户端进行通信。

服务端需要创建一个ServerThread类，继承Thread类，重写run()方法，并且这个类的对象需要持有与客户端通信的socket。

public class ServerThread extends Thread {private Socket socket;private InputStream is;private OutputStream os;public ServerThread(Socket socket){this.socket = socket;try {is = socket.getInputStream();os = socket.getOutputStream();}catch (IOException e){e.printStackTrace();}}public Socket getSocket(){return socket;}@Overridepublic void run() {String infor = socket.getInetAddress()+":"+socket.getPort();IOUtil.writeString(os,infor+"你好");while (true){//等待客户端发送消息String msg = IOUtil.readString(is);String response = infor+">"+msg;IOUtil.writeString(os,response);}}}

Server类只需要负责接收连接请求、创建和管理线程。

public class Server {private ServerSocket serverSocket = null;private HashMap<String,ServerThread> threads = new HashMap<>();public void run() throws IOException {serverSocket = new ServerSocket(8888);System.out.println("服务端启动");while (true){System.out.println("正在监听");Socket accept = serverSocket.accept();String address = accept.getInetAddress()+":"+accept.getPort();System.out.println("连接成功");ServerThread serverThread = new ServerThread(accept);threads.put(address,serverThread);serverThread.start();}}}

客户端，实现一个简单的界面。

public class TestClient extends JFrame {private Socket socket;private InputStream is;private OutputStream os;private String response = "";private JTextArea jTextArea;public void init(){setSize(800,600);setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);setLocationRelativeTo(null);setTitle("Client");setLayout(null);setBackground(Color.WHITE);Font font = new Font(null,Font.BOLD,16);try {socket = new Socket("localhost",8888);is = socket.getInputStream();os = socket.getOutputStream();} catch (IOException e) {e.printStackTrace();}jTextArea = new JTextArea();jTextArea.setBounds(0,0,800,500);jTextArea.setFont(font);add(jTextArea);response = IOUtil.readString(is);jTextArea.setText(response);JTextField jTextField = new JTextField();jTextField.setFont(font);jTextField.setBounds(0,500,800,50);add(jTextField);jTextField.addActionListener(new ActionListener() {@Overridepublic void actionPerformed(ActionEvent e) {String text = jTextField.getText();jTextField.setText("");IOUtil.writeString(os,text);}});setVisible(true);recive();}public void recive(){while (true){String s = IOUtil.readString(is);response += "\n"+s;jTextArea.setText(response);}}}

运行结果

第一句是连接后服务端发送的消息，在最下方输入框输入要发送的内容

回车发送

服务端将发送的内容处理后发送回来，发送格式为ip:端口>发送的内容

两个客户端同时向服务端发送消息

客户端之间的通信

以服务端为中介，可以实现客户端之间的通信。客户端指定发送的消息类型，如私聊、群聊、文件等，指定接收的对象，将这些信息和要发送的内容封装成一个对象发送给服务端，服务端将消息转发给指定的客户端，就能实现客户端之间的通信。为了确定发送的对象，需要使每一个客户端都有唯一的标识，本文为了方便就使用ip地址+端口来标识。

为了便于演示，采用 "消息类型 接收对象 消息内容"三个部分作为输入格式，每个部分之间用空格隔开，实现客户端之间的一对一的通信和向所有客户端的广播。

实现查询客户端列表

在实现之前，可以先设计一个用来查看客户端列表的功能，在客户端向服务端发送指定的字符串，服务端返回所有连接的客户端的列表。我使用“allClients来表示这个指令。

修改Server类和ServerThread类，需要将Server中用于管理全部客户端的列表传给每个thread。

public class Server {private ServerSocket serverSocket = null;private HashMap<String,ServerThread> threads = new HashMap<>();public void run() throws IOException {serverSocket = new ServerSocket(8888);System.out.println("服务端启动");while (true){System.out.println("正在监听");Socket accept = serverSocket.accept();String address = accept.getInetAddress()+":"+accept.getPort();System.out.println("连接成功");ServerThread serverThread = new ServerThread(accept,threads);threads.put(address,serverThread);serverThread.start();}}public static void main(String[] args) throws IOException {new Server().run();}}

public class ServerThread extends Thread {private Socket socket;private InputStream is;private OutputStream os;private HashMap<String,ServerThread> threads;public ServerThread(Socket socket,HashMap<String,ServerThread> threads){this.socket = socket;this.threads = threads;try {is = socket.getInputStream();os = socket.getOutputStream();}catch (IOException e){e.printStackTrace();}}public Socket getSocket(){return socket;}@Overridepublic void run() {String infor = socket.getInetAddress()+":"+socket.getPort();IOUtil.writeString(os,infor+"你好");while (true){//等待客户端发送消息String msg = IOUtil.readString(is);String response = infor+">"+msg;if(msg.split(" ")[0].equals("allClients")){String users = "";Iterator<String> iterator = threads.keySet().iterator();while (iterator.hasNext()){String user = iterator.next();if (!user.equals(infor)){users += "\n"+user;}}response += users;}IOUtil.writeString(os,response);}}}

运行结果

首先运行三个客户端

实现这个功能后可以更方便地在客户端进行操作。

实现客户端间一对一的通信和广播

同样需要定义命令。使用“sto”作为私聊的命令，即"Send to one"的缩写。使用“sta”作为广播的命令，“send to all”。服务端接收到客户端的消息后，使用splite()方法将字符串按照空格划分，得到一个String数组，取出第一个字符串，判断消息类型，是私聊则取出第二个字符串即接收的客户端，在全部客户端中查询，存在这个客户端就把第三字符串发送过去，不存在就返回一个提示；是广播就遍历客户端发送消息内容（数组中第二个字符串），除去发送的客户端。

修改ServerThread类中的run方法

public void run() {String infor = socket.getInetAddress()+":"+socket.getPort();IOUtil.writeString(os,infor+"你好");while (true){//等待客户端发送消息String msg = IOUtil.readString(is);String response = infor+">"+msg;if(msg.split(" ")[0].equals("allClients")){String users = "";Iterator<String> iterator = threads.keySet().iterator();while (iterator.hasNext()){String user = iterator.next();if (!user.equals(infor)){users += "\n"+user;}}response += users;}else if (msg.split(" ")[0].equals("sto")){String user = msg.split(" ")[1];if (threads.keySet().contains(user)){Socket userSocket = threads.get(user).getSocket();try {OutputStream userSocketOutputStream = userSocket.getOutputStream();IOUtil.writeString(userSocketOutputStream,infor+"说:"+msg.split(" ")[2]);} catch (IOException e) {e.printStackTrace();}}else {response = response+"\n用户不存在";}}else if (msg.split(" ")[0].equals("sta")){Iterator<String> users = threads.keySet().iterator();while (users.hasNext()){String user = users.next();if (!user.equals(infor)){Socket userSocket = threads.get(user).getSocket();try {OutputStream userOs = userSocket.getOutputStream();IOUtil.writeString(userOs,infor+"说:"+msg.split(" ")[1]);} catch (IOException e) {e.printStackTrace();}}}}IOUtil.writeString(os,response);}}

运行结果

本文的实现方式较为简单，仅提供一个多线程通信的实现思路，实际开发过程需要严格定义通信协议，封装消息对象，消息类型也更多。

完整代码

Server类

import java.io.IOException;import .ServerSocket;import .Socket;import java.util.HashMap;public class Server {private ServerSocket serverSocket = null;private HashMap<String,ServerThread> threads = new HashMap<>();public void run() throws IOException {serverSocket = new ServerSocket(8888);System.out.println("服务端启动");while (true){System.out.println("正在监听");Socket accept = serverSocket.accept();String address = accept.getInetAddress()+":"+accept.getPort();System.out.println("连接成功");ServerThread serverThread = new ServerThread(accept,threads);threads.put(address,serverThread);serverThread.start();}}public static void main(String[] args) throws IOException {new Server().run();}}

ServerThread类

import VedioOnline.Utils.IOUtil;import java.io.IOException;import java.io.InputStream;import java.io.OutputStream;import .Socket;import java.util.HashMap;import java.util.Iterator;public class ServerThread extends Thread {private Socket socket;private InputStream is;private OutputStream os;private HashMap<String,ServerThread> threads;public ServerThread(Socket socket,HashMap<String,ServerThread> threads){this.socket = socket;this.threads = threads;try {is = socket.getInputStream();os = socket.getOutputStream();}catch (IOException e){e.printStackTrace();}}public Socket getSocket(){return socket;}@Overridepublic void run() {String infor = socket.getInetAddress()+":"+socket.getPort();IOUtil.writeString(os,infor+"你好");while (true){//等待客户端发送消息String msg = IOUtil.readString(is);String response = infor+">"+msg;if(msg.split(" ")[0].equals("allClients")){String users = "";Iterator<String> iterator = threads.keySet().iterator();while (iterator.hasNext()){String user = iterator.next();if (!user.equals(infor)){users += "\n"+user;}}response += users;}else if (msg.split(" ")[0].equals("sto")){String user = msg.split(" ")[1];if (threads.keySet().contains(user)){Socket userSocket = threads.get(user).getSocket();try {OutputStream userSocketOutputStream = userSocket.getOutputStream();IOUtil.writeString(userSocketOutputStream,infor+"说:"+msg.split(" ")[2]);} catch (IOException e) {e.printStackTrace();}}else {response = response+"\n用户不存在";}}else if (msg.split(" ")[0].equals("sta")){Iterator<String> users = threads.keySet().iterator();while (users.hasNext()){String user = users.next();if (!user.equals(infor)){Socket userSocket = threads.get(user).getSocket();try {OutputStream userOs = userSocket.getOutputStream();IOUtil.writeString(userOs,infor+"说:"+msg.split(" ")[1]);} catch (IOException e) {e.printStackTrace();}}}}IOUtil.writeString(os,response);}}}

TestClient类

import javax.swing.*;import java.awt.*;import java.awt.event.ActionEvent;import java.awt.event.ActionListener;import java.io.IOException;import java.io.InputStream;import java.io.OutputStream;import .Socket;public class TestClient extends JFrame {private Socket socket;private InputStream is;private OutputStream os;private String response = "";private JTextArea jTextArea;public void init(){setSize(800,600);setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);setLocationRelativeTo(null);setTitle("Client");setLayout(null);setBackground(Color.WHITE);Font font = new Font(null,Font.BOLD,16);try {socket = new Socket("localhost",8888);is = socket.getInputStream();os = socket.getOutputStream();} catch (IOException e) {e.printStackTrace();}jTextArea = new JTextArea();jTextArea.setBounds(0,0,800,500);jTextArea.setFont(font);add(jTextArea);response = IOUtil.readString(is);jTextArea.setText(response);JTextField jTextField = new JTextField();jTextField.setFont(font);jTextField.setBounds(0,500,800,50);add(jTextField);jTextField.addActionListener(new ActionListener() {@Overridepublic void actionPerformed(ActionEvent e) {String text = jTextField.getText();jTextField.setText("");IOUtil.writeString(os,text);}});setVisible(true);recive();}public void recive(){while (true){String s = IOUtil.readString(is);response += "\n"+s;jTextArea.setText(response);}}public static void main(String[] args) {TestClient t = new TestClient();t.init();}}

IOUtil类

import java.io.IOException;import java.io.InputStream;import java.io.OutputStream;public class IOUtil {/*** 接收int类型数据* @param is 输入流* @return 输入流的前四位，整合为一个int数据*/public static int readInt(InputStream is) {int[] values = new int[4];try {for (int i = 0; i < 4; i++) {values[i] = is.read();}} catch (IOException e) {e.printStackTrace();}int value = values[0]<<24 | values[1]<<16 | values[2]<<8 | values[3]<<0;return value;}/*** 输出一个int类型的数据* @param os 输出流* @param value 要发送的int数据*/public static void writeInt(OutputStream os,int value) {int[] values = new int[4];values[0] = (value>>24)&0xFF;values[1] = (value>>16)&0xFF;values[2] = (value>>8)&0xFF;values[3] = (value>>0)&0xFF;try{for (int i = 0; i < 4; i++) {os.write(values[i]);}}catch (IOException e){e.printStackTrace();}}/*** 获取string输入* @param is 输入流* @return 输入的string*/public static String readString(InputStream is) {int len = readInt(is);byte[] sByte = new byte[len];try {is.read(sByte);} catch (IOException e) {e.printStackTrace();}String s = new String(sByte);return s;}/*** 发送string* @param os 输出流* @param s 要发送的字符串*/public static void writeString(OutputStream os,String s) {byte[] bytes = s.getBytes();int len = bytes.length;writeInt(os,len);try {os.write(bytes);} catch (IOException e) {e.printStackTrace();}}}