上一篇博客我们说了Java基础中BIO、NIO、AIO，其中用NIO和AIO开发中需要处理的东西比较多，Netty是IO这方面优秀的框架，我们来看个Netty的入门，以及其解决IO传输中的比较恶心的粘包和拆包问题。
首先我们到http://netty.io/ 去下载netty的jar包，我是通过Maven下载了5.0.0.Alpha2版本，然后我们看下入门实例。注意其中的几个点即可：1，服务端：ServerBootstrap、父子线程组、继承ChannelHandlerAdapter、重写channelRead、channelReadComplete、exceptionCaught；2，客户端：Bootstrap、线程组、继承ChannelHandlerAdapter、重写channelActive、channelRead、exceptionCaught。
实例：
public class TimeServer {
/**
* init一些thread、server等
* @param port
*/
public void bind(int port){
//配置服务端的NIO Reactor线程组
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b =new ServerBootstrap();
//绑定两个线程组，并设置IO事件的处理类，ChildChannelHandler
b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,1024).childHandler(new ChildChannelHandler());
//绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync();
//等待服务器监听端口关闭
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
//优雅退出，释放线程组
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
/**
* 子线程功能处理
*/
private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//指定handler
socketChannel.pipeline().addLast(new TimeServerHandler());
}
}
public static void main(String[] args) {
int port = 8090;
if(args !=null && args.length>0){
try{
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
new TimeServer().bind(port);
}
}
public class TimeServerHandler extends ChannelHandlerAdapter {
/**
* 异常处理
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
/**
* 处理逻辑
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println("The time server receive order : " + body);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
/**
* 处理完成后操作
* @param ctx
* @throws Exception
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
public class TimeClient {
/**
* connect server
* @param port
* @param host
*/
public void connect(int port , String host){
EventLoopGroup group =new NioEventLoopGroup();
try {
Bootstrap b =new Bootstrap();
b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
//发起异步连接操作
ChannelFuture f = b.connect(host,port).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
new TimeClient().connect(port,"127.0.0.1");
}
}
public class TimeClientHandler extends ChannelHandlerAdapter {
private static final Logger logger = Logger.getLogger(TimeClientHandler.class.getName());
private final ByteBuf firstMessage;
/**
* init
*/
public TimeClientHandler(){
byte[] req = "QUERY TIME ORDER".getBytes();
firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
}
/**
* some exception appear
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
logger.warning("Unexpected exception from downstream ：" + cause.getMessage());
ctx.close();
}
/**
* 1，connect success first do
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(firstMessage);
}
/**
* 2,receive data response to handler
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String (req,"UTF-8");
System.out.println("Now is : " + body);
}
}
好，Demo看完后，我们想TCP粘包和拆包问题，首相TCP流协议不关心业务，但是TCP缓冲区是有固定大小的，由于发送业务的大小和缓冲器的大小不一，就会出现一个完整包被拆成几个包，也可能几个包被合并进行处理，这就是粘包（合并）、拆包（拆分）。下边通过这张图，看下其中的几种情况：
出现的情况：
1，服务端，分两次读取到两个独立的数据包，分别D1和D2，没有粘包和拆包；
2，服务端一次接受两个数据包，D1和D2粘合在一起，被称为TCP粘包；
3，服务端分两次读取到两个数据包，第一次为D1包和D2的部分，第二次为D2的剩余内容，被称为TCP拆包；
4，服务端分两次读取到两个数据包，第一次为D1的部分内容，第二次为D1包的剩余内容和D2的包。
解决思路：
1，消息定长，例如每个报文的大小为固定长度200字节，如果不够空位补空格；
2，在包尾增加回车换行符进行分割，例如FTP协议；
3，将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）的字段；
4，更复杂的应用层协议。
Netty通过什么解决这个问题呢？
1，LineBasedFrameDecoder：处理数据中有“\n”、“\r\n”的位置进行换行，支持配置单行的最大长度，如果连续读取到最大长度还是没有发现换行符，则剖出异常，并忽略之前读到的异常码流。
2，DelimiterBasedFrameDecoder：可以自动完成以分隔符做结束标志的消息的解码。
3，FixedLengthFrameDecoder：可以自动完成对定长消息的解码。
三者的用法类似，我们看一个例子即可：
public class EchoServer {
public void bind(int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//注意几种解码器
//socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
//socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(20));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new EchoServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
new EchoServer().bind(port);
}
}
public class EchoServerHandler extends ChannelHandlerAdapter {
int counter = 0;
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String) msg;
System.out.println("This is" + ++counter + "times receive client : [" + body + "]");
body += "$_";
ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
ctx.writeAndFlush(echo);
}
}
public class EchoClient {
public void connect(int port, String host) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//几种编码器
//socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
//socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(20));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new EchoClientHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
new EchoClient().connect(port, "127.0.0.1");
}
}
public class EchoClientHandler extends ChannelHandlerAdapter {
private int counter;
static final String ECHO_REQ = "HI,LIUJIAHAN,WELCOME TO STUDY NETTY.$_";
public EchoClientHandler() {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 100; i++) {
ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("this is " + ++counter + "times receive server:[" + msg + "]");
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
好，这篇是Netty的基础入门，还有粘包和拆包Netty提供的基本解决方案，下边我们会更深入的学习。