Netty 入门与实战 (三) 自定义编码器

编写一个网络应用程序需要实现某种编解码器，编解码器的作用就是讲原始字节数据与自定义的消息对象进行互转。网络中都是以字节码的数据形式来传输数据的，服务器编码数据后发送到客户端，客户端需要对数据进行解码，因为编解码器由两部分组成：

• 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),我们将此类型转换为数字,获取当前时间加上此数字的时间后返回客户端，具体逻辑如下:

编码器的编写

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 方法即可

使用 Telnet 发送数据

telnet 的命令格式是

usage: telnet [-l user] [-a] [-s src_addr] host-name [port] 
​

可以看到大部分参数都是可选的，只有主机名称必填

发送数据的效果

继续在 telnet 中发送一个数据 5，我们分别看下服务端的打印的数据和 telnet 接收到的数据

服务端打印的数据如下:

telnet 端打印的接收到的数据

总结

至此，一个简单的编码器就完成，我们总结一下步骤

• 继承 MessageToMessageDecoder 抽象类，实现 decode()方法
• 配置 HandleAdapter 实现 channelRead 或者 messageReceived 方法
• 配置服务启动类，配置 ChannelPipeline，添加编码器和 HandleAdapter
• 编写客户端或者使用 telnet 或者其他手段测试