java网络编程——多线程数据收发并行
基本介绍与思路
收发并行
前一篇博客中，完成了客户端与服务端的简单TCP交互，但这种交互是触发式的：客户端发送一条消息，服务端收到后再回送一条。没有做到收发并行。收发并行的字面意思很容易理解，即数据的发送与接收互相不干扰，相互独立。当然，要保证服务端和客户端都能做到收发并行。
业务逻辑
脱离业务逻辑的实践是毫无意义的，先描述一下本实践中的业务逻辑：一个服务端接受多个客户端的连接，连接后，向各个客户端定时发送时间戳数据，同时在并行条件下，接受各个客户端发送来的数据并显示；客户端键盘输入字符串，发送给服务端，同时在并行条件下，接收服务器发来的时间戳数据并显示。
实现思路
实现发送与接收并行，思路其实非常直观，即建立两个线程，分别用来实现输入流和输出流。我的代码的设计方案如下图所示：
- 服务端：创建一个监听客户端连接的线程，线程中一旦接收到请求，创建一个对应该客户端收发处理的对象，对象中创建输入流线程，并使用单例线程池创建输出流线程。主线程使用键盘输入流System.in来进行阻塞。同时主线程中创建Timer定时器，定时向输出流发送数据。
- 客户端：主线程发送连接请求，与服务器建立连接。使用键盘输入流System.in来阻塞主线程，同时作为输出流使用；创建一个输入流线程，异步运行，接收服务器数据。
代码分析
源代码文件结构如下图所示
服务端
服务器端分为三个部分，分别是Server.java，TCPServer.java和ClientHandler.java
Server.java
package Server;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.TimerTask;
import java.util.Timer;
import java.util.Date;
public class Server {
private static SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd_HH:mm:ss");
public static void main(String[] args){
try {
TCPServer.accept();
new Timer("Timer").schedule(new TimerTask() {
@Override
public void run() {
TCPServer.broadcast(df.format(new Date()));
}
}, 1000,5000);
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
String str;
//因为ClientListen是异步线程，使用键盘输入流将主线程阻塞住，保证跟ClientListen线程同步，同时可控制ClientListen服务的退出
do{
str = bufferedReader.readLine();
}while (str.equalsIgnoreCase("serverExit"));
}catch (Exception e){
System.out.println("监听请求过程中异常退出");
}
try {
TCPServer.stop();
} catch (IOException e) {
System.out.println("关闭套接字过程中出现异常");
} finally {
System.out.println("服务器端套接字已关闭！");
}
}
}
TCPServer.java
package Server;
import java.io.IOException;
import java.net.*;
import java.util.ArrayList;
import java.util.UUID;
class TCPServer {
private static int LOCAL_PORT = 3001;
private static ClientListenHandle clientListenHandle;
private static ArrayList<ClientHandler> clientHandlerList = new ArrayList<ClientHandler>();
static void accept() throws IOException {
//创建服务器端套接字
ServerSocket serverSocket = createSocket();
InitSocket(serverSocket);
System.out.println("服务器准备就绪 addr: " + Inet4Address.getLocalHost() + " /port: " + LOCAL_PORT);
System.out.println("开始监听客户端连接...");
//创建线程监听客户端请求
clientListenHandle = new ClientListenHandle(serverSocket);
clientListenHandle.start();
}
static void stop() throws IOException {
for (ClientHandler clientHandler : clientHandlerList) {
clientHandler.socketClose();
}
clientHandlerList.clear();
clientListenHandle.exit();
}
private static ServerSocket createSocket() throws IOException {
ServerSocket socket = new ServerSocket(LOCAL_PORT, 50);
return socket;
}
private static void InitSocket(ServerSocket socket) throws SocketException {
// 是否复用未完全关闭的地址端口
socket.setReuseAddress(true);
// 等效Socket#setReceiveBufferSize
socket.setReceiveBufferSize(64 * 1024 * 1024);
// 设置serverSocket#accept超时时间，不设置即永久等待
// serverSocket.setSoTimeout(2000);
// 设置性能参数：短链接，延迟，带宽的相对重要性
socket.setPerformancePreferences(1, 1, 1);
}
static void broadcast(String msg) {
for (ClientHandler clientHandler : clientHandlerList) {
clientHandler.write(msg);
}
}
/**
* 监听客户端请求的线程
*/
static class ClientListenHandle extends Thread {
private final ServerSocket serverSocket;
private Boolean done = false;
ClientListenHandle(ServerSocket serverSocket) {
this.serverSocket = serverSocket;
}
@Override
public void run() {
super.run();
try {
do {
Socket client;
try {
client = serverSocket.accept();
} catch (Exception e) {
continue;//某一个客户端连接失败，要保证其它客户端能正常连接
}
String uuid = UUID.randomUUID().toString();//为客户端生成唯一标识
System.out.println("已接受连接client："+uuid+" /Addr:"+client.getInetAddress()+" /Port:"+client.getPort());
//为该客户端实例化一个ClientHandler对象，注入对象删除操作的lambda表达式
ClientHandler clientHandle = new ClientHandler(client, handler -> clientHandlerList.remove(handler), uuid);
clientHandle.read();
clientHandlerList.add(clientHandle);
} while (!done);
} catch (Exception e) {
if (!done) {
System.out.println("异常退出！");
}
}
}
void exit() throws IOException {
done = true;
serverSocket.close();
}
}
}
ClientHandler.java
package Server;
import java.io.*;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ClientHandler {
private final Socket client;
private final ReadHandler readHandler;
private final WriteHandle writeHandler;
private final Removable removable;
private final String uid;
ClientHandler(Socket socket, Removable removable, String uid) throws IOException {
this.client = socket;
this.readHandler = new ReadHandler(socket.getInputStream());
this.writeHandler = new WriteHandle(socket.getOutputStream());
this.removable = removable;
this.uid = uid;
}
void read() {
readHandler.start();
}
void write(String msg) {
System.out.println("Server -->> " + uid + " : " + msg);
writeHandler.write(msg);
}
/**
* 把输入输出流和套接字都关闭
*/
void socketClose(){
try {
readHandler.exit();
writeHandler.exit();
client.close();
} catch (IOException e) {
e.printStackTrace();
}finally {
System.out.println("客户端："+uid+" 套接字连接已关闭");
}
}
/**
* 把自身从对象列表中清除掉，具体方法是使用lambda表达式来注入的
*/
void removeClientHandler() {
removable.removeClientHandle(this);
}
/**
* 定义一个接口，接收lambda表达式
*/
interface Removable {
void removeClientHandle(ClientHandler clientHandler);
}
/**
* 输入流操作线程
*/
class ReadHandler extends Thread {
private final InputStream inputStream;
private Boolean flag = true;
ReadHandler(InputStream inputStream) {
this.inputStream = inputStream;
}
@Override
public void run() {
super.run();
BufferedReader socketInput = null;
try {
socketInput = new BufferedReader(new InputStreamReader(inputStream));
do {
String str = socketInput.readLine();
//不知道为什么，客户端关闭时，这里直接报异常，获取不到null
if (str.equalsIgnoreCase("exit")) {
System.out.println("已无法读取客户端数据！");
throw new Exception();
}
System.out.println(uid + " -->> server : " + str);
} while (flag);
} catch (Exception e) {
if (flag) {
System.out.println("读取客户端过程中异常退出");
ClientHandler.this.removeClientHandler();
ClientHandler.this.socketClose();
}
}
}
void exit() throws IOException {
flag = false;
inputStream.close();
}
}
/**
* 输出流操作线程，使用单例线程池，可以自动等待任务并处理，无需人工添加阻塞操作
*/
class WriteHandle {
private final OutputStream outputStream;
private final ExecutorService executorService;
WriteHandle(OutputStream outputStream) {
this.outputStream = outputStream;
this.executorService = Executors.newSingleThreadExecutor();
}
private void write(String msg){
executorService.execute(new WriteRunnable(msg,outputStream));
}
void exit() throws IOException{
outputStream.close();
executorService.shutdown();
}
class WriteRunnable implements Runnable{
private final String msg;
private final PrintStream printStream;
WriteRunnable(String msg, OutputStream outputStream) {
this.msg = msg;
this.printStream = new PrintStream(outputStream);
}
@Override
public void run() {
try {
printStream.println(msg);
} catch (Exception e) {
System.out.println("打印输出异常！");
}
}
}
}
}
客户端
Client.java
package Client;
import java.io.*;
import java.util.UUID;
import Client.bean.ServerInfo;
public class Client {
public static void main(String[] args)throws IOException {
ServerInfo serverInfo = new ServerInfo(UUID.randomUUID().toString(),"127.0.2.16",3001);
System.out.println("准备发起服务器连接...");
System.out.println("服务器信息：Addr:"+serverInfo.getAddress()+" /Port:"+serverInfo.getPort());
try {
TCPClient.connect(serverInfo);
}catch (Exception e){
System.out.println("连接失败，退出");
}
}
}
TCPClient.java
package Client;
import Client.bean.ServerInfo;
import java.io.*;
import java.net.*;
class TCPClient {
static void connect(ServerInfo serverInfo) throws IOException {
Socket clientSocket = createSocket();//建立套接字
InitSocket(clientSocket);//初始化套接字
//连接远程服务器
clientSocket.connect(new InetSocketAddress(serverInfo.getAddress(), serverInfo.getPort()), 3000);
System.out.println("已连接server");
try {
//输入流线程
ReadHandle readHandle = new ReadHandle(clientSocket.getInputStream());
readHandle.start();
//输出流
write(clientSocket);
//当输出流结束时，关闭输入流
readHandle.exit();
} catch (Exception e) {
System.out.println("出现异常！");
} finally {
clientSocket.close();
System.out.println("客户端结束");
}
}
private static Socket createSocket() throws IOException {
Socket socket = new Socket();
return socket;
}
private static void InitSocket(Socket socket) throws SocketException {
// 设置读取超时时间为2秒，超过2秒未获得数据时readline报超时异常；不设置即进行永久等待
//socket.setSoTimeout(2000);
// 是否复用未完全关闭的Socket地址，对于指定bind操作后的套接字有效
socket.setReuseAddress(true);
// 是否开启Nagle算法
socket.setTcpNoDelay(true);
// 是否需要在长时无数据响应时发送确认数据（类似心跳包），时间大约为2小时
socket.setKeepAlive(true);
// 对于close关闭操作行为进行怎样的处理；默认为false，0
// false、0：默认情况，关闭时立即返回，底层系统接管输出流，将缓冲区内的数据发送完成
// true、0：关闭时立即返回，缓冲区数据抛弃，直接发送RST结束命令到对方，并无需经过2MSL等待
// true、200：关闭时最长阻塞200毫秒，随后按第二情况处理
socket.setSoLinger(true, 20);
// 是否让紧急数据内敛，默认false；紧急数据通过 socket.sendUrgentData(1);发送
socket.setOOBInline(true);
// 设置接收发送缓冲器大小
socket.setReceiveBufferSize(64 * 1024 * 1024);
socket.setSendBufferSize(64 * 1024 * 1024);
// 设置性能参数：短链接，延迟，带宽的相对重要性
socket.setPerformancePreferences(1, 1, 1);
}
/**
* 输出流方法
*/
private static void write(Socket socket) throws IOException {
//构建键盘输入流
InputStream in = System.in;
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
//得到socket输出流并转化为打印流
OutputStream outputStream = socket.getOutputStream();
PrintStream printStream = new PrintStream(outputStream);
for(;;){
String str = bufferedReader.readLine();//从键盘输入获取内容
printStream.println(str);//通过打印流输出
if(str.equalsIgnoreCase("exit")){
break;
}
}
printStream.close();
System.out.println("输出流关闭");
}
/**
* 输入流线程
*/
static class ReadHandle extends Thread {
private final InputStream inputStream;
private Boolean done = false;
ReadHandle(InputStream inputStream) {
this.inputStream = inputStream;
}
@Override
public void run() {
super.run();
try {
//获取输入流
BufferedReader socketInput = new BufferedReader(new InputStreamReader(inputStream));
do {
String str;
str = socketInput.readLine();
if (str==null) {
break;
}
System.out.println("From server: "+ str);
} while (!done);
} catch (Exception e) {
if (!done) {
System.out.println("异常断开，或者输入异常");
}
}
}
void exit() {
done = true;
try {
inputStream.close();
} catch (IOException e) {
e.printStackTrace();
}finally {
System.out.println("输入流关闭");
}
}
}
}
关于代码的具体分析，由于代码已有很多注释，博文中便不再赘述。
运行结果
运行结果如下所示
-
服务端
连接成功后，服务端每隔5秒向各个客户端发送时间戳信息，同时接收两个客户端发来的信息
-
客户端1
输入“I am client1”并向服务端发送，同时接收服务端发来的时间戳信息
-
客户端2
输入“I am client2”并向服务端发送，同时接收服务端发来的时间戳信息
本篇博客记录一次实践学习，使用多线程+socket编程，实现了单服务器与多客户端之间的数据收发并行，除此之外，通过思维流程图，整理了代码的设计思路并展示出来。
正文到此结束
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
