Netty 系列（三）Netty 入门
Netty 是一个提供异步事件驱动的网络应用框架，用以快速开发高性能、高可靠性的网络服务器和客户端程序。更多请参考：Netty Github 和 Netty中文入门。
一、获得 Netty
可以通过Maven安装Netty。查看Netty之HelloWorld快速入门，更多API
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>5.0.0.Alpha2</version>
</dependency>
二、Netty 服务端开发
现在让我们从服务端的处理器的实现开始，处理器是由 Netty 生成用来处理 I/O 事件的。
public class ServerHandler extends ChannelHandlerAdapter { // (1)
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // (2)
//1. 接收客户端的请求数据
ByteBuf buf = (ByteBuf)msg;
byte[] data = new byte[buf.readableBytes()];
buf.readBytes(data);
String request = new String(data, "utf-8");
System.out.println("收到 client 请求数据：" + request);
//2. 返回响应数据，ctx.write()后自动释放msg
ChannelFuture f = ctx.writeAndFlush(Unpooled.copiedBuffer("netty".getBytes())); // (3)
//2.1 写完成后会自动关闭 client，否则与 client 建立长连接
f.addListener(ChannelFutureListener.CLOSE); // (4)
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // (5)
cause.printStackTrace();
ctx.close();
}
}
-
DisCardServerHandler 继承自 ChannelHandlerAdapter，这个类实现了ChannelHandler接口，ChannelHandler提供了许多事件处理的接口方法，然后你可以覆盖这些方法。现在仅仅只需要继承ChannelHandlerAdapter类而不是你自己去实现接口方法。
-
这里我们覆盖了chanelRead()事件处理方法。每当从客户端收到新的数据时，这个方法会在收到消息时被调用，这个例子中，收到的消息的类型是ByteBuf
-
ByteBuf是一个引用计数对象，这个对象必须显示地调用release()方法来释放。ctx.write()后自动释放 msg，否则，channelRead()方法就需要像下面的这段代码一样来手动释放 msg：
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { // Do something with msg } finally { // ((ByteBuf) msg).release(); ReferenceCountUtil.release(msg); } } -
写完成后程序不会自动关闭与 client 的连接，你需要手动绑定 ChannelFuture 的监听事件，写完成后才会关闭连接，ChannelFutureListener.CLOSE 的实现如下：
ChannelFutureListener CLOSE = new ChannelFutureListener() { public void operationComplete(ChannelFuture future) { future.channel().close(); } }; -
exceptionCaught()事件处理方法是当出现Throwable对象才会被调用，即当Netty由于IO错误或者处理器在处理事件时抛出的异常时。在大部分情况下，捕获的异常应该被记录下来并且把关联的channel给关闭掉。然而这个方法的处理方式会在遇到不同异常的情况下有不同的实现，比如你可能想在关闭连接之前发送一个错误码的响应消息。
到目前为止一切都还比较顺利，接下来我拉需要编写一个 main() 方法来启动服务端的 ServerHandler。
public class Server {
private int port;
public Server(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
//1. 第一个线程组是用于接收Client端连接
bossGroup = new NioEventLoopGroup();
//2. 第二个线程组是用于处理实现的业务操作
workerGroup = new NioEventLoopGroup();
//3. ServerBootstrap 是一个启动NIO服务的辅助启动类
ServerBootstrap b = new ServerBootstrap(); // (2)
//3.1 将两个工作线程组加进来
b.group(bossGroup, workerGroup)
//3.2 指定使用NioServerSocketChannel这种类型的通道
.channel(NioServerSocketChannel.class) // (3)
//3.3 使用childHandler来绑定具体的事件处理器
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ServerHandler());
}
})
//3.4 设置TCP缓冲区大小，默认128，一般不用改
.option(ChannelOption.SO_BACKLOG, 128) // (5)
//3.5 设置发送缓冲区大小
.option(ChannelOption.SO_SNDBUF, 32 * 1034)
//3.6 设置接收缓冲区大小
.option(ChannelOption.SO_RCVBUF, 32 * 1034)
//3.7 KEEPALIVE
.childOption(ChannelOption.SO_KEEPALIVE, true);
//4. 绑定端口
ChannelFuture f = b.bind(port).sync(); // (7)
//5. 监听通道关闭 <=> 阻塞程序，不然Server直接执行完成后关闭，client就不可能连接上了
//Thread.sleep(Integer.MAX_VALUE);
f.channel().closeFuture().sync();
} finally {
//6. 修优雅退出，释放线程池资源
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8765;
}
new Server(port).run();
}
}
-
NioEventLoopGroup 是用来处理I/O操作的多线程事件循环器，Netty提供了许多不同的EventLoopGroup的实现用来处理不同传输协议。在这个例子中我们实现了一个服务端的应用，因此会有2个NioEventLoopGroup会被使用。第一个经常被叫做‘boss’，用来接收进来的连接。第二个经常被叫做‘worker’，用来处理已经被接收的连接，一旦‘boss’接收到连接，就会把连接信息注册到‘worker’上。如何知道多少个线程已经被使用，如何映射到已经创建的Channels上都需要依赖于EventLoopGroup的实现，并且可以通过构造函数来配置他们的关系。
-
ServerBootstrap 是一个启动NIO服务的辅助启动类。你可以在这个服务中直接使用Channel，但是这会是一个复杂的处理过程，在很多情况下你并不需要这样做。
-
这里我们指定使用NioServerSocketChannel类来举例说明一个新的Channel如何接收进来的连接。
-
这里的事件处理类经常会被用来处理一个最近的已经接收的Channel。ChannelInitializer是一个特殊的处理类，他的目的是帮助使用者配置一个新的Channel。
-
你可以设置这里指定的通道实现的配置参数。我们正在写一个TCP/IP的服务端，因此我们被允许设置socket的参数选项比如tcpNoDelay和keepAlive。请参考ChannelOption和详细的ChannelConfig实现的接口文档以此可以对ChannelOptions的有一个大概的认识。
通过以步骤，一个服务端就搭建好了。
三、客户端开发
public class Client {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup workgroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workgroup)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new ClientHandler());
}
});
//发起异步连接操作
ChannelFuture f = b.connect("127.0.0.1", 8080).sync();
//向服务器发送数据 buf
f.channel().writeAndFlush(Unpooled.copiedBuffer("777".getBytes()));
//等待客户端链路关闭
f.channel().closeFuture().sync();
} finally {
//优雅退出，释放 NIO 线程组
workgroup.shutdownGracefully();
}
}
}
客户端业务处理ClientHandler
public class ClientHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
try {
//读取buf中的数据
ByteBuf buf = (ByteBuf) msg;
byte[] data = new byte[buf.readableBytes()];
buf.readBytes(data);
System.out.println(new String(data));
} finally {
//释放 (ByteBuf) msg
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
e.printStackTrace();
ctx.close();
}
}
总结
Netty自定义协议 https://my.oschina.net/OutOfMemory/blog/290180
每天用心记录一点点。内容也许不重要，但习惯很重要！