Netty解锁接收物联网16进制数据的办法

因为和物联网设备对接的时候，16进制数据传输是普遍的传输方式，在物联网实际的项目中会频繁遇到，因此总结一份详细的攻略给大家，希望大家会喜欢

一般情况下物联网设备给服务器发送数据都会有一个开始符号和结束符号，我这边以开头字符为10 02 结束字符为 10 03的样例（开头和结束都是16进制），因为考虑到设备在断网后再接入可能会突然激增一大堆数据，如果不对数据进行分包拆包处理会导致业务数据不全，业务数据不全的情况下会导致真实数据出现乱码的情况，所以需要作出分包拆包逻辑

分包拆包逻辑如下：


import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
public class BootNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        //分割符
        ByteBuf delimiter = Unpooled.copiedBuffer(new byte[]{16,3});
        //打印分隔符
        ch.pipeline().addLast("frameDecoder", new DelimiterBasedFrameDecoder(1024,false, delimiter));
        // ChannelOutboundHandler，依照逆序执行
        ch.pipeline().addLast("encoder", new StringEncoder());
        // 属于ChannelInboundHandler，依照顺序执行
        ch.pipeline().addLast("decoder", new StringDecoder());
        /*ch.pipeline().addLast("decoder", new MyDecoder());*/
        /**
        * 自定义ChannelInboundHandlerAdapter
        */
        ch.pipeline().addLast(new BootNettyChannelInboundHandlerAdapter());
    }
}

如果大家的业务不需要做分包拆包则可把这两行去掉，这两行代码的意思是，以10 03为结尾的数据单独分出一个包让后台进行接收

      //分割符
        ByteBuf delimiter = Unpooled.copiedBuffer(new byte[]{16,3});
        //打印分隔符
        ch.pipeline().addLast("frameDecoder", new DelimiterBasedFrameDecoder(1024,false, delimiter));

BootNettyChannelInboundHandlerAdapter 类如下所示，在这里我做了一个很关键的处理，因为byte最多存储8位，当接收数字大于十进制的125（也就是16进制的79的时候，数据会出现乱码，如果要解决这个乱码的问题可以采用我下面这一个机制）：


        byte[] bytebuf=((String) msg).getBytes();
        // 此解码为解决 16进制在超过79的情况乱码问题
        for(int i = 0, len = bytebuf.length; i < len; i ++) {
            String hex =  Integer.toHexString(bytebuf[i] & 0xff).toUpperCase();
            message += hex.length()  == 1 ? "0" + hex : hex;
        }


import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.List;
import com.alibaba.fastjson.JSON;
import com.dao.CarDeviceRepository;
import com.dao.CarParkRepository;
import com.dao.DeviceToParkRepository;
import com.dto.ParkDto;
import com.entity.CarDevice;
import com.entity.CarPark;
import com.entity.DeviceToPark;
import com.netty.encrypt.HexConvert;
import com.util.HttpRequestUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
    public final Logger log = LoggerFactory.getLogger(BootNettyChannelInboundHandlerAdapter.class);
    private static final int TIMEOUT = 10 * 1000; //超时时间 10s
    /**
    * 从客户端收到新的数据时，这个方法会在收到消息时被调用
    *
    * @param ctx
    * @param msg
    */
    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception, IOException
    {
        // 解决16进制编解码问题
        String message = "";
        byte[] bytebuf=((String) msg).getBytes();
        // 此解码为解决 16进制在超过79的情况乱码问题
        for(int i = 0, len = bytebuf.length; i < len; i ++) {
            String hex =  Integer.toHexString(bytebuf[i] & 0xff).toUpperCase();
            message += hex.length()  == 1 ? "0" + hex : hex;
        }       
    }
    /**
    * 从客户端收到新的数据、读取完成时调用
    *
    * @param ctx
    */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws IOException
    {
        //System.out.println("channelReadComplete");
        ctx.flush();
    }

    /**
    * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
    *
    * @param ctx

    * @param cause
    */

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException
    {
        log.info("exceptionCaught");
        cause.printStackTrace();
        ctx.close();//抛出异常，断开与客户端的连接
    }

    /**
    * 客户端与服务端第一次建立连接时 执行
    *
    * @param ctx
    * @throws Exception
    */

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException
    {
        super.channelActive(ctx);
        ctx.channel().read();
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();
        //此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接
        log.info("channelActive:"+clientIp+ctx.name());
    }

    /**
    * 客户端与服务端 断连时 执行
    *
    * @param ctx
    * @throws Exception
    */

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException
    {
        super.channelInactive(ctx);
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();
        ctx.close(); //断开连接时，必须关闭，否则造成资源浪费，并发量很大情况下可能造成宕机
        log.info("channelInactive:"+clientIp);
    }

    /**
    * 服务端当read超时, 会调用这个方法
    *
    * @param ctx
    * @param evt
    * @throws Exception
    */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException
    {
        super.userEventTriggered(ctx, evt);
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = insocket.getAddress().getHostAddress();
        ctx.close();//超时时断开连接
        log.info("userEventTriggered:"+clientIp);
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception{
        log.info("channelRegistered");
    }
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception{
        log.info("channelUnregistered");
    }
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception{
        log.info("channelWritabilityChanged");
    }
}

BootNettyServer启动类如下所示：


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class BootNettyServer {
    public void bind(int port) throws Exception {
        /**

        * 配置服务端的NIO线程组

        * NioEventLoopGroup 是用来处理I/O操作的Reactor线程组

        * bossGroup：用来接收进来的连接，workerGroup：用来处理已经被接收的连接,进行socketChannel的网络读写，

        * bossGroup接收到连接后就会把连接信息注册到workerGroup

        * workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍

        */

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {

            /**

            * ServerBootstrap 是一个启动NIO服务的辅助启动类

            */

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            /**

            * 设置group，将bossGroup， workerGroup线程组传递到ServerBootstrap

            */

            serverBootstrap = serverBootstrap.group(bossGroup, workerGroup);

            /**
            * ServerSocketChannel是以NIO的selector为基础进行实现的，用来接收新的连接，这里告诉Channel通过NioServerSocketChannel获取新的连接
            */
            serverBootstrap = serverBootstrap.channel(NioServerSocketChannel.class);
            /**
            * option是设置 bossGroup，childOption是设置workerGroup
            * netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费    最小  初始化  最大 (根据生产环境实际情况来定)
            * 使用对象池，重用缓冲区
            */
            // 2097152
            //serverBootstrap = serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576));
            //serverBootstrap = serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576));
            serverBootstrap = serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 20971520));
            serverBootstrap = serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 20971520));
            /**
            * 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
            */
            serverBootstrap = serverBootstrap.childHandler(new BootNettyChannelInitializer<SocketChannel>());
            System.out.println("netty server start success!");
            /**
            * 绑定端口，同步等待成功
            */
            ChannelFuture f = serverBootstrap.bind(port).sync();
            /**
            * 等待服务器监听端口关闭
            */
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
        } finally {
            /**
            * 退出，释放线程池资源
            */
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}

最后只需要在Springboot的启动类里面加上


new BootNettyServer().bind(6099);

一个基于Netty搭建的TCP长连接接收16进制数据的服务器程序就搭建成功啦，就是这么简单，如果觉得实用欢迎打赏哈

微信图片_20200605205729.jpg