沉淀再出发：关于netty的一些理解和使用
一、前言
Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。也就是说，Netty 是一个基于NIO的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。“快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议（包括FTP、SMTP、HTTP等各种二进制文本协议）的实现经验，并经过相当精心设计的项目。最终，Netty 成功的找到了一种方式，在保证易于开发的同时还保证了其应用的性能，稳定性和伸缩性。netty在底层的数据通信和封装之中有着重要的作用，下面我们就来看看netty的简单使用过程，以及背后的原理。
二、netty的简单使用
2.1、netty的环境部署和使用
在这里我们使用myeclipse平台，maven管理工具进行开发，其实使用eclipse或者其他软件也可以。首先我们新建一个maven项目，项目名和包名自定：
之后我们修改pom.xml文件，增加netty依赖：
保存之后，系统就会自动为我们下载和安装了，非常的方便，这样，我们的环境就部署完毕了。
2.2、一个简单的案例
下面我们看一个简单地案例：
我们新建一个包，然后写入两个文件：
首先我们编写一个处理连接的类 HelloServerHandler ：
package com.coder.server; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil; public class HelloServerHandler extends ChannelInboundHandlerAdapter {
/**
* 收到数据时调用
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
ByteBuf in = (ByteBuf)msg;
System.out.print(in.toString(CharsetUtil.UTF_8));
} finally {
// 抛弃收到的数据
ReferenceCountUtil.release(msg);
}
} /**
* 当Netty由于IO错误或者处理器在处理事件时抛出异常时调用
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 当出现异常就关闭连接
cause.printStackTrace();
ctx.close();
}
}
其次，我们编写接收连接，并且派发和处理的类 HelloServer :
package com.coder.server; import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel; public class HelloServer {
private int port; public HelloServer(int port) {
this.port = port;
} public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已经被接收的连接
System.out.println("准备运行端口：" + port); try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 这里告诉Channel如何接收新的连接
.childHandler( new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 自定义处理类
ch.pipeline().addLast(new HelloServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口，开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // 等待服务器socket关闭
f.channel().closeFuture().sync();
} catch (Exception e) {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
int port = 12345;
new HelloServer(port).run();
}
}
然后运行，等待连接就好了，那么问题来了，使用什么进行连接呢？在windows中，我们可以使用Telnet，这个比较方便和简单，但是我们需要打开控制面板的程序和功能模块，并且启动服务，之后最好重启一下电脑：
下面我们运行程序，并使用Telnet客户端测试一下：
在telnet中‘ctrl+]’可以显示输入的文字，否则将看不到输入。
三、使用netty自定义时间服务器
本例中我们试图在服务器和客户端连接被创立时发送一个消息，然后在客户端解析收到的消息并输出。并且，在这个项目中使用 POJO 代替 ByteBuf 来作为传输对象。
3.1、pojo对象创建
Time 类：
package com.coder.pojo; import java.util.Date; /**
* 自定义时间数据类
*
*/
public class Time {
private final long value; public Time() {
// 除以1000是为了使时间精确到秒
//注意这里的this，其实就是调用了 public Time(long value) ，并且更加的方便和快捷。
this(System.currentTimeMillis() / 1000L);
} public Time(long value) {
this.value = value;
} public long value() {
return value;
} @Override
public String toString() {
return new Date((value()) * 1000L).toString();
}
}
3.2、服务器程序
TimeEncoderPOJO类：
package com.coder.server; import com.coder.pojo.Time; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder; /**
* 服务器数据编码类
*
*/
public class TimeEncoderPOJO extends MessageToByteEncoder<Time> { // 发送数据时调用
@Override
protected void encode(ChannelHandlerContext ctx, Time msg, ByteBuf out) throws Exception {
// 只传输当前时间，精确到秒
out.writeInt((int)msg.value());
} }
TimeServerHandlerPOJO类：连接建立并且准备通信的时候进行处理，发送当前时间，并增加监听。
package com.coder.server; import com.coder.pojo.Time; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 服务器解码器
* 连接建立时发送当前时间
*
*/
public class TimeServerHandlerPOJO extends ChannelInboundHandlerAdapter {
/**
* 连接建立的时候并且准备进行通信时被调用
*/
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
// 发送当前时间信息
ChannelFuture f = ctx.writeAndFlush(new Time());
// 发送完毕之后关闭 Channel
f.addListener(ChannelFutureListener.CLOSE);
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
TimeServerPOJO类：服务器的主程序
package com.coder.server; import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel; public class TimeServerPOJO {
private int port; public TimeServerPOJO(int port) {
this.port = port;
} public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已经被接收的连接
System.out.println("准备运行端口：" + port); try {
ServerBootstrap b = new ServerBootstrap(); // 启动NIO服务的辅助启动类
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 这里告诉Channel如何接收新的连接
.childHandler( new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 自定义处理类
// 注意添加顺序
ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口，开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // 等待服务器socket关闭
f.channel().closeFuture().sync();
} catch (Exception e) {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
int port = 12345;
new TimeServerPOJO(port).run();
}
}
其中ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());方法的含义为：Handles an I/O event or intercepts an I/O operation, and forwards it to its next handler in its ChannelPipeline.也就是说当我们添加一些处理的时候会按照管道的方式，一步步的处理，因此先后顺序非常重要。
3.3、客户端程序
先来看看解码器（服务器端发送了编码后的时间信息，因此，这里客户端收到之后需要解码）：
TimeDecoderPOJO 类：
package com.coder.client; import java.util.List; import com.coder.pojo.Time; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder; public class TimeDecoderPOJO extends ByteToMessageDecoder {
/**
* 有新数据接收时调用
* 为防止分包现象，先将数据存入内部缓存，到达满足条件之后再进行解码
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in.readableBytes() < 4) {
return;
} // out添加对象则表示解码成功
out.add(new Time(in.readUnsignedInt()));
}
}
再看看客户端数据处理类：
TimeClientHandlerPOJO类：
package com.coder.client; import com.coder.pojo.Time; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 客户端数据处理类
*
*/
public class TimeClientHandlerPOJO extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 直接将信息转换成Time类型输出即可
Time time = (Time)msg;
System.out.println(time);
ctx.close();
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
最后是客户端的主程序：
TimeClientPOJO类：
package com.coder.client; import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel; public class TimeClientPOJO {
public static void main(String[] args) throws Exception{
String host = "127.0.0.1"; // ip
int port = 12345; // 端口
EventLoopGroup workerGroup = new NioEventLoopGroup(); try {
Bootstrap b = new Bootstrap(); // 与ServerBootstrap类似
b.group(workerGroup); // 客户端不需要boss worker
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true); // 客户端的socketChannel没有父亲
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// POJO
ch.pipeline().addLast(new TimeDecoderPOJO() ,new TimeClientHandlerPOJO());
}
}); // 启动客户端，客户端用connect连接
ChannelFuture f = b.connect(host, port).sync(); // 等待连接关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
}
至此程序编写完毕，先运行服务器，再运行客户端程序，然后测试即可，我们会发现服务器一直等待着请求，当客户端连接上之后，服务器就会发出带着格式的时间，客户端接收到之后进行解码，然后显示出来并且退出。在同一个myeclipse之中可以运行多个程序，使用下图中的按钮可以进行切换。
四、netty的基本组成部分
4.1、Channel
Channel 是 Java NIO 的一个基本构造。它代表一个到实体（如一个硬件设备、一个文件、一个网络套接字或者一个能够执行一个或者多个不同的I/O操作的程序组件）的开放连接，如读操作和写操作。目前，可以把 Channel 看作是传入（入站）或者传出（出站）数据的载体。因此，它可以被打开或者被关闭，连接或者断开连接。
4.2、Callback（回调）
Netty 在内部使用了回调来处理事件；当一个回调被触发时，相关的事件可以被一个 interfaceChannelHandler 的实现处理。
4.3、Future
Future 提供了另一种在操作完成时通知应用程序的方式。这个对象可以看作是一个异步操作的结果的占位符；它将在未来的某个时刻完成，并提供对其结果的访问。JDK 预置了 interface java.util.concurrent.Future，但是其所提供的实现，只允许手动检查对应的操作是否已经完成，或者一直阻塞直到它完成。这是非常繁琐的，所以 Netty 提供了它自己的实现ChannelFuture，用于在执行异步操作的时候使用。
4.4、Event 和 Handler
Netty 使用不同的事件来通知我们状态的改变或者是操作的状态。这使得我们能够基于已经发生的事件来触发适当的动作。这些动作可能是：记录日志、数据转换、流控制、应用程序逻辑。Netty 是一个网络编程框架，所以事件是按照它们与入站或出站数据流的相关性进行分类的。可能由入站数据或者相关的状态更改而触发的事件包括：连接已被激活或者连接失活、数据读取、用户事件、错误事件。出站事件是未来将会触发的某个动作的操作结果，这些动作包括：打开或者关闭到远程节点的连接、将数据写到或者冲刷到套接字。
Netty 的 ChannelHandler 为处理器提供了基本的抽象，目前可以认为每个 ChannelHandler 的实例都类似于一种为了响应特定事件而被执行的回调。Netty 提供了大量预定义的可以开箱即用的 ChannelHandler 实现，包括用于各种协议（如 HTTP 和 SSL/TLS）的 ChannelHandler。在内部 ChannelHandler 自己也使用了事件和 Future。
五、netty聊天程序
5.1、服务器端
SimpleChatServerInitializer类：
package com.coder.server; 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; /**
* 服务器配置初始化
* 添加多个处理器
*/
public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 添加处理类
// 使用'\r''\n'分割帧
pipeline.addLast("framer",
new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码、编码器
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 处理器
pipeline.addLast("handler", new SimpleChatServerHandler()); System.out.println("SimpleChatClient: " + ch.remoteAddress() + "连接上");
} }
SimpleChatServerHandler类：
package com.coder.server; import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor; /**
* 服务端处理器
*/
public class SimpleChatServerHandler extends SimpleChannelInboundHandler<String> { public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); /**
* 收到新的客户端连接时调用
* 将客户端channel存入列表，并广播消息
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
// 广播加入消息
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
channels.add(incoming); // 存入列表
} /**
* 客户端连接断开时调用
* 广播消息
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
// 广播离开消息
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
// channel会自动从ChannelGroup中删除
} /**
* 收到消息时调用
* 将消息转发给其他客户端
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
Channel incoming = ctx.channel();
for(Channel channel : channels) { // 遍历所有连接的客户端
if(channel != incoming) { // 其他客户端
channel.writeAndFlush("[" + incoming.remoteAddress() + "] " + msg + "\n" );
} else { // 自己
channel.writeAndFlush("[you] " + msg + "\n" );
}
}
} /**
* 监听到客户端活动时调用
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 在线");
} /**
* 监听到客户端不活动时调用
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 掉线");
} /**
* 当Netty由于IO错误或者处理器在处理事件抛出异常时调用
* 关闭连接
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 异常");
}
}
SimpleChatServer类：
package com.coder.server; import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel; /**
* 服务端 main 启动
*/
public class SimpleChatServer {
private int port; // 端口 public SimpleChatServer(int port) {
this.port = port;
} // 配置并开启服务器
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已接收的连接 try {
ServerBootstrap sb = new ServerBootstrap(); // 启动NIO服务的辅助启动类
sb.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 设置如何接受连接
.childHandler(new SimpleChatServerInitializer()) // 配置Channel
.option(ChannelOption.SO_BACKLOG, 128) // 设置缓冲区
.childOption(ChannelOption.SO_KEEPALIVE, true); // 启用心跳机制 System.out.println("SimpleChatServer 启动了");
ChannelFuture future = sb.bind(port).sync(); // 绑定端口，开始接收连接
future.channel().closeFuture().sync(); // 等待关闭服务器（不会发生）
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
System.out.println("SimpleChatServer 关闭了");
}
} public static void main(String[] args) throws Exception {
int port = 8080;
new SimpleChatServer(port).run(); // 开启服务器
}
}
5.2、客户端程序
SimpleChatClientInitializer类：
package com.coder.client; 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; /**
* 客户端配置初始化
* 与服务端类似
*/
public class SimpleChatClientInitializer extends ChannelInitializer<SocketChannel> { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 添加处理类
// 使用'\r''\n'分割帧
pipeline.addLast("framer",
new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码、编码器
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 处理器
pipeline.addLast("handler", new SimpleChatClientHandler());
} }
SimpleChatClientHandler类：
package com.coder.client; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler; /**
* 客户端处理类
* 直接输出收到的消息
*/
public class SimpleChatClientHandler extends SimpleChannelInboundHandler<String> { @Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(msg); // 直接输出消息
} }
SimpleChatClient类：
package com.coder.client; import java.io.BufferedReader;
import java.io.InputStreamReader; import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* 客户端
* 开启客户端，接收控制台输入并发送给服务端
*/
public class SimpleChatClient {
private final String host; // IP
private final int port; // 端口 public SimpleChatClient(String host, int port) {
this.host = host;
this.port = port;
} // 配置并运行客户端
public void run() throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap(); // 客户端辅助启动类
b.group(group) // 客户端只需要一个用来接收并处理连接
.channel(NioSocketChannel.class) // 设置如何接受连接
.handler(new SimpleChatClientInitializer());// 配置 channel
// 连接服务器
Channel channel = b.connect(host, port).sync().channel();
// 读取控制台输入字符
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true) {
// 每行成一帧输出，以"\r\n"结尾
channel.writeAndFlush(in.readLine() + "\r\n");
}
} catch (Exception e) {
e.printStackTrace(); // 输出异常
} finally {
group.shutdownGracefully(); // 关闭
}
} public static void main(String[] args) throws Exception {
new SimpleChatClient("localhost", 8080).run(); // 启动客户端
} }
运行结果：
六、总结
通过代码的形式，我们对netty有了直观的了解和实际上的掌握。