Netty基础概念
欢迎访问微信原文：Netty入门实践
Bootstrap 和 ServerBootstrap：引导类，提供了一个用于应用程序网络层配置的容器；Bootstrap用于客户端，ServerBootstrap用于服务端，轮询客户端的Bootstrap或DatagramChannel，监听其是否连接到服务器
Channel：底层网络传输API必须提供给应用I/O操作的接口，定义了与 socket 丰富交互的操作集，如bind, close, config, connect, isActive, isOpen, isWritable, read, write等等
ChannelHandler：支持多协议，提供用于数据处理的容器，常用接口ChannelInboundHandler，用来定义处理入站事件的方法，接受到数据后，可在其中进行业务逻辑的处理；
ChannelPipeline：为ChannelHandler链提供一个API，用于管理沿着ChannelHandler链入站和出站的事件；每个 Channel 都有自己的ChannelPipeline，当 Channel 创建时自动创建ChannelPipeline
EventLoop: 用于处理 Channel 的 I/O 操作,一个单一的 EventLoop通常会处理多个 Channel的 I/O 操作；一个 EventLoopGroup 可以可以包含 EventLoop，可以自动检索下一个 EventLoop
ChannelFuture：用于异步接受数据，通过注册ChannelFutureListener来监听数据是否到达
ChannelHandlerContext：ChanneHandler链中，事件可以通过ChanneHandlerContext传递给下一个ChanneHandler；其作用之一是绑定ChannelHandler和ChannelPipeline；
ChannelHandler是如何安装在ChannelPipeline：主要是实现了ChannelHandler 的抽象 ChannelInitializer。ChannelInitializer子类 通过 ServerBootstrap 进行注册。当它的方法 initChannel() 被调用时，这个对象将安装自定义的 ChannelHandler(下文中的EchoServerHandler与EchoClientHandler) 集到 pipeline。当这个操作完成时，ChannelInitializer 子类则 从 ChannelPipeline 自动删除自身。
Netty demo实践
依赖jar包
<dependency>
[ <groupId>io.netty](http://io.netty/)</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.29.Final</version>
</dependency>
服务端代码
EchoServerHandler处理器绑定到Channel的ChannelPipeline
/**
* 通过 EchoServerHandler 实例给每一个新的 Channel 初始化
* ChannelHandler.Sharable注解：标识这类的实例之间可以在 channel 里面共享
* @date 2018/9/27
* @description
*/
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
/**
* 每个信息入站都会调用,信息入站后，可在其中处理业务逻辑
* @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));
//将所接受的消息返回给发送者
ctx.write(in);
}
/**
* 通知处理器最后的 channelread() 是当前批处理中的最后一条消息时调用
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
//冲刷所有待审消息到远程节点，监听到关闭通道后，操作完成
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}
/**
* 读操作时捕获到异常时调用:打印异常及关闭通道
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
服务端启动类代码
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)
//设置 socket 地址使用所选的端口
.localAddress(port)
//添加 EchoServerHandler 到 Channel 的 ChannelPipeline
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new EchoServerHandler());
}
});
//绑定的服务器 同步等待服务器绑定完成
ChannelFuture future = server.bind().sync();
System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
//等待channel关闭
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}finally {
//关闭NioEventLoopGroup 释放所有资源
group.shutdownGracefully().sync();
}
}
}
客户端代码
客户端处理数据的处理器
@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
/**
* 服务器的连接被建立后调用
* @param ctx
*/
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.writeAndFlush(Unpooled.copiedBuffer("hello netty, how are you?", CharsetUtil.UTF_8));
}
/**
* 接收到服务器返回的数据后调用
* @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));
}
/**
* 捕获一个异常时调用
* @param ctx
* @param cause
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
客户端启动类代码
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))
//当建立一个连接和一个新的通道时，创建添加到 EchoClientHandler 实例 到 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关闭
future.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}finally {
//关闭线程池和释放所有资源
group.shutdownGracefully().sync();
}
}
}
测试代码
先启动服务端EchoServer.main()方法，在启动客户端EchoClient.main()方法
服务端打印日志：
198733200670178 server received: hello netty, how are you?
客户端打印日志：
198733224597828 Client received：hello netty, how are you?
欢迎关注微信公众号
