Netty基礎(chǔ)概念
歡迎訪問微信原文：Netty入門實(shí)踐
Bootstrap 和 ServerBootstrap：引導(dǎo)類，提供了一個(gè)用于應(yīng)用程序網(wǎng)絡(luò)層配置的容器；Bootstrap用于客戶端，ServerBootstrap用于服務(wù)端，輪詢客戶端的Bootstrap或DatagramChannel，監(jiān)聽其是否連接到服務(wù)器
Channel：底層網(wǎng)絡(luò)傳輸API必須提供給應(yīng)用I/O操作的接口，定義了與 socket 豐富交互的操作集，如bind, close, config, connect, isActive, isOpen, isWritable, read, write等等
ChannelHandler：支持多協(xié)議，提供用于數(shù)據(jù)處理的容器，常用接口ChannelInboundHandler，用來定義處理入站事件的方法，接受到數(shù)據(jù)后，可在其中進(jìn)行業(yè)務(wù)邏輯的處理；
ChannelPipeline：為ChannelHandler鏈提供一個(gè)API，用于管理沿著ChannelHandler鏈入站和出站的事件；每個(gè) Channel 都有自己的ChannelPipeline，當(dāng) Channel 創(chuàng)建時(shí)自動(dòng)創(chuàng)建ChannelPipeline
EventLoop: 用于處理 Channel 的 I/O 操作,一個(gè)單一的 EventLoop通常會(huì)處理多個(gè) Channel的 I/O 操作；一個(gè) EventLoopGroup 可以可以包含 EventLoop，可以自動(dòng)檢索下一個(gè) EventLoop
ChannelFuture：用于異步接受數(shù)據(jù)，通過注冊(cè)ChannelFutureListener來監(jiān)聽數(shù)據(jù)是否到達(dá)
ChannelHandlerContext：ChanneHandler鏈中，事件可以通過ChanneHandlerContext傳遞給下一個(gè)ChanneHandler；其作用之一是綁定ChannelHandler和ChannelPipeline；
ChannelHandler是如何安裝在ChannelPipeline：主要是實(shí)現(xiàn)了ChannelHandler 的抽象 ChannelInitializer。ChannelInitializer子類 通過 ServerBootstrap 進(jìn)行注冊(cè)。當(dāng)它的方法 initChannel() 被調(diào)用時(shí)，這個(gè)對(duì)象將安裝自定義的 ChannelHandler(下文中的EchoServerHandler與EchoClientHandler) 集到 pipeline。當(dāng)這個(gè)操作完成時(shí)，ChannelInitializer 子類則 從 ChannelPipeline 自動(dòng)刪除自身。
Netty demo實(shí)踐
依賴jar包
<dependency>
[ <groupId>io.netty](http://io.netty/)</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.29.Final</version>
</dependency>
服務(wù)端代碼
EchoServerHandler處理器綁定到Channel的ChannelPipeline
/**
* 通過 EchoServerHandler 實(shí)例給每一個(gè)新的 Channel 初始化
* ChannelHandler.Sharable注解：標(biāo)識(shí)這類的實(shí)例之間可以在 channel 里面共享
* @date 2018/9/27
* @description
*/
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
/**
* 每個(gè)信息入站都會(huì)調(diào)用,信息入站后，可在其中處理業(yè)務(wù)邏輯
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf in = (ByteBuf) msg;
System.out.println(System.nanoTime() + " server received: " + in.toString(CharsetUtil.UTF_8));
//將所接受的消息返回給發(fā)送者
ctx.write(in);
}
/**
* 通知處理器最后的 channelread() 是當(dāng)前批處理中的最后一條消息時(shí)調(diào)用
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
//沖刷所有待審消息到遠(yuǎn)程節(jié)點(diǎn)，監(jiān)聽到關(guān)閉通道后，操作完成
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}
/**
* 讀操作時(shí)捕獲到異常時(shí)調(diào)用:打印異常及關(guān)閉通道
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
服務(wù)端啟動(dòng)類代碼
public class EchoServer {
private final int port;
public EchoServer(int port) {
this.port = port;
}
public static void main(String[] args) throws InterruptedException {
new EchoServer(8080).start();
}
public void start() throws InterruptedException {
//使用NioEventLoopGroup接受和處理新連接
NioEventLoopGroup group = new NioEventLoopGroup();
try{
ServerBootstrap server = new ServerBootstrap();
server.group(group)
//指定NIO的傳輸channel
.channel(NioServerSocketChannel.class)
//設(shè)置 socket 地址使用所選的端口
.localAddress(port)
//添加 EchoServerHandler 到 Channel 的 ChannelPipeline
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new EchoServerHandler());
}
});
//綁定的服務(wù)器 同步等待服務(wù)器綁定完成
ChannelFuture future = server.bind().sync();
System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
//等待channel關(guān)閉
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}finally {
//關(guān)閉NioEventLoopGroup 釋放所有資源
group.shutdownGracefully().sync();
}
}
}
客戶端代碼
客戶端處理數(shù)據(jù)的處理器
@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
/**
* 服務(wù)器的連接被建立后調(diào)用
* @param ctx
*/
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.writeAndFlush(Unpooled.copiedBuffer("hello netty, how are you?", CharsetUtil.UTF_8));
}
/**
* 接收到服務(wù)器返回的數(shù)據(jù)后調(diào)用
* @param channelHandlerContext
* @param byteBuf
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
//記錄接收到的消息
System.out.println(System.nanoTime() + " Client received：" + byteBuf.toString(CharsetUtil.UTF_8));
}
/**
* 捕獲一個(gè)異常時(shí)調(diào)用
* @param ctx
* @param cause
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
客戶端啟動(dòng)類代碼
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) throws InterruptedException {
new EchoClient("localhost", 8080).start();
}
public void start() throws InterruptedException {
//指定 EventLoopGroup 來處理客戶端事件
EventLoopGroup group = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(host, port))
//當(dāng)建立一個(gè)連接和一個(gè)新的通道時(shí)，創(chuàng)建添加到 EchoClientHandler 實(shí)例 到 channel pipeline
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new EchoClientHandler());
}
});
//等待鏈接完成
ChannelFuture future = bootstrap.connect().sync();
//阻塞直到channel關(guān)閉
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}finally {
//關(guān)閉線程池和釋放所有資源
group.shutdownGracefully().sync();
}
}
}
測(cè)試代碼
先啟動(dòng)服務(wù)端EchoServer.main()方法，在啟動(dòng)客戶端EchoClient.main()方法
服務(wù)端打印日志：
198733200670178 server received: hello netty, how are you?
客戶端打印日志：
198733224597828 Client received：hello netty, how are you?
歡迎關(guān)注微信公眾號(hào)
