IO模型
IO多路復(fù)用模式：Reactor、Proactor
NIO實(shí)現(xiàn)的是Reactor模式汪榔。通過select、epoll函數(shù)，用戶可以一個(gè)線程同時(shí)處理多個(gè)Channel的IO請(qǐng)求贫奠。當(dāng)數(shù)據(jù)就緒唬血，再通過實(shí)際的用戶線程進(jìn)行數(shù)據(jù)拷貝，邏輯處理
- 注冊(cè)讀事件及其對(duì)應(yīng)的事件處理器
- 事件分離器(select\epoll)等待事件
- 事件到來唤崭，分離器調(diào)用相應(yīng)的處理器
- 事件處理器完成讀操作拷恨，處理數(shù)據(jù)
AIO實(shí)現(xiàn)的是Proactor模式。由操作系統(tǒng)內(nèi)核負(fù)責(zé)IO數(shù)據(jù)讀寫谢肾，然后回調(diào)函數(shù)進(jìn)行邏輯處理
- 事件處理器發(fā)起讀請(qǐng)求
- 事件分離器等待讀事件完成
- 在分離器等待過程中腕侄，操作系統(tǒng)利用并行的內(nèi)核線程執(zhí)行實(shí)際的讀操作，并將結(jié)果數(shù)據(jù)存入用戶自定義緩沖區(qū)芦疏，最后通知事件分離器讀操作完成
- 事件分離器通知事件處理器冕杠，讀操作已完成
- 事件處理器處理緩沖區(qū)數(shù)據(jù)
兩者主要區(qū)別：用戶線程或是操作系統(tǒng)內(nèi)核線程進(jìn)行IO數(shù)據(jù)讀寫
引入Netty
Netty中使用的Reactor模式，引入了多Reactor(1個(gè)select線程+N個(gè)IO線程+M個(gè)worker線程)酸茴。即一個(gè)主Reactor負(fù)責(zé)監(jiān)控所有的連接請(qǐng)求分预，多個(gè)子Reactor負(fù)責(zé)監(jiān)控并處理讀/寫請(qǐng)求，減輕了主Reactor的壓力薪捍，降低了主Reactor壓力太大而造成的延遲笼痹。
并且每個(gè)子Reactor分別屬于一個(gè)獨(dú)立的線程，每個(gè)成功連接后Channel的所有操作由同一個(gè)線程處理酪穿。這樣保證了同一請(qǐng)求的所有狀態(tài)和上下文在同一個(gè)線程中凳干，避免了不必要的上下文切換，同時(shí)也方便了監(jiān)控請(qǐng)求響應(yīng)狀態(tài)被济。
多Reactor模式
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.22.Final</version>
</dependency>
Netty服務(wù)端\客戶端都需要以下兩部分
- 至少一個(gè)ChannelHandler: 該組件實(shí)現(xiàn)了接收的數(shù)據(jù)處理救赐，即消息的業(yè)務(wù)邏輯
- 引導(dǎo)Bootstrap: 服務(wù)器\客戶端啟動(dòng)配置。比如監(jiān)聽端口溉潭、IO處理線程數(shù)净响、Channel處理邏輯
編寫Echo服務(wù)端
事件處理器 ChannelHandler
@Slf4j
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
private final ChannelGroup channels = new DefaultChannelGroup("Echo-Server", GlobalEventExecutor.INSTANCE);
/**
* 客戶端連接到服務(wù)端
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
log.info("channel id: {}", ctx.channel().id().asLongText());
// 廣播消息給所有channels
channels.writeAndFlush("client: " + ctx.channel().remoteAddress() + " add");
channels.add(ctx.channel());
}
/**
* 客戶端斷開連接
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
channels.writeAndFlush("client: " + ctx.channel().remoteAddress() + " remove");
channels.remove(ctx.channel());
}
/**
* Channel處于活動(dòng)狀態(tài)，已經(jīng)連接到遠(yuǎn)程節(jié)點(diǎn)喳瓣。在線馋贤！
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("{} active", ctx.channel().remoteAddress());
}
/**
* Channel未連接到遠(yuǎn)程節(jié)點(diǎn)。掉線畏陕！
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.warn("{} inactive", ctx.channel().remoteAddress());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf in = (ByteBuf) msg;
log.info("Server received: {}", in.toString(StandardCharsets.UTF_8));
ctx.writeAndFlush(in); // write是把數(shù)據(jù)寫入到OutboundBuffer(不真正發(fā)送數(shù)據(jù))配乓，flush是真正的發(fā)送數(shù)據(jù)
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
log.info("Server read complete.");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
log.error("Server error.");
ctx.close();
}
}
服務(wù)端繼承ChannelInboundHandlerAdapter類，一般只需要實(shí)現(xiàn)channelRead()惠毁、exceptionCaught()方法即可
引導(dǎo) Bootstrap
public class EchoServer {
private final int port;
public EchoServer(int port) {
this.port = port;
}
public static void main(String[] args) {
new EchoServer(9002).start();
}
public void start() {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new EchoServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
// 阻塞綁定port犹芹，直到成功
ChannelFuture future = bootstrap.bind(port).sync();
// 阻塞等待，直到服務(wù)器的Channel關(guān)閉
future.channel().closeFuture().sync();
} catch (Exception ignore) {
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
bossGroup線程監(jiān)聽Channel鞠绰，只需要一個(gè)腰埂，多了沒用；workerGroup負(fù)責(zé)IO讀寫
編寫Echo客戶端
事件處理器 ChannelHandler
@Slf4j
@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("Channel active.");
ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", StandardCharsets.UTF_8));
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
log.info("Client received: {}", msg.toString(StandardCharsets.UTF_8));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
log.error("Client error.");
ctx.close();
}
}
客戶端繼承SimpleChannelInboundHandler類蜈膨，此類繼承ChannelInboundHandlerAdapter并實(shí)現(xiàn)了channelRead()方法屿笼，業(yè)務(wù)handler覆寫channelRead0()方法
引導(dǎo) Bootstrap
public class EchoClient {
private final String host;
private final int port;
public EchoClient(String host, int port) {
this.host = host;
this.port = port;
}
public static void main(String[] args) {
new EchoClient("localhost", 9002).start();
}
public void start() {
EventLoopGroup workerGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(workerGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.SO_KEEPALIVE, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new EchoClientHandler());
}
});
try {
ChannelFuture future = bootstrap.connect(host, port).sync();
future.channel().closeFuture().sync();
} catch (Exception ignore) {
} finally {
workerGroup.shutdownGracefully();
}
}
}
