java架构之路-（netty专题）netty的编解码（出入战）与粘包拆包
小菜技术 人气:0上次回归：
上次博客我们主要说了netty的基本使用，都是一些固定的模式去写的，我们只需要关注我们的拦截器怎么去写就可以了，然后我们用我们的基础示例，改造了一个简单的聊天室程序，可以看到内部加了一个StringEncoder和StringDecoder，这个就是用来编解码我们字符串的，这次我们就来说说这个编解码。
编码&解码：
上次我们写的那个简单的聊天室程序大家还记得吧，内部加了两个类似拦截器的玩意。
ch.pipeline().addLast(new StringEncoder()); ch.pipeline().addLast(new StringDecoder());
一个是编码的，一个是解码的，也是借着这个东西，来个大家说一下我们的ChannelPipline，上次只是简单了说了一下我们的ChannelPipline内部存放了ChannelHandler，而且是双向链表的结构来存储的，我们这次来细化一下我们这个拦截器是怎么工作的。
内部大概是这个样子的，每次放入的时候有顺序的放置，暂时先说有顺序，后面我会详细解释这个什么时候顺序生效，记住是由由头开始放置，这里涉及到两个概念就是入站和出站。
就是说我们从客户端发送数据到服务端，叫做出站，会经过一系列的ChannelOutboundHandler，可以方便记忆为一系列的出站拦截器，我们想出站，就要经过出站拦截器。
反之我们的入站就是和出站相对应的，是由服务端发送过来的数据，经由我们的一系列ChannelInboundHandler，到达我们的客户端，其实出站入站你站在客户端的角度来看就很好理解了，我们客户端想发出去数据，就是出站，想进来数据（接收数据），就是入站，出站会经过out拦截器，入站会经过in拦截器。切记，是双向的，客户端和服务端不是共有一个ChannelPipline，而且这个出站和入站都是相对的，可能还是有一点抽象，我们来拿着我们聊天室的例子来看一下。
我们需要先明确我们的StringEncoder和StringDecoder是ChannelInboundHandler还是ChannelOutboundHandler。
StringEncoder是ChannelOutboundHandler，StringDecoder是ChannelInboundHandler，这回我们按照我们的代码画一下图。先弄一个客户端发送消息的。
简单解释一下，我们的服务端和客户端都有自己的ChannelPipline，我们的客户端要发送消息，相当于客户端是出站操作，我们要发送，数据外流，显然是数据要出去，出站操作啊， 出站要经过Encoder然后是我们自己的Clienthandler，走你，进入网络传输，对于我们的服务端来说，要接收数据，数据要进来，一定是入站操作啊，经过我们的Decoder，然后经过我们自己的Serverhandler，到达我们的服务端。
客户端往外发送消息，客户端是出站操作，经过Encoder，然后经过我们ServerHandler，进入网络，我们客户端是入站操作，经过Decoder，经过我们的ClientHandler，到达我们的服务端。这样说应该就理解了吧，你可以自己在decode和encode方法上打断点，自己调试一下，后面我会说源码，自己也是先熟悉一下源码。出站一定是从尾到头，入站一定是从头到尾，别问我为什么，我自己写了测试类，测试一下午了.....
我们可以看到Handler是按照顺序执行的，这个顺序只是对于相同类型的Handlery有效果的，像我们的Decoder和Encoder，一个是入站的Handler，一个是出站的Handler，他俩谁在前，谁在后无所谓的。
粘包&拆包：
粘包和拆包比上面那个出入战好理解很多很多，我们先来看一段代码。
package com.xiaocai.packing;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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;
import io.netty.util.CharsetUtil;
public class NettyClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();// 开启工作线程组
try {
Bootstrap bootstrap = new Bootstrap(); //创建一个和服务端相对应的server
bootstrap.group(group) //设置线程组
.channel(NioSocketChannel.class) //使用NioSocketChannel作为客户端的通道实现
.handler(new ChannelInitializer<SocketChannel>() {//设置回调函数
@Override
protected void initChannel(SocketChannel ch) {
}
});
ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();//启动客户端去连接服务器端
//对通道关闭进行监听
System.out.println("netty client start。。准备开始发送数据");
for (int i = 0; i < 2000; i++) {
ByteBuf buf = Unpooled.copiedBuffer("hello，xiaocaiJAVA!".getBytes(CharsetUtil.UTF_8));
cf.channel().writeAndFlush(buf);
}
System.out.println("发送数据完毕");
cf.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();//关闭线程组
}
}
}
就是什么意思呢？我们建立一个客户端连接，然后我们多次向我们的服务端发送消息，理论上我们每次收到的消息都应该是hello，xiaocaiJAVA!，我们来看一下结论。
我们可以看到，有部分是正常的，有一部分是hello，xiaocaiJAVA!hello，xiaocaiJAVA!有的还是o，什么什么，这个明显错误的，也就是我们的粘包拆包，为什么会出现这个呢？netty收到我们的消息不是马上发送出去，大概会等待一个瞬间，然后再发送我们的消息，在等待的瞬间再次进来的消息，他会一次性的发送出去，但是netty自身并不知道我们的消息该从何位置截断，所以就出现了我们看到的粘包拆包问题，我们来看一下解决方法。
我们每次可以把数据发过去，而且把数据的长度带过去就OK了，然后客户端每次优先判断一下数据的长度就可以了，看一下我这的解决方案。
package com.xiaocai.packing;
/**
* 自定义协议包
*/
public class MyMessageProtocol {
//定义一次发送包体长度
private int len;
//一次发送包体内容
private byte[] content;
public int getLen() {
return len;
}
public void setLen(int len) {
this.len = len;
}
public byte[] getContent() {
return content;
}
public void setContent(byte[] content) {
this.content = content;
}
}
package com.xiaocai.packing;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class MyMessageEncoder extends MessageToByteEncoder<MyMessageProtocol> {
@Override
protected void encode(ChannelHandlerContext ctx, MyMessageProtocol msg, ByteBuf out) throws Exception {
out.writeInt(msg.getLen());
out.writeBytes(msg.getContent());
}
}
package com.xiaocai.packing;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;
public class MyMessageDecoder extends ByteToMessageDecoder {
int length = 0;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in.readableBytes() >= 4) {
if (length == 0){
length = in.readInt();
}
if (in.readableBytes() < length) {
System.out.println("当前可读数据不够，继续等待。。");
return;
}
byte[] content = new byte[length];
if (in.readableBytes() >= length){
in.readBytes(content);
//封装成MyMessageProtocol对象，传递到下一个handler业务处理
MyMessageProtocol messageProtocol = new MyMessageProtocol();
messageProtocol.setLen(length);
messageProtocol.setContent(content);
out.add(messageProtocol);
}
length = 0;
}
}
}
package com.xiaocai.packing;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
public class PackingServerHandler extends SimpleChannelInboundHandler<MyMessageProtocol> {
private int count;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MyMessageProtocol msg) throws Exception {
System.out.println("====服务端接收到消息如下====");
System.out.println("长度=" + msg.getLen());
System.out.println("内容=" + new String(msg.getContent(), CharsetUtil.UTF_8));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
package com.xiaocai.packing;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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;
import io.netty.util.CharsetUtil;
public class PackingClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new MyMessageEncoder());
}
});
ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
for(int i = 0; i< 200; i++) {
String msg = "你好，小菜JAVA！";
//创建协议包对象
MyMessageProtocol messageProtocol = new MyMessageProtocol();
messageProtocol.setLen(msg.getBytes(CharsetUtil.UTF_8).length);
messageProtocol.setContent(msg.getBytes(CharsetUtil.UTF_8));
cf.channel().writeAndFlush(messageProtocol);
}
cf.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
简单来说就是我们封装一个协议包，包含我们的内容和长度，我们要客户端传消息之前，信息进行处理，给予长度，然后我客户端接收数据时优先判断长度，长度不够继续等待，这样就解决了我们的拆包粘包问题，有的人还会提出用什么特殊符号的方法，也是可行的，但是你的数据中一定不要包含那个特殊符号，而且每次来一个新的开发人员，都要了解你们的特殊符号协议，还是比较麻烦的。
总结：
这次我们主要说了ChannelPipline内部的结构和addList时的放置顺序，netty的入战出战，是相对的，出站走out拦截器，入站走in拦截器，入站一定是从头到尾的，出站一定是从尾到头的，切记~！！！
最进弄了一个公众号，小菜技术，欢迎大家的加入
加载全部内容