前言
我们今天继续来分析 Netty 的编解码器，这次我们要自己动手实现自定义的编码器、解码器和编解码器。
自定义基于换行的解码器
LineBasedFrameDecoder 类
LineBasedFrameDecoder 类是基于换行的，意味着只要在接收数据时遇到以换行符\n或者回车换行符\r\n结尾时，就表明数据已经接收完成可以被处理了。
LineBasedFrameDecoder 类继承自 ByteToMessageDecoder，并重写了 decode方法。
public class LineBasedFrameDecoder extends ByteToMessageDecoder {
private final int maxLength;
private final boolean failFast;
private final boolean stripDelimiter;
private boolean discarding;
private int discardedBytes;
private int offset;
public LineBasedFrameDecoder(final int maxLength) {
this(maxLength, true, false);
}
public LineBasedFrameDecoder(final int maxLength, final boolean stripDelimiter, final boolean failFast) {
this.maxLength = maxLength;
this.failFast = failFast;
this.stripDelimiter = stripDelimiter;
}
@Override
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
Object decoded = decode(ctx, in);
if (decoded != null) {
out.add(decoded);
}
}
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
final int eol = findEndOfLine(buffer);
if (!discarding) {
if (eol >= 0) {
final ByteBuf frame;
final int length = eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
if (length > maxLength) {
buffer.readerIndex(eol + delimLength);
fail(ctx, length);
return null;
}
if (stripDelimiter) {
frame = buffer.readRetainedSlice(length);
buffer.skipBytes(delimLength);
} else {
frame = buffer.readRetainedSlice(length + delimLength);
}
return frame;
} else {
final int length = buffer.readableBytes();
if (length > maxLength) {
discardedBytes = length;
buffer.readerIndex(buffer.writerIndex());
discarding = true;
offset = 0;
if (failFast) {
fail(ctx, "over " + discardedBytes);
}
}
return null;
}
} else {
if (eol >= 0) {
final int length = discardedBytes + eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
buffer.readerIndex(eol + delimLength);
discardedBytes = 0;
discarding = false;
if (!failFast) {
fail(ctx, length);
}
} else {
discardedBytes += buffer.readableBytes();
buffer.readerIndex(buffer.writerIndex());
offset = 0;
}
return null;
}
}
private void fail(final ChannelHandlerContext ctx, int length) {
fail(ctx, String.valueOf(length));
}
private void fail(final ChannelHandlerContext ctx, String length) {
ctx.fireExceptionCaught(
new TooLongFrameException(
"frame length (" + length + ") exceeds the allowed maximum (" + maxLength + ')'));
}
private int findEndOfLine(final ByteBuf buffer) {
int totalLength = buffer.readableBytes();
int i = buffer.forEachByte(buffer.readerIndex() + offset, totalLength - offset, ByteProcessor.FIND_LF);
if (i >= 0) {
offset = 0;
if (i > 0 && buffer.getByte(i - 1) == '\r') {
i--;
}
} else {
offset = totalLength;
}
return i;
}
}
从上述代码可以看出，LineBasedFrameDecoder是通过查找回车换行符来找到数据结束的标志的。
定义解码器
定义了解码器MyLineBasedFrameDecoder，该解码器继承自LineBasedFrameDecoder，因此可以使用LineBasedFrameDecoder上的所有功能。
代码如下：
public class MyLineBasedFrameDecoder extends LineBasedFrameDecoder {
private final static int MAX_LENGTH = 1024;
public MyLineBasedFrameDecoder() {
super(MAX_LENGTH);
}
}
在上述代码中，通过MAX_LENGTH常量，来限制解码器帧的大小。超过该常量值的限制的话，则会抛出TooLongFrameException异常。
定义 ChannelHandler
ChannelHandler 定义如下：
public class MyLineBasedFrameDecoderServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
System.out.println("Client -> Server: " + msg);
}
}
MyLineBasedFrameDecoderServerHandler业务非常简单，把收到的消息打印出来即可。
定义 ChannelInitializer
定义一个 ChannelInitializer 用于容纳解码器 MyLineBasedFrameDecoder 和 MyLineBasedFrameDecoderServerHandler，代码如下：
public class MyLineBasedFrameDecoderChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new MyLineBasedFrameDecoder());
channel.pipeline().addLast(new StringDecoder());
channel.pipeline().addLast(new MyLineBasedFrameDecoderServerHandler());
}
}
注意添加到ChannelPipeline的ChannelHandler的顺序，MyLineBasedFrameDecoder 在前，MyLineBasedFrameDecoderServerHandler 在后，意味着数据先经过MyLineBasedFrameDecoder解码，然后再交给MyLineBasedFrameDecoderServerHandler处理。
StringDecoder实现将数据转换为字符串。
编写服务器
定义服务器 MyLineBasedFrameDecoderServer代码如下：
public class MyLineBasedFrameDecoderServer {
public static int DEFAULT_PORT = 8023;
public static void main(String[] args) throws Exception {
int port = DEFAULT_PORT;
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new MyLineBasedFrameDecoderChannelInitializer())
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture f = b.bind(port).sync();
System.out.println("MyLineBasedFrameDecoderServer已启动，端口：" + port);
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
MyLineBasedFrameDecoderServer 中唯一需要注意的是，在 ServerBootstrap 中指定MyLineBasedFrameDecoderChannelInitializer，这样服务器就能应用咱们自定义的编码器和ChannelHandler了。
编写客户端
为了测试服务器，编写了一个简单的 TCP 客户端，代码如下：
public class TcpClient {
public static void main(String[] args) throws IOException {
Socket socket = null;
OutputStream out = null;
try {
socket = new Socket("localhost", 8023);
out = socket.getOutputStream();
String lines = "床前明月光\r\n疑是地上霜\r\n举头望明月\r\n低头思故乡\r\n";
byte[] outputBytes = lines.getBytes("UTF-8");
out.write(outputBytes);
out.flush();
} finally {
out.close();
socket.close();
}
}
}
上述客户端在启动后会发送一段文本，而后关闭连接。该文本每句用回车换行符\r\n结尾，这样服务器就能一句一句地解析了。
测试
先启动服务器，观察控制台，可以看到如下输出的内容：
MyLineBasedFrameDecoderServer已启动，端口：8023
然后启动客户端。启动完成之后，再次观察服务器的控制台，可以看到如下输出内容：
MyLineBasedFrameDecoderServer已启动，端口：8023
Client -> Server: 床前明月光
Client -> Server: 疑是地上霜
Client -> Server: 举头望明月
Client -> Server: 低头思故乡
上述的输出内容说明，MyLineBasedFrameDecoderServerHandler接收到了 4 条数据。那么为啥客户端发送了 1 条数据，到这里就变成了 4 条了呢？这是因为在前面介绍的MyLineBasedFrameDecoderChannelInitializer中，MyLineBasedFrameDecoder先被添加到ChannelPipeline，然后才添加到MyLineBasedFrameDecoderServerHandler,意味着数据先经过解码，再交给MyLineBasedFrameDecoderServerHandler处理，而在数据解码过程中，MyLineBasedFrameDecoderServerHandler是按照换行解码的，而客户端所发送的数据里面又包含了 4 个回车换行符，因此，数据被解码为了 4 条。
自定义编码器
定义消息通信协议
消息通信协议是连接客户端和服务器的密语，只有熟知双方的通信协议，客户端和服务器才能正常识别消息的内容。常见的消息通信协议有 HTTP、MQTT、XMPP、STOMP、AMQP和 RTMP等。
下图展示了消息通信协议的内容格式：
| 类型 | 名称 | 字节序列 | 取值范围 | 备注 |
|---|
| 消息头 | msgType | 0 | 0x00-0xff | 消息类型 | | 消息头 | len | 1-4 | 0-2147483647 | 消息体长度 | | 消息体 | body | 变长 | 0- | 消息体 |
从上述协议中可以看出，消息主要是由消息头和消息体组成，并说明如下：
- msgType 表示消息的类型。在本节示例中，请求用
EMGW_LOGIN_REQ(0x00)，响应用EMGW_LOGIN_RES(0x01)表示。 - len 表示消息体的长度。
- body 表示消息体。
定义了如下MsgType枚举类型来表示消息类型：
public enum MsgType {
EMGW_LOGIN_REQ((byte) 0x00),
EMGW_LOGIN_RES((byte) 0x01);
private byte value;
public byte getValue() {
return value;
}
private MsgType(byte value) {
this.value = value;
}
}
消息头类 MsgHeader定义如下：
public class MsgHeader {
private byte msgType;
private int len;
public MsgHeader() {
}
public MsgHeader(byte msgType, int len) {
this.msgType = msgType;
this.len = len;
}
public byte getMsgType() {
return msgType;
}
public void setMsgType(byte msgType) {
this.msgType = msgType;
}
public int getLen() {
return len;
}
public void setLen(int len) {
this.len = len;
}
}
消息类 Msg 定义如下：
public class Msg {
private MsgHeader msgHeader = new MsgHeader();
private String body;
public MsgHeader getMsgHeader() {
return msgHeader;
}
public void setMsgHeader(MsgHeader msgHeader) {
this.msgHeader = msgHeader;
}
public String getBody() {
return body;
}
public void setBody(String body) {
this.body = body;
}
}
定义编码器
public class MyEncoder extends MessageToByteEncoder<Msg> {
@Override
protected void encode(ChannelHandlerContext ctx, Msg msg, ByteBuf out) throws Exception {
if (msg == null | msg.getMsgHeader() == null) {
throw new Exception("The encode message is null");
}
MsgHeader header = msg.getMsgHeader();
String body = msg.getBody();
byte[] bodyBytes = body.getBytes(Charset.forName("utf-8"));
int bodySize = bodyBytes.length;
System.out.printf("MyEncoder header: %s, body: %s", header.getMsgType(), body);
out.writeByte(MsgType.EMGW_LOGIN_RES.getValue());
out.writeInt(bodySize);
out.writeBytes(bodyBytes);
}
}
MyEncoder会将 Msg 消息转为 ByteBuf 类型。
定义解码器
public class MyDecoder extends LengthFieldBasedFrameDecoder {
private static final int MAX_FRAME_LENGTH = 1024 * 1024;
private static final int LENGTH_FIELD_LENGTH = 4;
private static final int LENGTH_FIELD_OFFSET = 1;
private static final int LENGTH_ADJUSTMENT = 0;
private static final int INITIAL_BYTES_TO_STRIP = 0;
private static final int HEADER_SIZE = 5;
private byte msgType;
private int len;
public MyDecoder() {
super(MAX_FRAME_LENGTH,
LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH,
LENGTH_ADJUSTMENT, INITIAL_BYTES_TO_STRIP);
}
@Override
protected Msg decode(ChannelHandlerContext ctx, ByteBuf in2) throws Exception {
ByteBuf in = (ByteBuf) super.decode(ctx, in2);
if (in == null) {
return null;
}
if (in.readableBytes() < HEADER_SIZE) {
return null;
}
msgType = in.readByte();
len = in.readInt();
if (in.readableBytes() < len) {
return null;
}
ByteBuf buf = in.readBytes(len);
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
Msg msg = new Msg();
MsgHeader msgHeader = new MsgHeader(msgType, len);
msg.setBody(body);
msg.setMsgHeader(msgHeader);
return msg;
}
}
MyDecoder集成自 Netty 内嵌的解码器LengthFieldBasedFrameDecoder。LengthFieldBasedFrameDecoder是一种基于灵活长度的解码器。在数据包中，加了一个长度字段，保存上层包的长度。解码时，会按照这个长度，进行上层 ByteBuf 应用包的提取。其中，初始化LengthFieldBasedFrameDecoder时，需要指定以下参数：
- maxFrameLength：发送数据包最大的长度。
- lengthFieldOffset：长度域偏移量，指的是长度域位于整个数据包字节数组中的下标。
- lengthFieldLength：长度域的字节长度。
- lengthAdjustment：长度域的偏移量矫正。
- initialBytesToStrip：丢弃的初始字节数。丢弃处于有效数据前面的字节数量。
定义服务器 ChannelHandler
public class MyServerHandler extends SimpleChannelInboundHandler<Object> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object obj) throws Exception {
Channel incoming = ctx.channel();
if (obj instanceof Msg) {
Msg msg = (Msg) obj;
System.out.println("Client->Server:" + incoming.remoteAddress() + msg.getBody());
incoming.write(obj);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
MyServerHandler逻辑比较简单，只是把收到的消息内容打印出来。
定义客户端 ChannelHandler
public class MyClientHandler extends SimpleChannelInboundHandler<Object> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object obj) throws Exception {
Channel incoming = ctx.channel();
if (obj instanceof Msg) {
Msg msg = (Msg) obj;
System.out.println("Server->Client:" + incoming.remoteAddress() + msg.getBody());
} else {
System.out.println("Server->Client:" + incoming.remoteAddress() + obj.toString());
}
}
}
MyClientHandler逻辑比较简单，只是把收到的消息内容打印出来。
定义服务器的主程序
public class MyServer {
private int port;
public MyServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("decoder", new MyDecoder());
ch.pipeline().addLast("encoder", new MyEncoder());
ch.pipeline().addLast(new MyServerHandler());
}
}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);
ChannelFuture f = b.bind(port).sync();
System.out.println("Server start listen at " + port);
f.channel().closeFuture().sync();
} finally {
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 = 8082;
}
new MyServer(port).run();
}
}
注意添加到ChannelPipeline的ChannelHandler的顺序，MyDecoder在前，MyEncoder在后，业务处理的MyServerHandler在最后。
定义客户端主程序
public class MyClient {
private String host;
private int port;
public MyClient(String host, int port) {
this.host = host;
this.port = port;
}
public void run() throws InterruptedException {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("decoder", new MyDecoder());
ch.pipeline().addLast("encoder", new MyEncoder());
ch.pipeline().addLast(new MyClientHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
while (true) {
Msg msg = new Msg();
MsgHeader msgHeader = new MsgHeader();
msgHeader.setMsgType(MsgType.EMGW_LOGIN_REQ.getValue());
String body = "床前明月光，疑是地上霜。";
byte[] bodyBytes = body.getBytes(Charset.forName("utf-8"));
int bodySize = bodyBytes.length;
msgHeader.setLen(bodySize);
msg.setMsgHeader(msgHeader);
msg.setBody(body);
f.channel().writeAndFlush(msg);
Thread.sleep(2000);
}
} finally {
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new MyClient("localhost", 8082).run();
}
}
注意添加到ChannelPipeline的ChannelHandler的顺序，MyDecoder在前，MyEncoder在后，业务处理的MyClientHandler在最后。
上述的客户端程序，会每隔 2 秒给服务器发送一条消息。
测试
分别运行服务器和客户端程序。
客户端输出如下：
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
服务端输出如下：
Server start listen at 8082
Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:62927床前明月光，疑是地上霜。
自定义编解码器
前面我们实现了编码器 MyEncoder 和 解码器 MyDecoder。这些代码无须做任何改动。
自定义编解码器
使用CombinedChannelDuplexHandler 类对编码器 MyEncoder 和 解码器 MyDecoder进行组合。代码如下：
public class MyCodec extends CombinedChannelDuplexHandler<MyDecoder, MyEncoder> {
public MyCodec() {
super(new MyDecoder(), new MyEncoder());
}
}
使用编解码器
分别修改 MyServer 和 MyClient 类，添加编解码器，修改代码如下：
ch.pipeline().addLast("codec", new MyCodec());
上述代码将原来的 MyEncoder 和 MyDecoder从ChannelPipeline中剔除掉了，取而代之是MyEncoder。
测试
分别运行服务器和客户端。
客户端输出如下：
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
MyEncoder header: 0, body: 床前明月光，疑是地上霜。Server->Client:localhost/127.0.0.1:8082床前明月光，疑是地上霜。
服务端输出如下：
Server start listen at 8082
Client->Server:/127.0.0.1:56181床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:56181床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:56181床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:56181床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。Client->Server:/127.0.0.1:56181床前明月光，疑是地上霜。
MyEncoder header: 1, body: 床前明月光，疑是地上霜。
总结
以上就是关于一些自定义的编码器、解码器、编解码器的示例应用，我们下节继续深入 Netty 源码。
结尾
我是一个正在被打击还在努力前进的码农。如果文章对你有帮助，记得点赞、关注哟，谢谢！
|