编写一个网络应用程序需要实现某种编解码器,编解码器的作用就是讲原始字节数据与自定义的消息对象进行互转。网络中都是以字节码的数据形式来传输数据的,服务器编码数据后发送到客户端,客户端需要对数据进行解码,因为编解码器由两部分组成:
- Decoder(解码器)
- Encoder(编码器)
解码器负责处理“入站”数据,编码器负责处理“出站”数据。编码器和解码器的结构很简单,消息被编码后解码后会自动通过 ReferenceCountUtil.release(message)释放。
需要补充说明的是,Netty 中有两个方向的数据流
-
入站(ChannelInboundHandler):从远程主机到用户应用程序则是“入站(inbound)”
-
出站(ChannelOutboundHandler):从用户应用程序到远程主机则是“出站(outbound)”
今天我们主要学习编码器,也就是 Encoder
实现逻辑
完成一个编码器的编写主要是实现一个抽象类 MessageToMessageEncoder,其中我们需要重写方法是
/**
* Encode from one message to an other. This method will be called for each written message that can be handled
* by this encoder.
*
* @param ctx the {@link ChannelHandlerContext} which this {@link MessageToMessageEncoder} belongs to
* @param msg the message to encode to an other one
* @param out the {@link List} into which the encoded msg should be added
* needs to do some kind of aggragation
* @throws Exception is thrown if an error accour
*/
protected abstract void encode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;
其中泛型参数 I 表示我们需要接收的参数类型,如你需要将 ByteBuf 类型转换为 Date 类型,那么泛型 I 就是 ByteBuf (事实上当实现ByteBuf编码为其他类型的时候是不需要使用MessageToMessageEncoder,Netty提供了ByteToMessageCodec,其本质也是实现了MessageToMessageEncoder)
代码编写
需求说明
客户端发过来一个数字(ByteBuf),我们将此类型转换为数字,获取当前时间加上此数字的时间后返回客户端,具体逻辑如下:
![](http://static.zhoutao123.com/18261b926b654e57fd48e19d3022c215.png)
编码器的编写
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.time.LocalDateTime;
import java.util.List;
public class TimeEncoder extends MessageToMessageDecoder<ByteBuf> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
//将ByteBuf转换为String,注意此处,我是Mac OS,数据结尾是\r\n,如果是其他类型的OS,此处可能需要调整
String dataStr = msg.toString(CharsetUtil.UTF_8).replace("\r\n","");
//将String转换为Integer
Integer dataInteger = Integer.valueOf(dataStr);
//获取当前时间N小时后的数据
LocalDateTime now = LocalDateTime.now();
LocalDateTime dataLocalDatetime = now.plusHours(dataInteger);
out.add(dataLocalDatetime);
}
}
服务端处理代码
此处的服务端 HandleAdapter 和前面两个章节的 HandleAdapter 有所区别的是:其继承了 SimpleChannelInboundHandler<I>
并且传递了一个泛型参数,这里需要说明一下,SimpleChannelInboundHandler 是 ChannelInboundHandler 一个子类,他能够自动帮我们处理一些数据,在 ChannelInboundHandler 中,我们使用 channel 方法来接收数据,那么在 SimpleChannelInboundHandler 中我们使用 protected abstract void messageReceived(ChannelHandlerContext ctx, I msg) throws Exception;
来接收客户端的参数,可以看到的是,其参数中自动的实现了我们需要处理的泛型 I msg,另外看一下 SimpleChannelInboundHandler 中 channelRead 方法的实现代码
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
boolean release = true;
try {
//acceptInboundMessage() 检查参数的类型是否和设定的泛型是否匹配
//可以看到匹配的话,会进行强制类型转换并调用messageReceived方法
//否则的话,不执行,也就是说,这里的泛型一定要和编码器转换的结果类型一致,否则将接收不到参数
//当前如果你需要自己转换,那么你也可以和ChannelInBoundHandleAdapter一样,重写channelRead方法
if (acceptInboundMessage(msg)) {
@SuppressWarnings("unchecked")
I imsg = (I) msg;
messageReceived(ctx, imsg);
} else {
release = false;
ctx.fireChannelRead(msg);
}
} finally {
if (autoRelease && release) {
ReferenceCountUtil.release(msg);
}
}
}
那么继续实现我们的 HandleAdapter,代码非常简单,这里不再赘述。需要注意的是,我们这里没有做解码器,也就是说入站的时候需要 ByteBuf 类型的数据,因此使用 channel.writeAndFlush(Object)的时候,需要的就是 ByteBuf 类型的数据类型(当然如果 pipeline 中添加了 StringDecoder 解码器,那么你就可以直接使用字符串类型的数据了)
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.nio.charset.Charset;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
public class TimeServerChannelHandleAdapter extends SimpleChannelInboundHandler<LocalDateTime> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("添加了新的连接信息:id = " + ctx.channel().id());
}
@Override
protected void messageReceived(ChannelHandlerContext ctx, LocalDateTime msg) throws Exception {
// 转换时间格式
DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String dateFormat = msg.format(dateTimeFormatter);
System.out.println("接收到参数:" + dateFormat);
ctx.channel().writeAndFlush(Unpooled.copiedBuffer(dateFormat, Charset.defaultCharset()));
}
}
服务端启动代码
服务前启动代码和以前的代码非常类似,只需要在 pipeline 添加上适配的编码器即可,当然需要注意顺序(这个知识点以后我在仔细的阐述)
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class TimeServer {
public void start() throws Exception {
EventLoopGroup boosGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap
.group(boosGroup, workGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childHandler(
new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//设置编码
pipeline.addLast(new TimeEncoder());
//设置服务处理
pipeline.addLast(new TimeServerChannelHandleAdapter());
}
});
ChannelFuture sync = bootstrap.bind(9998).sync();
System.out.println("Netty Server start with 9998 port");
sync.channel().closeFuture().sync();
} finally {
workGroup.shutdownGracefully();
boosGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
TimeServer server = new TimeServer();
server.start();
}
}
效果展示
这里为了不写太多的代码,防止造成知识的不理解,迷惑,这里我们使用 telnet 命令来测试数据,
启动服务器端
运行 TimeServer 代码的中的 main 方法即可
![](http://static.zhoutao123.com/8289041c83dca541f77e1706fe2908f3.png)
使用 Telnet 发送数据
telnet 的命令格式是
usage: telnet [-l user] [-a] [-s src_addr] host-name [port]
可以看到大部分参数都是可选的,只有主机名称必填
![](http://static.zhoutao123.com/a99f6b1b2df1cb49f916a5d71f5fe99b.png)
发送数据的效果
继续在 telnet 中发送一个数据 5,我们分别看下服务端的打印的数据和 telnet 接收到的数据
服务端打印的数据如下:
![](http://static.zhoutao123.com/48e65b3dbdfd1dd00228664d2337ff65.png)
telnet 端打印的接收到的数据
![](http://static.zhoutao123.com/76f9ded78f16e2dad072850a688ad3fc.png)
总结
至此,一个简单的编码器就完成,我们总结一下步骤
- 继承 MessageToMessageDecoder 抽象类,实现 decode()方法
- 配置 HandleAdapter 实现 channelRead 或者 messageReceived 方法
- 配置服务启动类,配置 ChannelPipeline,添加编码器和 HandleAdapter
- 编写客户端或者使用 telnet 或者其他手段测试
欢迎来到这里!
我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。
注册 关于