Netty 核心组件
Channel- 回调
Future- 事件和
ChannelHandler
Channel
Channel 是 Java NIO 的一个基本构造，可以把它看作是传入或者传出数据的载体（通道），因此可以被打开或者关闭。
回调
回调函数就是一个通过函数指针调用的函数。如果你把函数的指针（地址）作为参数传递给另一个函数，当这个指针被用来调用其所指向的函数时，我们就说这是回调函数。
Netty 在内部是通过回调来处理事件；当事件发生时，Netty会回调用实现了 interface-ChannelHandler 的函数处理该事件。
Future
Future用于表示异步操作的结果，每个Netty 的出战 I/O 操作都会返回一个 ChannelFuture。
事件和 ChannelHandler
Netty 使用不同的事件来通知我们状态的改变或者操作的状态，我们可以通过发生的事件来触发相对于的动作。
发生的每个事件都会被分发给 ChannelHander 类中的某个使用户实现的方法。大部分情况我们都需要自己实现 ChannelHandler 然后通过通过重写方法实现自己的业务逻辑。
基于 Netty 的服务器和客户端
要做的很简单，就是：客户端发送消息给服务器，然后服务器接收客户端发动的消息并回复一条消息给客户端。
Netty 依赖
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.32.Final</version>
</dependency>
</dependencies>
服务器
基于 Neety 的服务器需要两部分：
- ChannelHandler 实现类，我们需要通过实现它来接收客服端的数据（业务逻辑）
- 引导——服务器配置及启动代码（配置）
ChannelHandler 实现
package com.project.demo;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
@Sharable // 标识 Channel-Handler 可以被多个 Channel 安全共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
// 读取客服端消息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 读取消息
ByteBuf in = (ByteBuf) msg;
System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));
// 将消息写给发送者, 并不发送，需要调用 flush 方法才会发送
ctx.write(Unpooled.copiedBuffer("你好！", CharsetUtil.UTF_8));
}
// 一条消息读取完成(一条消息可能需要多次读取)时调用
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
// 将消息发送给客服端，并关闭该 Channel
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
.addListener(ChannelFutureListener.CLOSE);
}
// 在读取消息期间，发生异常时调用
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
上面代码中，我们继承一个ChannelHandler的实现类 ChannelInboundHandlerAdapter
引导
ChannelHander 写好了，我还还需要做以下工作才能使服务器工作：
- 设置监听端口
- 配置 Channel，使入站消息发送给我们刚刚编写的 EchoServerHandler 处理。
package com.project.demo;
import java.net.InetSocketAddress;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* 1. 创建一个ServerBootstrap的实例以引导和绑定服务器
* 2. 创建并分配一个NioEventLoopGroup实例以进行事件的处理
* 3. 指定服务器绑定的本地的InetSocketAddress
*/
public class EchoServer {
private int port;
public EchoServer(int port) {
this.port = port;
}
public void start() throws Exception {
// 消息处理器实例
final EchoServerHandler serverHandler = new EchoServerHandler();
// Nio
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
// 配置
b.group(group)
.channel(NioServerSocketChannel.class) // 指定使用的 NIO
.localAddress(new InetSocketAddress(port)) // 指定端口
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(Channel ch) throws Exception {
// 注册自定义的 ChannelHandle 使之生效
ch.pipeline().addLast(serverHandler);
}
});
// 绑定服务器
ChannelFuture f = b.bind().sync();
// 获取 channel 的 CloseFuture
f.channel().closeFuture().sync();
} finally {
// 关闭EventLoopGroup，释放所有的资源
group.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoServer(8081).start(); // 启动
}
}
客户端
客服端同服务器一样也需要编写两个部分的代码，ChannelHandler实现（业务逻辑）和引导（配置和启动）
ChannelHandler实现
package com.project.demo;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
// 客户端消息处理器
@Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
// 接收数据时被调用
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
System.out.println("Chient received: " + msg.toString(CharsetUtil.UTF_8));
}
// 建立时被调用
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 发送消息到服务器
ctx.writeAndFlush(Unpooled.copiedBuffer("Hello！", CharsetUtil.UTF_8));
}
// 处理过程中发生异常时被调用
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
// 关闭 Channel
ctx.close();
}
}
引导
客户端的引导需要指定主机及端口
package com.project.demo;
import java.net.InetSocketAddress;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* 1. 创建了一个Bootstrap实例
* 2. 为进行事件处理分配了一个NioEventLoopGroup实例
* 3. 为服务器连接创建了一个InetSocketAddress实例
* 4. 当连接建立时，EchoClientHandle 实例会被安装到 ChannelPipeline 中
* 5. 配置完成后，通过调用Bootstrap.connect()方法连接到远程节点（服务器）
*/
public class EchoClient {
private final String host;
private final int port;
public EchoClient(String host, int port) {
this.host = host;
this.port = port;
}
public void start() throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(host, port)) // 指定 host，port
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new EchoClientHandler());
}
});
// 连接到远程节点，阻塞等待直到连接完成
ChannelFuture f = b.connect().sync();
// 阻塞，直到 Channel 关闭
f.channel().closeFuture().sync();
} finally {
// 关闭连接次并释放资源
group.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoClient("127.0.0.1", 8081).start();
}
}
到这里这个基于 Netty 的简单程序就编写完了，想测试是否成功可以先启动 EchoServer 然后再启动 EchoClient，客户端启动会发送一条 Hello 给服务器，服务器接收到消息后会回复 你好 给客户端。
Netty 常用组件
Netty网络抽象的代表：
-
Channel——Socket; -
EventLoop——控制流、多线程处理、并发； -
ChannelFuture——异步通知。
