最近看了Java的IO包源码，对BIO有了较深入的理解。Socket编程其实也是基于IO流操作，并且其流操作都是阻塞的，就想着写一个Socket程序并对其一步一步优化，来加深对IO的理解。本文主要从简单的Socket连接开始，一步一步优化，最后使用线程池等技术提高并发。Socket源码本篇未涉及，等有时间我再研究一番。
一. 基本概念
Socket编程的基本流程如下图（图片来自网络），一个IP地址和一个端口号称为一个套接字（socket）。
Socket
Socket编程是BIO的，对于服务端，accept()、read()、write()都会堵塞。
- accept是阻塞的，只有新连接来了，accept才会返回，主线程才能继续
- read是阻塞的，只有请求消息来了，read才能返回，子线程才能继续处理
- write是阻塞的，只有客户端把消息收了，write才能返回，子线程才能继续读取下一个请求
Socket开发Java提供了两个类，Socket用于BIO连接和信息收发，ServerSocket用于构建一个服务端，其accept()方法获得一个Socket对象，最终客户端服务器都是使用Socket进行通信。
二. 最基本的Socket
如下，最基本的客户端发送消息，服务端接收消息输入。需要注意的是，由于中文的utf8编码是3个字节，如果使用buffer来分段接收字节流，可能导致乱码。另外，read()是堵塞的，如果不判断read() == -1来表示结束，那么read()方法会一直堵塞。
package me.zebin.demo.javaio;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.*;
@RunWith(SpringRunner.class)
@SpringBootTest
public class JavaioApplicationTests {
@Test
public void server() throws Exception {
// 指定端口
ServerSocket ss = new ServerSocket(9999);
System.out.println("server starting...");
// 等待连接
Socket s = ss.accept();
// 获取输入流，接收客户端的消息
InputStream is = s.getInputStream();
// 缓存buffer，utf8编码中文是3个字节，这里也可是使用BufferedReader解码
byte[] buffer = new byte[5];
while(true){
int cnt = is.read(buffer);
// 如果不判断流结束，上面的read()读不到数据会一直堵塞
if(cnt == -1){
break;
}
String str = new String(buffer, 0, cnt, "utf8");
System.out.println(str);
}
s.close();
ss.close();
}
@Test
public void client() throws Exception{
// 指定端口
Socket s = new Socket("127.0.0.1", 9999);
// 获取输出流，向服务端发消息
OutputStream os = s.getOutputStream();
// 发送消息，utf8编码中文是3个字节，服务端使用buffer可能导致乱码
String str = "我是客户端";
os.write(str.getBytes("utf8"));
s.close();
}
}
以上程序，如果buffer设置为5，运行结果如下，出现乱码。
乱码
当然，解决方案可以整行读取，将InputStream转为Reader再转BufferedReader即可读取一行。也可使用Scanner来解决，服务端代码改为如下：
@Test
public void server() throws Exception {
// 指定端口
ServerSocket ss = new ServerSocket(9999);
System.out.println("server starting...");
// 等待连接
Socket s = ss.accept();
// 获取输入流，接收客户端的消息
InputStream is = s.getInputStream();
// 输入字节流封装为Scanner，读取整行
Scanner sc = new Scanner(is, "utf8");
while (sc.hasNextLine()){
System.out.println(sc.nextLine());
}
s.close();
ss.close();
}
运行结果如下，没有乱码了。
Scanner
服务端判断流关闭，一般使用两种方法。
- 使用特殊符号：既然上面可以获取到行，服务端客户端就可以约定相关的结束符，如接收到一个空行就结束，服务端进行判断关闭流即可。
- 使用长度界定：类似http协议就有content-length界定结束符，我们也可以在客户端发送byte[]数组前，在byte[]数据前两个字节标识消息长度。当然，两个字节能表示的消息长度就只有2^16-1，即大小是2^16字节，即64k大小。
三. 多线程版本
上面的版本有一个弊端，就是一个服务器只能提供给一个客户端进行连接，如果将连接的用线程处理，服务器可以处理更多的客户端连接，代码如下：
@Test
public void server() throws Exception {
// 指定端口
ServerSocket ss = new ServerSocket(9998);
System.out.println("server starting...");
while(true){
// 等待连接
Socket s = ss.accept();
System.out.println("获得连接");
Thread t = new Thread(new ServerThread(s));
t.start();
}
}
class ServerThread implements Runnable{
private Socket s;
ServerThread(Socket s){
this.s = s;
}
@Override
public void run(){
// 获取输入流，接收客户端的消息
InputStream is = null;
try {
is = s.getInputStream();
// 使用Scanner封装
Scanner sc = new Scanner(is, "utf8");
while (sc.hasNextLine()){
System.out.println(sc.nextLine());
}
s.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
四. 线程池版本
以上多线程版本我们使用了多线程来处理并发，不过线程的创建和销毁都会消耗大量的资源和时间，同时，高并发下会创建非常多的线程，且不说操作系统能开启的线程数有限，操作系统维护和切换大量的线程也会非常耗时。所以使用线程池，只用4个线程，用队列将未执行到的线程排队处理，减少了线程数量，同时也避免了创建和销毁线程带来的性能问题。
@Test
public void server() throws Exception {
// 指定端口
ServerSocket ss = new ServerSocket(9998);
System.out.println("server starting...");
// 创建线程队列
BlockingQueue bq = new ArrayBlockingQueue(100);
// 拒绝策略
RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
Executor executor = new ThreadPoolExecutor(4, 8, 1, TimeUnit.MINUTES, bq, handler);
while(true){
// 等待连接
Socket s = ss.accept();
System.out.println("获得连接");
Thread t = new Thread(new ServerThread(s));
executor.execute(t);
}
}
以上，本篇结束。
参考资料
- 通俗地讲，Netty 能做什么？ - 知乎用户的回答 - 知乎
https://www.zhihu.com/question/24322387/answer/282001188 - https://juejin.im/post/5ad9dd61518825671c0e1d71
网友评论