开源物联网平台Thingsboard——Netty实现与硬件设备通讯

1.Netty的业务场景

​ 平台主要需求是和充电桩对接，并定时对设备进行监控检查，需要使用Netty作为通信中间件来监听端口，充电桩通过TCP连接向服务端发送指令，后台主要是通过netty的ChannelHandler来实现对硬件数据的接收和处理。

2. Netty的主要组件

2.1 Channel

​ Channel作为Netty网络通信的主体，可以看作是通讯的载体，主要有三个状态：打开、关闭、连接。

​ Channel主要的IO操作：读(read)、写(write)、连接(connect)、绑定(bind)，均为异步，也就是说在调用如上方法后，并不保证IO操作完成，但会在IO操作成功、失败或取消后，生成相应的记录保存在一个凭证中并返回。

2.2 ChannelHandler

​ 负责Channel中的逻辑处理，可针对性地拦截处理Channel负责的IO操作或事件，然后在它的ChannelPipeline中将其递交给下一个handler。ChannelHandler中有许多方法需要实现，一般通过继承ChannelHandlerAdapter来实现。

 

2.3 ChannelPipeline

​ Netty中，ChannelPipeline相当于ChannelHandler的容器，它们可用于拦截和处理channel事件，关系如下图：

 

 

​ ChannelPipeline相当于ChannelHandler的容器，channel事件消息在ChannelPipeline中传播流动，而ChannelHandler可以针对性地对事件进行拦截处理、传递、忽略或者终止。一个ChannelHandler会绑定一个ChannelHandlerContext对象，ChannelHandler会通过与其对应的Context对象和ChannelPipeline交互，比如向上或向下传递events，动态地修改ChannelPipeline，或者通过ChannelHandlerContext中的AttributeKeys存储与handler相关的信息。

3. Netty服务端的启动

	ServerBootstrap serverBootstrap = new ServerBootstrap();    //启动NIO服务的辅助启动类
            serverBootstrap.group(parentGroup, childGroup).channel(NIOServerSocketChannel.class)  //启动服务时, 通过反射创建一个NioServerSocketChannel对象
                    //服务器初始化时执行, 属于AbstracBootstrap的方法
                    .handler(new LoggingHandler(LogLevel.INFO))    //handler在初始化时就会执行，可以设置打印日志级别
                    .option(ChannelOption.SO_BACKLOG, 1024)      //设置tcp缓冲区, 可连接队列大小
                    .option(ChannelOption.SO_REUSEADDR, true)    //允许重复使用本地地址和端口
                    //客户端连接成功之后执行, 属于ServerBootstrap的方法,继承自AbstractBootstrap
                    .childOption(ChannelOption.SO_KEEPALIVE, true)    //两小时没有数据通信时, 启用心跳保活机制探测客户端是否连接有效
                    .childOption(ChannelOption.SO_REUSEADDR, true)
                    .childHandler(serverChannelInit);    //childHandler在客户端成功连接后才执行，实例化ChannelInitializer
            ChannelFuture cf = serverBootstrap.bind(port).sync();    //绑定端口, 添加异步阻塞等待服务器启动完成
            if (cf.isSuccess() == true) {
                logger.info("NettyServer启动成功");
            } else {
                logger.error("NettyServer启动失败", cf.cause());
            }
            cf.channel().closeFuture().sync();    //等待服务器套接字关闭

 

4. Netty中的编解码

4.1 解码器

​ 解码(decode)就是根据约定的协议格式，对二进制数据进行解析解码(decode)，这一功能由解码器(decoder)完成。这部分的主要工作是：确定协议、编写协议对应的解码器。Netty中有一套编解码框架，输入的数据由ChannelInboundHandler处理，自定义的解码器实际上就是这个接口的特殊实现类。

​ 对于解码器(decoder)，Netty主要提供了抽象基类ByteToMessageDecoder和MessageToMessageDecoder。

 

4.1.1 抽象类ByteToMessageDecoder

​ 用于将接收的二进制数据(Byte)解码，得到完整有效的请求报文(Message)。

​ 一般ByteToMessageDecoder解码内容后，会得到一个ByteBuf实例，每个ByteBuf实例都包含了一个完整的报文信息。可以直接把这些ByteBuf实例交给之后的ChannelInboundHandler处理，或将ByteBuf实例解析封装到不同的JAVA实例对象后，再交给它处理。不管哪一种情况，之后的ChannelInboundHandler在处理时不需要再考虑粘包、拆包问题。

​ ByteToMessageDecoder中常见的实现类：

FixedLengthFrameDecoder：定长协议解码器，可以指定固定字节数算一个完整报文

LineBasedFrameDecoder：行分隔符解码器，遇到n或者rn，则认为是一个完整的报文

DelimiterBasedFrameDecoder：分隔符解码器，与LineBasedFrameDecoder类似，但可以自己指定分隔符

LengthFieldBasedFrameDecoder：长度编码解码器，将报文划分为报文头/报文体，根据报文头中的Length字段确定报文体的长度，因此报文体的长度是可变的

JsonObjectDecoder：json格式解码器，当检测到匹配数量的"{" 、”}”或”[””]”时，则认为是一个完整的json对象或json数组

这些实现类，都只是将接收到的二进制数据解码，转成ByteBuf实例后直接交给之后的ChannelInboundHandler处理，并没有将ByteBuf实例中的信息封装到Java对象中，因为Netty并不清楚报文具体内容，以及需要封装到哪个Java对象，所以需要自己手动来解析ByteBuf实例并封装。

​ 可以自定义一个解码类继承ByteToMessageDecoder抽象类，重写它的decode方法：

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

​ 参数列表：

ByteBuf in：解码前的二进制数据

List out：解码后的有效报文列表，由于tcp可能出现的粘包问题，入参的in中可能含有多个有效报文，所以需要将解码后的报文添加到List中，或者可能出现拆包，那么in中的数据就不足以构成一个有效报文，这时无需向List中添加元素。

​ 解码时需要尤其注意的是，应该先判断是否能构成一整个有效报文，再调用ByteBuf的read方法来读取数据，通过与in.readableBytes比较，来判断in中可读字节数是否大于约定的基本数据帧长度，只有在大于等于的情况下，我们才进行解码，即读取指定长度的字节，添加到List中。

4.1.2 抽象类MessageToMessageDecoder

​ 用于将一个本身就包含完整报文信息的对象转成另一个Java对象。

​ ByteToMessageDecoder解码后将包含了报文信息的ByteBuf实例交给后面的ChannelInboundHandler处理，此时可以在ChannelPipeline中再添加一个MessageToMessageDecoder，将ByteBuf中的信息解析后封装到Java对象中，简化接下来的ChannelInboundHandler操作。或者是要将已经封装好的Java对象转成其他Java对象，所以会出现MessageToMessageDecoder之后接着另一个MessageToMessageDecoder的情况。比如，Tomcat将浏览器发送过来的二进制数据解析为HttpServletRequest对象后，我们还需要将其中的数据提取出来封装成自定义的POJO类，即将现有的Java对象（HttpServletRequest）转换成另一个Java对象（POJO类）。

​ 继承MessageToMessageDecoder抽象类，也是要重写它的decode方法：

protected abstract void decode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;

​ 参数列表：

I msg：配置需要进行解码的参数

List out：经过MessageToMessageDecoder解析后，得到的Java对象存入列表

​ 那么在ChannelPipieline中它们的处理顺序如下：

ChannelPipieline ch=...
ch.addLast(new ByteToMessageDecoder());//ByteToMessageDecoder实现类
ch.addLast(new MessageToMessageDecoder());//MessageToMessageDecoder实现类
ch.addLast(new MessageToMessageDecoder());
...

 

​ 需要注意的是，即便是指定MessageToMessageDecoder的传入类型为ByteBuf，也绝对不可以用它来代替ByteToMessageDecoder报文解析的工作，因为ByteToMessageDecoder的内部设计才是针对接收到的二进制数据进行解码，所以除了解码，它其中还有对尚不完整的报文进行拆包缓存的功能逻辑，这是MessageToMessageDecoder所不具备的。

​ 因此，通常会先用ByteToMessageDecoder解析报文以及粘拆包处理，得到完整有效的ByteBuf实例，之后再交由一个或多个MessageToMessageDecoder对ByteBuf实例中的数据进行解析并封装成POJO类。

4.2 编码器

​ 相对应地，在ChannelOutboundHandler接口下，Netty也提供了MessageToByteEncoder和MessageToMessageEncoder两个抽象类来完成编码。没有解码器的内部逻辑复杂，编码只要将数据转成约定的二进制格式发送即可，而解码器除了解析数据，还要处理粘拆包问题。

4.3 编码解码器Codec

​ Codec同时具备编解码功能，它同时实现了ChannelInboundHandler和ChannelOutboundHandler两个接口，因此数据的输入输出都能处理。

​ Netty提供了一个ChannelDuplexHandler适配器类，编解码器的抽象基类ByteToMessageCodec和MessageToMessageCodec都继承了它，整体继承关系如下：

 

​ ByteToMessageCodec中维护了一个ByteToMessageDecoder和一个MessageToByteEncoder实例，结合二者的功能，泛型参数I可指定接受的编码类型：

public abstract class ByteToMessageCodec<I> extends ChannelDuplexHandler {
    private final TypeParameterMatcher outboundMsgMatcher;
    private final MessageToByteEncoder<I> encoder;
    private final ByteToMessageDecoder decoder = new ByteToMessageDecoder(){…}
  
    ...
    protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
    protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

 

MessageToMessageCodec中维护了一个MessageToMessageDecoder和一个 MessageToMessageEncoder实例，结合二者的功能，泛型参数INBOUND_IN和OUTBOUND_IN分别表示需要解码和编码的数据类型：一个简单的总结：ByteToMessageCodec和MessageToMessageCodec中分别实现了字节和对象之间、对象和对象之间的编解码。