?
前言
本文從操作系統(tǒng)實際調(diào)用角度(以CentOS Linux release 7.5操作系統(tǒng)為示例)，力求追根溯源看IO的每一步操作到底發(fā)生了什么。
關(guān)于如何查看系統(tǒng)調(diào)用，Linux可以使用 strace 來查看任何軟件的系統(tǒng)調(diào)動(這是個很好的分析學習方法)：strace -ff -o ./out java TestJava
一 BIO
/**
* Alipay.com Inc. Copyright (c) 2004-2020 All Rights Reserved.
*/
package io;
import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
/**
* @author xiangyong.ding
* @version $Id: TestSocket.java, v 0.1 2020年08月02日 20:56 xiangyong.ding Exp $
*/
public class BIOSocket {
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(8090);
System.out.println("step1: new ServerSocket ");
while (true) {
Socket client = serverSocket.accept();
System.out.println("step2: client\t" + client.getPort());
new Thread(() -> {
try {
InputStream in = client.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
while (true) {
System.out.println(reader.readLine());
}
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
}
1 發(fā)生的系統(tǒng)調(diào)用
啟動時
socket(AF_INET, SOCK_STREAM, IPPROTO_IP) = 5
bind(5, {sa_family=AF_INET, sin_port=htons(8090), sin_addr=inet_addr("0.0.0.0")}, 16) = 0
listen(5, 50) = 0
poll([{fd=5, events=POLLIN|POLLERR}], 1, -1) = 1 ([{fd=5, revents=POLLIN}])
poll函數(shù)會阻塞直到其中任何一個fd發(fā)生事件。
有客戶端連接后
accept(5, {sa_family=AF_INET, sin_port=htons(10253), sin_addr=inet_addr("42.120.74.252")}, [16]) = 6
clone(child_stack=0x7f013e5c4fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7f013e5c59d0, tls=0x7f013e5c5700, child_tidptr=0x7f013e5c59d0) = 13168
poll([{fd=5, events=POLLIN|POLLERR}], 1, -1拋出線程(即我們代碼里的 new Thread() )后，繼續(xù)poll阻塞等待連接。
clone出來的線程
recvfrom(6, "hello,bio\n", 8192, 0, NULL, NULL) =
關(guān)于對recvfrom函數(shù)的說明，其中第四個參數(shù)0 表示這是一個阻塞調(diào)用。
客戶端發(fā)送數(shù)據(jù)后
recvfrom(6, "hello,bio\n", 8192, 0, NULL, NULL) = 10
2 優(yōu)缺點
優(yōu)點
代碼簡單，邏輯清晰。
缺點
- 由于stream的read操作是阻塞讀，面對多個連接時 每個連接需要每線程。無法處理大量連接(C10K問題)。
- 誤區(qū)：可見JDK1.8中對于最初的BIO，在Linux OS下仍然使用的poll，poll本身也是相對比較高效的多路復用函數(shù)(支持非阻塞、多個socket同時檢查event)，只是限于JDK最初的stream API限制，無法支持非阻塞讀取。
二 NIO(non block)
改進：使用NIO API，將阻塞變?yōu)榉亲枞?不需要大量線程。
/**
* Alipay.com Inc. Copyright (c) 2004-2020 All Rights Reserved.
*/
package io;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
/**
* @author xiangyong.ding
* @version $Id: NioSocket.java, v 0.1 2020年08月09日 11:25 xiangyong.ding Exp $
*/
public class NIOSocket {
private static LinkedList<SocketChannel> clients = new LinkedList<>();
private static void startClientChannelHandleThread(){
new Thread(() -> {
while (true){
ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
//處理客戶端連接
for (SocketChannel c : clients) {
// 非阻塞, >0 表示讀取到的字節(jié)數(shù)量, 0或-1表示未讀取到或讀取異常
int num = 0;
try {
num = c.read(buffer);
} catch (IOException e) {
e.printStackTrace();
}
if (num > 0) {
buffer.flip();
byte[] clientBytes = new byte[buffer.limit()];
//從緩沖區(qū) 讀取到內(nèi)存中
buffer.get(clientBytes);
System.out.println(c.socket().getPort() + ":" + new String(clientBytes));
//清空緩沖區(qū)
buffer.clear();
}
}
}
}).start();
}
public static void main(String[] args) throws IOException {
//new socket,開啟監(jiān)聽
ServerSocketChannel socketChannel = ServerSocketChannel.open();
socketChannel.bind(new InetSocketAddress(9090));
//設(shè)置阻塞接受客戶端連接
socketChannel.configureBlocking(true);
//開始client處理線程
startClientChannelHandleThread();
while (true) {
//接受客戶端連接; 非阻塞，無客戶端返回null(操作系統(tǒng)返回-1)
SocketChannel client = socketChannel.accept();
if (client == null) {
//System.out.println("no client");
} else {
//設(shè)置讀非阻塞
client.configureBlocking(false);
int port = client.socket().getPort();
System.out.println("client port :" + port);
clients.add(client);
}
}
}
}
1 發(fā)生的系統(tǒng)調(diào)用
主線程
socket(AF_INET, SOCK_STREAM, IPPROTO_IP) = 4
bind(4, {sa_family=AF_INET, sin_port=htons(9090), sin_addr=inet_addr("0.0.0.0")}, 16) = 0
listen(4, 50) = 0
fcntl(4, F_SETFL, O_RDWR|O_NONBLOCK) = 0
accept(4, 0x7fe26414e680, 0x7fe26c376710) = -1 EAGAIN (Resource temporarily unavailable)
有連接后，子線程
read(6, 0x7f3f415b1c50, 4096) = -1 EAGAIN (Resource temporarily unavailable)
read(6, 0x7f3f415b1c50, 4096) = -1 EAGAIN (Resource temporarily unavailable)
...
資源使用情況：
?
??
??
2 優(yōu)缺點
優(yōu)點
線程數(shù)大大減少。
缺點
需要程序自己掃描 每個連接read，需要 O(n)時間復雜度系統(tǒng)調(diào)用 (此時可能只有一個連接發(fā)送了數(shù)據(jù))，高頻系統(tǒng)調(diào)用(導致CPU 用戶態(tài)內(nèi)核態(tài)切換)高。導致CPU消耗很高。
三 多路復用器(select、poll、epoll)
改進：不需要用戶掃描所有連接，由kernel 給出哪些連接有數(shù)據(jù)，然后應(yīng)用從有數(shù)據(jù)的連接讀取數(shù)據(jù)。
1 epoll
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Set;
/**
* 多路復用socket
*
* @author xiangyong.ding
* @version $Id: MultiplexingSocket.java, v 0.1 2020年08月09日 12:19 xiangyong.ding Exp $
*/
public class MultiplexingSocket {
static ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
public static void main(String[] args) throws Exception {
LinkedList<SocketChannel> clients = new LinkedList<>();
//1.啟動server
//new socket,開啟監(jiān)聽
ServerSocketChannel socketChannel = ServerSocketChannel.open();
socketChannel.bind(new InetSocketAddress(9090));
//設(shè)置非阻塞，接受客戶端
socketChannel.configureBlocking(false);
//多路復用器（JDK包裝的代理，select /poll/epoll/kqueue）
Selector selector = Selector.open(); //java自動代理，默認為epoll
//Selector selector = PollSelectorProvider.provider().openSelector();//指定為poll
//將服務(wù)端socket 注冊到 多路復用器
socketChannel.register(selector, SelectionKey.OP_ACCEPT);
//2. 輪訓多路復用器
// 先詢問有沒有連接,如果有則返回數(shù)量以及對應(yīng)的對象(fd)
while (selector.select() > 0) {
System.out.println();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iter = selectionKeys.iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
//2.1 處理新的連接
if (key.isAcceptable()) {
//接受客戶端連接; 非阻塞，無客戶端返回null(操作系統(tǒng)返回-1)
SocketChannel client = socketChannel.accept();
//設(shè)置讀非阻塞
client.configureBlocking(false);
//同樣，把client也注冊到selector
client.register(selector, SelectionKey.OP_READ);
System.out.println("new client : " + client.getRemoteAddress());
}
//2.2 處理讀取數(shù)據(jù)
else if (key.isReadable()) {
readDataFromSocket(key);
}
}
}
}
protected static void readDataFromSocket(SelectionKey key) throws Exception {
SocketChannel socketChannel = (SocketChannel) key.channel();
// 非阻塞, >0 表示讀取到的字節(jié)數(shù)量, 0或-1表示未讀取到或讀取異常
// 請注意：這個例子降低復雜度，不考慮報文大于buffer size的情況
int num = socketChannel.read(buffer);
if (num > 0) {
buffer.flip();
byte[] clientBytes = new byte[buffer.limit()];
//從緩沖區(qū) 讀取到內(nèi)存中
buffer.get(clientBytes);
System.out.println(socketChannel.socket().getPort() + ":" + new String(clientBytes));
//清空緩沖區(qū)
buffer.clear();
}
}
}
2 發(fā)生的系統(tǒng)調(diào)用
啟動
socket(AF_INET, SOCK_STREAM, IPPROTO_IP) = 4
bind(4, {sa_family=AF_INET, sin_port=htons(9090), sin_addr=inet_addr("0.0.0.0")}, 16) = 0
listen(4, 50)
fcntl(4, F_SETFL, O_RDWR|O_NONBLOCK) = 0
epoll_create(256) = 7
epoll_ctl(7, EPOLL_CTL_ADD, 5, {EPOLLIN, {u32=5, u64=4324783852322029573}}) = 0
epoll_ctl(7, EPOLL_CTL_ADD, 4, {EPOLLIN, {u32=4, u64=158913789956}}) = 0
epoll_wait(7
關(guān)于對epoll_create(對應(yīng)著Java的 Selector selector = Selector.open()) 的說明，本質(zhì)上是在內(nèi)存的操作系統(tǒng)保留區(qū)，創(chuàng)建一個epoll數(shù)據(jù)結(jié)構(gòu)。用于后面當有client連接時，向該epoll區(qū)中添加監(jiān)聽。
有連接
epoll_wait(7,[{EPOLLIN, {u32=4, u64=158913789956}}], 8192, -1) = 1
accept(4, {sa_family=AF_INET, sin_port=htons(29597), sin_addr=inet_addr("42.120.74.252")}, [16]) = 8
fcntl(8, F_SETFL, O_RDWR|O_NONBLOCK) = 0
epoll_ctl(7, EPOLL_CTL_ADD, 8, {EPOLLIN, {u32=8, u64=3212844375897800712}}) = 0關(guān)于epoll_ctl (對應(yīng)著Java的 client.register(selector, SelectionKey.OP_READ) )。其中 EPOLLIN 恰好對應(yīng)著Java的 SelectionKey.OP_READ 即監(jiān)聽數(shù)據(jù)到達讀取事件。
客戶端發(fā)送數(shù)據(jù)
epoll_wait(7,[{EPOLLIN, {u32=8, u64=3212844375897800712}}], 8192, -1) = 1
read(8, "hello,multiplex\n", 4096) = 16
epoll_wait(7,note：epoll_wait第四個參數(shù)-1表示block。
poll 和 epoll 對比
根據(jù)“1.BIO”中的poll函數(shù)調(diào)用和epoll函數(shù)對比如下：
?
??
??
?
??
??
poll和epoll本質(zhì)上都是同步IO， 區(qū)別于BIO的是 多路復用充分降低了 system call，而epoll更進一步，再次降低了system call的時間復雜度。
3 優(yōu)缺點
優(yōu)點
- 線程數(shù)同樣很少，甚至可以把acceptor線程和worker線程使用同一個。
- 時間復雜度低，Java實現(xiàn)的Selector(在Linux OS下使用的epoll函數(shù))支持多個clientChannel事件的一次性獲取，且時間復雜度維持在O(1)。
- CPU使用低：得益于Selector，我們不用向 “2.NIO”中需要自己一個個ClientChannel手動去檢查事件，因此使得CPU使用率大大降低。
缺點
- 數(shù)據(jù)處理麻煩：目前socketChannel.read 讀取數(shù)據(jù)完全是基于字節(jié)的，當我們需要需要作為HTTP服務(wù)網(wǎng)關(guān)時，對于HTTP協(xié)議的處理完全需要自己解析，這是個龐大、煩雜、容易出錯的工作。
- 性能
- 現(xiàn)有socket數(shù)據(jù)的讀取(socketChannel.read(buffer))全部通過一個buffer 緩沖區(qū)來接受，一旦連接多起來，這無疑是一個單線程讀取，性能無疑是個問題。
- 那么此時buffer我們每次讀取都重新new出來呢?如果每次都new出來，這樣的內(nèi)存碎片對于GC無疑是一場災(zāi)難。如何平衡地協(xié)調(diào)好buffer的共享，既保證性能，又保證線程安全，這是個難題。
四 Netty
1 研究的目標源碼(netty提供的入門example)
TelnetServer
package telnet;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
/**
* Simplistic telnet server.
*/
public final class TelnetServer {
static final boolean SSL = System.getProperty("ssl") != null;
static final int PORT = Integer.parseInt(System.getProperty("port", SSL? "8992" : "8023"));
public static void main(String[] args) throws Exception {
// Configure SSL.
final SslContext sslCtx;
if (SSL) {
SelfSignedCertificate ssc = new SelfSignedCertificate();
sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
} else {
sslCtx = null;
}
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new TelnetServerInitializer(sslCtx));
b.bind(PORT).sync().channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
TelnetServerHandler
package telnet;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.net.InetAddress;
import java.util.Date;
/**
* Handles a server-side channel.
*/
@Sharable
public class TelnetServerHandler extends SimpleChannelInboundHandler<String> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// Send greeting for a new connection.
ctx.write("Welcome to " + InetAddress.getLocalHost().getHostName() + "!\r\n");
ctx.write("It is " + new Date() + " now.\r\n");
ctx.flush();
}
@Override
public void channelRead0(ChannelHandlerContext ctx, String request) throws Exception {
// Generate and write a response.
String response;
boolean close = false;
if (request.isEmpty()) {
response = "Please type something.\r\n";
} else if ("bye".equals(request.toLowerCase())) {
response = "Have a good day!\r\n";
close = true;
} else {
response = "Did you say '" + request + "'?\r\n";
}
// We do not need to write a ChannelBuffer here.
// We know the encoder inserted at TelnetPipelineFactory will do the conversion.
ChannelFuture future = ctx.write(response);
// Close the connection after sending 'Have a good day!'
// if the client has sent 'bye'.
if (close) {
future.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
TelnetServerInitializer
package telnet;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.ssl.SslContext;
/**
* Creates a newly configured {@link ChannelPipeline} for a new channel.
*/
public class TelnetServerInitializer extends ChannelInitializer<SocketChannel> {
private static final StringDecoder DECODER = new StringDecoder();
private static final StringEncoder ENCODER = new StringEncoder();
private static final TelnetServerHandler SERVER_HANDLER = new TelnetServerHandler();
private final SslContext sslCtx;
public TelnetServerInitializer(SslContext sslCtx) {
this.sslCtx = sslCtx;
}
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
if (sslCtx != null) {
pipeline.addLast(sslCtx.newHandler(ch.alloc()));
}
// Add the text line codec combination first,
pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// the encoder and decoder are static as these are sharable
pipeline.addLast(DECODER);
pipeline.addLast(ENCODER);
// and then business logic.
pipeline.addLast(SERVER_HANDLER);
}
}
2 啟動后的系統(tǒng)調(diào)用
主線程(23109)
## 256無實際作用，這里只為了兼容舊版kernel api
epoll_create(256) = 7epoll_ctl(7, EPOLL_CTL_ADD, 5, {EPOLLIN, {u32=5, u64=5477705356928876549}}) = 0
epoll_create(256) = 10epoll_ctl(10, EPOLL_CTL_ADD, 8, {EPOLLIN, {u32=8, u64=17041805914081853448}}) = 0
epoll_create(256) = 13
epoll_ctl(13, EPOLL_CTL_ADD, 11, {EPOLLIN, {u32=11, u64=17042151607409573899}}) = 0
epoll_create(256) = 16
epoll_ctl(16, EPOLL_CTL_ADD, 14, {EPOLLIN, {u32=14, u64=17042497300737294350}}) = 0
epoll_create(256) = 19
epoll_ctl(19, EPOLL_CTL_ADD, 17, {EPOLLIN, {u32=17, u64=17042561450368827409}}) = 0
epoll_create(256) = 10
socket(AF_INET, SOCK_STREAM, IPPROTO_IP) = 20
clone(child_stack=0x7fc3c509afb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7fc3c509b9d0, tls=0x7fc3c509b700, child_tidptr=0x7fc3c509b9d0) = 23130
概括為：
- 向OS新建socket，并開啟clone boss線程23130。
- 為BOSS創(chuàng)建了一個epoll(論證參見下面“boss”)，每個worker創(chuàng)建一個epoll數(shù)據(jù)結(jié)構(gòu)(本質(zhì)上是在kernel內(nèi)存區(qū)創(chuàng)建了一個數(shù)據(jù)結(jié)構(gòu)，用于后續(xù)監(jiān)聽)。
- 創(chuàng)建boss線程監(jiān)聽的socket(本質(zhì)上在kernel中創(chuàng)建一個數(shù)據(jù)結(jié)構(gòu))。
boss(23130)
bind(20, {sa_family=AF_INET, sin_port=htons(8023), sin_addr=inet_addr("0.0.0.0")}, 16) = 0
listen(20, 128) = 0
getsockname(20, {sa_family=AF_INET, sin_port=htons(8023), sin_addr=inet_addr("0.0.0.0")}, [16]) = 0
getsockname(20, {sa_family=AF_INET, sin_port=htons(8023), sin_addr=inet_addr("0.0.0.0")}, [16]) = 0
##將fd為7號epoll和fd為20號的socket綁定，事件：epoll_ctl_add和epoll_ctl_mod
epoll_ctl(7, EPOLL_CTL_ADD, 20, {EPOLLIN, {u32=20, u64=14198059139132817428}}) = 0
epoll_ctl(7, EPOLL_CTL_MOD, 20, {EPOLLIN, {u32=20, u64=20}}) = 0
epoll_wait(7, [{EPOLLIN, {u32=5, u64=17295150779149058053}}], 8192, 1000) = 1
epoll_wait(7, [], 8192, 1000) = 0(不斷輪訓，1S超時一次)概括為：
- 將上一步中main線程創(chuàng)建的fd：20綁定端口8023，并開啟監(jiān)聽(網(wǎng)卡負責監(jiān)聽和接受連接和數(shù)據(jù)，kernel則負責路由到具體進程，具體參見：關(guān)于socket和bind和listen，TODO )。
- 將7號socket對應(yīng)的fd綁定到20號對應(yīng)的epoll數(shù)據(jù)結(jié)構(gòu)上去(都是操作kernel中的內(nèi)存)。
- 開始1S中一次阻塞等待epoll有任何連接或數(shù)據(jù)到達。
3 客戶端連接
boss (23130)
accept(20, {sa_family=AF_INET, sin_port=htons(11144), sin_addr=inet_addr("42.120.74.122")}, [16]) = 24
getsockname(24, {sa_family=AF_INET, sin_port=htons(8023), sin_addr=inet_addr("192.168.0.120")}, [16]) = 0
getsockname(24, {sa_family=AF_INET, sin_port=htons(8023), sin_addr=inet_addr("192.168.0.120")}, [16]) = 0
setsockopt(24, SOL_TCP, TCP_NODELAY, [1], 4) = 0
getsockopt(24, SOL_SOCKET, SO_SNDBUF, [87040], [4]) = 0
getsockopt(24, SOL_SOCKET, SO_SNDBUF, [87040], [4]) = 0
##拋出 work線程
clone(child_stack=0x7fc3c4c98fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7fc3c4c999d0, tls=0x7fc3c4c99700, child_tidptr=0x7fc3c4c999d0) = 2301worker (2301)
writev(24, [{"Welcome to iZbp14e1g9ztpshfrla9m"..., 37}, {"It is Sun Aug 23 15:44:14 CST 20"..., 41}], 2) = 78
epoll_ctl(13, EPOLL_CTL_ADD, 24, {EPOLLIN, {u32=24, u64=24}}) = 0
epoll_ctl(13, EPOLL_CTL_MOD, 24, {EPOLLIN, {u32=24, u64=14180008216221450264}}) = 0
epoll_wait(13, [{EPOLLIN, {u32=11, u64=17042151607409573899}}], 8192, 1000) = 1
read(11, "\1", 128) = 1
##開始無限loop
epoll_wait(13, [], 8192, 1000) = 0
epoll_wait(13, [{EPOLLIN, {u32=24, u64=24}}], 8192, 1000) = 1概括：
- 當BOSS輪訓epoll_wait等到了連接后，首先accept得到該socket對應(yīng)的fd。
- 連接建立后 BOSS立馬拋出一個線程(clone函數(shù))。
- worker(即新建的線程)寫入了一段數(shù)據(jù)(這里是業(yè)務(wù)邏輯)。
- worker將該client對應(yīng)的fd綁定到了13號epoll上。
- worker繼續(xù)輪訓監(jiān)聽13號epoll。
4 客戶端主動發(fā)送數(shù)據(jù)
worker(2301)
read(24, "i am daojian\r\n", 1024) = 14
write(24, "Did you say 'i am daojian'?\r\n", 29) = 29
##繼續(xù)無限loop
epoll_wait(13, [], 8192, 1000) = 0
概括為：
- wait到數(shù)據(jù)后，立即read到用戶控件內(nèi)存中(讀取1024個字節(jié)到 用戶控件某個buff中)。
- 寫入數(shù)據(jù)(業(yè)務(wù)邏輯，不必太關(guān)注)。
- 繼續(xù)輪訓等待13號epoll。
5 客戶端發(fā)送bye報文，服務(wù)器斷開TCP連接
worker(2301)
read(24, "bye\r\n", 1024) = 5
write(24, "Have a good day!\r\n", 18) = 18
getsockopt(24, SOL_SOCKET, SO_LINGER, {onoff=0, linger=0}, [8]) = 0
dup2(25, 24) = 24
##從epoll數(shù)據(jù)結(jié)構(gòu)中（OS）中刪除fd為24的socket
epoll_ctl(13, EPOLL_CTL_DEL, 24, 0x7f702dd531e0) = -1 ENOENT
##關(guān)閉24 socket
close(24) = 0
##繼續(xù)等待13 epoll數(shù)據(jù)
epoll_wait(13, [], 8192, 1000) = 0
斷開客戶端連接概括為：
- 從epoll中刪除該客戶端對應(yīng)的fd(這里觸發(fā)源頭沒找到，可能是boss)。
- close關(guān)閉客戶端24號fd。
- 繼續(xù)輪訓epoll。
6 五個客戶端同時連接
boss線程(23130)
accept(20, {sa_family=AF_INET, sin_port=htons(1846), sin_addr=inet_addr("42.120.74.122")}, [16]) = 24
clone(child_stack=0x7f702cc51fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7f702cc529d0, tls=0x7f702cc52700, child_tidptr=0x7f702cc529d0) = 10035
accept(20, {sa_family=AF_INET, sin_port=htons(42067), sin_addr=inet_addr("42.120.74.122")}, [16]) = 26
clone(child_stack=0x7f702cb50fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7f702cb519d0, tls=0x7f702cb51700, child_tidptr=0x7f702cb519d0) = 10067
...woker線程(10035，第一個連接)
epoll_ctl(13, EPOLL_CTL_ADD, 24, {EPOLLIN, {u32=24, u64=24}}) = 0
epoll_ctl(13, EPOLL_CTL_MOD, 24, {EPOLLIN, {u32=24, u64=3226004877247250456}}) = 0
epoll_wait(13, [{EPOLLIN, {u32=11, u64=17042151607409573899}}], 8192, 1000) = 1 = 1
epoll_wait(13, [], 8192, 1000) = 0worker線程(10067，第二個連接)
epoll_ctl(16, EPOLL_CTL_ADD, 26, {EPOLLIN, {u32=26, u64=26}}) = 0
epoll_ctl(16, EPOLL_CTL_MOD, 26, {EPOLLIN, {u32=26, u64=3221483685433835546}}) = 0
epoll_wait(16, [{EPOLLIN, {u32=14, u64=17042497300737294350}}], 8192, 1000) = 1
epoll_wait(16, [], 8192, 1000) = 0
epoll_wait(16, [], 8192, 1000) = 0worker線程(10067，第二個連接)
epoll_ctl(19, EPOLL_CTL_ADD, 27, {EPOLLIN, {u32=27, u64=27}}) = 0
epoll_ctl(19, EPOLL_CTL_MOD, 27, {EPOLLIN, {u32=27, u64=3216966479350071323}}) = 0worker線程(8055，第四個連接)
epoll_ctl(10, EPOLL_CTL_ADD, 28, {EPOLLIN, {u32=28, u64=28}}) = 0
epoll_ctl(10, EPOLL_CTL_MOD, 28, {EPOLLIN, {u32=28, u64=3302604828697427996}}) = 0worker線程(10035，第五個連接，不在clone線程，而是復用了第一個epoll對應(yīng)的worker)
epoll_ctl(13, EPOLL_CTL_ADD, 29, {EPOLLIN, {u32=29, u64=29}}) = 0
epoll_ctl(13, EPOLL_CTL_MOD, 29, {EPOLLIN, {u32=29, u64=29}}) = 0概括為：
- epoll和boss、worker之間的關(guān)系：一共有4個worker對應(yīng)著4個epoll對象，boss和每個worker都有對應(yīng)自己的epoll。
- boss根據(jù)epoll數(shù)量，平衡分配連接到每個worker對應(yīng)的epoll中。
7 總結(jié)
下圖通過對系統(tǒng)調(diào)用的調(diào)查得出 netty 和 kernel 交互圖：
?
??
??
初始化直接創(chuàng)建5個epoll，其中7號為boss使用，專門用于處理和客戶端連接;其余4個用來給worker使用，用戶處理和客戶端的數(shù)據(jù)交互。
work的線程數(shù)量，取決于初始化時創(chuàng)建了幾個epoll，worker的復用本質(zhì)上是epoll的復用。
work之間為什么要獨立使用epoll?為什么不共享?
- 為了避免各個worker之間發(fā)生爭搶連接處理，netty直接做了物理隔離，避免競爭。各個worker只負責處理自己管理的連接，并且后續(xù)該worker中的每個client的讀寫操作完全由 該線程單獨處理，天然避免了資源競爭，避免了鎖。
- worker單線程，性能考慮：worker不僅僅要epoll_wait，還是處理read、write邏輯，加入worker處理了過多的連接，勢必造成這部分消耗時間片過多，來不及處理更多連接，性能下降。
8 優(yōu)缺點
優(yōu)點
- 數(shù)據(jù)處理：netty提供了大量成熟的數(shù)據(jù)處理組件(ENCODER、DECODER)，HTTP、POP3拿來即用。
- 編碼復雜度、可維護性：netty充分使得業(yè)務(wù)邏輯與網(wǎng)絡(luò)處理解耦，只需要少量的BootStrap配置即可，更多的集中在業(yè)務(wù)邏輯處理上。
- 性能：netty提供了的ByteBuf(底層Java原生的ByteBuffer)，提供了池化的ByteBuf，兼顧讀取性能和ByteBuf內(nèi)存分配(在后續(xù)文檔中會再做詳解)。
缺點
五 AIO
1 啟動
main線程
epoll_create(256) = 5
epoll_ctl(5, EPOLL_CTL_ADD, 6, {EPOLLIN, {u32=6, u64=11590018039084482566}}) = 0
##創(chuàng)建BOSS 線程(Proactor)
clone(child_stack=0x7f340ac06fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7f340ac079d0, tls=0x7f340ac07700, child_tidptr=0x7f340ac079d0) = 22704
socket(AF_INET6, SOCK_STREAM, IPPROTO_IP) = 8
setsockopt(8, SOL_IPV6, IPV6_V6ONLY, [0], 4) = 0
setsockopt(8, SOL_SOCKET, SO_REUSEADDR, [1], 4) = 0
bind(8, {sa_family=AF_INET6, sin6_port=htons(9090), inet_pton(AF_INET6, "::", &sin6_addr), sin6_flowinfo=0, sin6_scope_id=0}, 28) = 0
listen(8, 50)
accept(8, 0x7f67d01b3120, 0x7f67d9246690) = -1
epoll_ctl(5, EPOLL_CTL_MOD, 8, {EPOLLIN|EPOLLONESHOT, {u32=8, u64=15380749440025362440}}) = -1 ENOENT (No such file or directory)
epoll_ctl(5, EPOLL_CTL_ADD, 8, {EPOLLIN|EPOLLONESHOT, {u32=8, u64=15380749440025362440}}) = 0
read(0,
22704(BOSS 線程(Proactor))
epoll_wait(5, <unfinished ...>
2 請求連接
22704(BOSS 線程(Proactor))處理連接
epoll_wait(5,[{EPOLLIN, {u32=9, u64=4294967305}}], 512, -1) = 1
accept(8, {sa_family=AF_INET6, sin6_port=htons(55320), inet_pton(AF_INET6, "::ffff:36.24.32.140", &sin6_addr), sin6_flowinfo=0, sin6_scope_id=0}, [28]) = 9
clone(child_stack=0x7ff35c99ffb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7ff35c9a09d0, tls=0x7ff35c9a0700, child_tidptr=0x7ff35c9a09d0) = 26241
epoll_wait(5, <unfinished ...>26241
#將client 連接的FD加入到BOSS的epoll中，以便BOSS線程監(jiān)聽網(wǎng)絡(luò)事件
epoll_ctl(5, EPOLL_CTL_MOD, 9, {EPOLLIN|EPOLLONESHOT, {u32=9, u64=4398046511113}}) = -1 ENOENT (No such file or directory)
epoll_ctl(5, EPOLL_CTL_ADD, 9, {EPOLLIN|EPOLLONESHOT, {u32=9, u64=4398046511113}}) = 0
accept(8, 0x7ff3440008c0, 0x7ff35c99f4d0) = -1 EAGAIN (Resource temporarily unavailable)
epoll_ctl(5, EPOLL_CTL_MOD, 8, {EPOLLIN|EPOLLONESHOT, {u32=8, u64=8}}) = 0
3 客戶端發(fā)送數(shù)據(jù)
22704(BOSS 線程(Proactor))處理連接
epoll_wait(5,[{EPOLLIN, {u32=9, u64=4294967305}}], 512, -1) = 1
##數(shù)據(jù)讀出
read(9, "daojian111\r\n", 1024) = 12
##數(shù)據(jù)處理交給其他線程，這里由于線程池為空，需要先clone線程
clone(child_stack=0x7ff35c99ffb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr=0x7ff35c9a09d0, tls=0x7ff35c9a0700, child_tidptr=0x7ff35c9a09d0) = 26532復制線程處理，線程號26532
write(1, "pool-1-thread-2-10received : dao"..., 41) = 41
write(1, "\n", 1)
accept(8, 0x7f11c400b5f0, 0x7f11f42fd4d0) = -1 EAGAIN (Resource temporarily unavailable)
epoll_ctl(5, EPOLL_CTL_MOD, 8, {EPOLLIN|EPOLLONESHOT, {u32=8, u64=8}}) = 0
4 總結(jié)
- 從系統(tǒng)調(diào)用角度，Java的AIO事實上是以多路復用(Linux上為epoll)等同步IO為基礎(chǔ)，自行實現(xiàn)了異步事件分發(fā)。
- BOSS Thread負責處理連接，并分發(fā)事件。
- WORKER Thread只負責從BOSS接收的事件執(zhí)行，不負責任何網(wǎng)絡(luò)事件監(jiān)聽。
?
??
??
5 優(yōu)缺點
優(yōu)點
相比于前面的BIO、NIO，AIO已經(jīng)封裝好了任務(wù)調(diào)度，使用時只需關(guān)心任務(wù)處理。
缺點
- 事件處理完全由Thread Pool完成，對于同一個channel的多個事件可能會出現(xiàn)并發(fā)問題。
- 相比netty，buffer API不友好容易出錯;編解碼工作復雜。
