基于 Netty 的自定义帧高可靠性读取方案

发表于 2017-03-26 更新于 2020-08-09 分类于 Java 阅读次数：

本文采用 Netty 这一最流行的 Java NIO 框架，作为 Java 服务器通信部分的基础框架，探索使用一个通道、一台服务器对多个客户端提供服务。

完成客户端 - 服务器通信，需要基于 TCP 协议之上，自定义一套简单的通信协议，其中数据交换方式需要使用自定义帧。为实现以上方案，本文采用 Netty 框架实现 Java 服务器的通信部分。

Netty 是由 JBoss 提供的一个 Java开源框架。Netty 提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

也就是说，Netty 是一个基于 NIO 的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户，服务端应用。Netty 相当简化和流线化了网络应用的编程开发过程，例如，TCP 和 UDP 的 socket 服务开发。

本项目的硬件设备集群使用 CAN 总线作为通信协议，硬件设备产生的数据和工作人员的控制指令均由服务器后端应用程序处理并存储。由于服务器并未原生支持 CAN 总线，故为了方便起见，使用「CAN转以太网」模块作为 CAN 协议和 TCP 协议交换的中介，以谋求实现的简单化。项目总体架构图如下图所示：

CAN - bus，即控制器局域网，是国际上应用最广泛的现场总线之一。

作为一种技术先进、可靠性高、功能完善、成本合理的远程网络通信控制方式，CAN - bus 已经被广泛应用到各个自动化控制系统中。从高速的网络到低价位的多路接线都可以使用 CAN - bus。例如，在汽车电子、自动控制、智能大厦、电路系统、安防监控等领域。

1. Netty 框架的学习

以下提供几篇不错的文章，帮助大家学习 Netty 这一颇受瞩目的框架。

2. Bootstrapping 服务器方案

以下代码是 Bootstrapping 服务器的实现方案：

public class KyServer{
  private SuccessfulListener launchListener;
  private SuccessfulListener finishListener;
  private NioEventLoopGroup group;

  public void start() {
    new Thread(() -> {
      group = new NioEventLoopGroup();
      ServerBootstrap bootstrap = new ServerBootstrap();
      bootstrap.group(group)
          .channel(NioServerSocketChannel.class)
          .childHandler(new ServerChannelInitializerTest());
      ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress(30232));
      channelFuture.addListener(
          (ChannelFutureListener) future -> startListenerHandle(future, launchListener));
    }).start();
  }

  private void startListenerHandle(Future future, SuccessfulListener listener) {
    if (!future.isSuccess()) future.cause().printStackTrace();
    if (listener != null) listener.onSuccess(future.isSuccess());
  }

  public void setLaunchSuccessfulListener(
      SuccessfulListener successfulListener) {
    this.launchListener = successfulListener;
  }

  public void setFinishSuccessfulListener(
      SuccessfulListener finishListener) {
    this.finishListener = finishListener;
  }

  public void shutdown() {
    if (group != null) {
      Future<?> futureShutdown = group.shutdownGracefully();
      futureShutdown.addListener(future -> startListenerHandle(future, finishListener));
    }
  }
}

2.1 Bootstrapping 服务器的设计要点

创建一个 ServerBootstrap 实例来启动和绑定服务器
创建并且分配一个 NioEventLoopgroup 实例来处理 event，比如接受新的连接和读/写数据
指定本地 InetSocketAddress 到服务器绑定的端口
用 ChannelHandler 实例来初始化 Channel
调用 ServerBootstrap.bind() 来绑定服务器

2.2 服务器监听器的设计「观察者模式」

首先在该类中设置成员变量：

1 2	private SuccessfulListener launchListener; private SuccessfulListener finishListener;

而后添加该变量的 set 方法，以及监听器的处理方法：

private void startListenerHandle(Future future, SuccessfulListener listener) {
    if (!future.isSuccess()) future.cause().printStackTrace();
    if (listener != null) listener.onSuccess(future.isSuccess());
}

public void setLaunchSuccessfulListener(SuccessfulListener successfulListener) {
    this.launchListener = successfulListener;
}

public void setFinishSuccessfulListener(SuccessfulListener finishListener) {
    this.finishListener = finishListener;
}

在服务器启动监听时，执行如下代码

1 2	ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress(30232)); channelFuture.addListener(future -> startListenerHandle(future, launchListener));

在外部关闭服务器时，执行该方法：

public void shutdown() {
    if (group != null) {
        Future<?> futureShutdown = group.shutdownGracefully();
        futureShutdown.addListener(future -> startListenerHandle(future, finishListener));
    }
}

通过如上方法，外部操作者可以方便得知服务器是否启动成功以及是否结束成功，使用观察者模式，完美实现了对服务器启动及关闭的监听。

3. 服务器业务逻辑的实现

首先使用初始化方法 ServerChannelInitializer 完成所有 ChannelHandler 对 Channel 的绑定操作：

public class ServerChannelInitializer extends ChannelInitializer<NioSocketChannel> {

    @Override
    protected void initChannel(NioSocketChannel ch) {
        ch.pipeline().addLast(new LoggingHandler("NO1"));
        byte head = 0x11;
        ch.pipeline().addLast(new FrameIdentifierChannelInboundHandler(head));
        ch.pipeline().addLast(new ShowByteBufAsFrameInBoundHandler());
    }
}

3.1 输入数据处理方案

3.1.1 自定义帧方案

自定义帧包括「帧标识位」、「数据长度」、「数据体」，如下图所示，：

帧标识位：0x11。
数据长度：两个字节，可表示数据部分大小最大为 2 ^ 16 - 1 。
数据体：实际有用的数据。

3.1.2 输入数据处理器

以下为输入数据的第一个处理器，可以保证无论 TCP 帧经历怎样的粘包、拆包，均可以准确提取每一个自定义帧的数据部分。

/**
 * 入端自定义帧提取处理器
 * 将数据流提取出完整的自定义帧并传入下一个处理器
 */
public class FrameIdentifierChannelInboundHandler extends SimpleChannelInboundHandler<ByteBuf> {
  private byte[] frameHead;
  private int frameHeadLength;
  private int frameBodyLength;
  private FrameReceivedEnum frameStatus = FrameReceivedEnum.READY;
  private ByteBuf holdByteBuf = Unpooled.buffer(1024);

  FrameIdentifierChannelInboundHandler(byte... frameHead) {
    this();
    this.frameHead = frameHead;
    frameHeadLength = frameHead.length;
  }

  @Override
  protected void channelRead0 (ChannelHandlerContext ctx, ByteBuf msg) {
    //数据读入本地buffer
    holdByteBuf.writeBytes(msg);
    while (true) {
      //若读取状态为: 开始读取
      if (frameStatus == FrameReceivedEnum.READY) {
        if (!matchFrameHead(holdByteBuf)) {
          holdByteBuf.clear();
          break;
        }
      }

      //若读取状态为: 帧长读取
      //数据体完全包含在 buffer 内，则可通过此状态
      if (frameStatus == FrameReceivedEnum.READING_LENGTH) {
        if (holdByteBuf.readableBytes() <= 1) break;
        //无符号 short 需要用 int 型引用
        int currentFrameLength = holdByteBuf.getUnsignedShort(holdByteBuf.readerIndex());

        //可读byte数为长度计数(2)+数据体长度; 所以当前帧长+2 <= 可读帧长
        if (currentFrameLength + 2 <= holdByteBuf.readableBytes()) {
          frameBodyLength = holdByteBuf.readUnsignedShort();
          frameStatus = FrameReceivedEnum.READING_BODY;
        } else {
          break;
        }
      }

      //若读取状态为: 数据体读取
      //预设数据体完全包含在buffer内，否则抛出异常
      if (frameStatus == FrameReceivedEnum.READING_BODY) {
        if (frameBodyLength == 0) {
          frameStatus = FrameReceivedEnum.READY;
          frameBodyLength = -1;
          holdByteBuf.discardReadBytes();
        } else if (frameBodyLength > 0) {
          ByteBuf returnBuf = Unpooled.buffer(frameBodyLength);
          holdByteBuf.readBytes(returnBuf);
          frameStatus = FrameReceivedEnum.READY;
          //    ctx.fireChannelRead(returnBuf);
          ctx.writeAndFlush(returnBuf);
          frameBodyLength = -1;
          holdByteBuf.discardReadBytes();
        } else {
          throw new FrameLoadException("自定义帧长度计数异常");
        }
      } else {
        throw new FrameLoadException("自定义帧读取异常");
      }
    }
  }

  private boolean matchFrameHead(ByteBuf byteBuf) {
    while (true) {
      if (byteBuf.readableBytes() < frameHeadLength) {
        return false;
      }
      if (frameHead[0] == byteBuf.readByte()) {
        frameStatus = FrameReceivedEnum.READING_LENGTH;
        return true;
      }
    }
  }
}

以下为第二个输入数据处理器，可将前一处理器的结果「优雅」打印到控制台上并原样发送至客户端：

public class ShowByteBufAsFrameInBoundHandler extends SimpleChannelInboundHandler<ByteBuf> {
  
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        System.out.println(ByteBufUtil.prettyHexDump(byteBuf));
        ctx.pipeline().writeAndFlush(Unpooled.copiedBuffer(msg));
    }
}

比特麒麟