Netty实现按字节解析的socket协议

Netty内部实现了很多通用协议的编码和解码。如果要实现自定义的协议，则需要自己实现编码或解码的功能。
继承ChannelInboundHandlerAdapter类，就可以实现一个自定义的解码器。但如果发送比较长的内容，则会出现内容读取不完整的问题。
其实比较简单的一个实现，就是设定协议头的几个字节为消息的长度即可，并在发送消息和处理消息时，处理消息的长度即可。
Server端的代码如下：
TcpServer.java

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class TcpServer {

private static final Logger logger = Logger.getLogger(TcpServer.class);
private static final String IP = "127.0.0.1";
private static final int PORT = 9999;

protected static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors()*2; //默认

protected static final int BIZTHREADSIZE = 4;

private static final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
private static final EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);

protected static void run() throws Exception {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ChannelInitializer() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
//                pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
//                pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
pipeline.addLast(new SocketByteHandler());
}
});

b.bind(IP, PORT).sync();
logger.info("TCP服务器已启动");
}

protected static void shutdown() {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}

public static void main(String[] args) throws Exception {
PropertyConfigurator.configure("log/log4j.properties");
logger.info("开始启动TCP服务器...");
TcpServer.run();
//      TcpServer.shutdown();
}
}

LengthFieldBasedFrameDecoder和LengthFieldPrepender就是设定协议头长度的，我这里设定协议头的长度为4个字节。

协议处理类：
SocketByteHandler.java

import java.nio.ByteBuffer;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class SocketByteHandler extends ChannelInboundHandlerAdapter {

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
// super.channelRead(ctx, msg);
ByteBuf result = (ByteBuf) msg;
byte[] result1 = new byte[result.readableBytes()];
// msg中存储的是ByteBuf类型的数据，把数据读取到byte[]中
result.readBytes(result1);
String resultStr = new String(result1);
System.out.println("Client said:" + resultStr);
// 释放资源，这行很关键
result.release();
String response = "I am ok!";
// 在当前场景下，发送的数据必须转换成ByteBuf数组
ByteBuf encoded = ctx.alloc().buffer(4 * response.length());
encoded.writeBytes(response.getBytes());
ctx.write(encoded);
ctx.flush();
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
// TODO Auto-generated method stub
super.channelReadComplete(ctx);
ctx.flush();
}

}

客户端可以使用最简单的socket来实现即可，如：

public static void sendMsgBySocket(byte[] msg){
try {
Socket socket = new Socket();
socket.connect(new InetSocketAddress(HOST,PORT));
socket.setKeepAlive(true);
OutputStream out = socket.getOutputStream();
ByteBuffer header = ByteBuffer.allocate(4);
header.putInt(msg.length);
out.write(header.array());
out.write(msg);
out.flush();
InputStream in = socket.getInputStream();
byte[] buff = new byte[4096];
int readed = in.read(buff);
if(readed > 0){
String str = new String(buff,4,readed);
logger.info("client received msg from server:"+str);
}
out.close();
} catch (UnknownHostException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}

可以看到，header中就是给出的要发送的消息内容的总长度，但不包含协议头。这样，server端收到后，会自动忽略掉协议头的内容，这就是Netty的好处。

客户端如果读取server端的返回内容，而且用基本的socket（非Netty），则需要自己处理协议头：
String str = new String(buff,4,readed);
跳过协议头的4个字节长度后，就是Server端返回的真正的内容。