实战Netty！基于私有协议，怎样快速开发网络通信服务？

向右看齐 2023-10-03 21:35 52阅读 0赞

### 前言 ###

今天我们一起来来聊聊怎么使用netty，如果对netty感兴趣的小伙伴，可以看看我的 [Netty专栏][Netty]。

在工作中，我经常使用netty开发一些服务，掌握netty的工作原理，开发一些服务端以及客户端是非常简单，本篇文章，我们就以具体的协议来进行一个简单的服务的开发。

### 正文 ###

#### 私有协议 ####

编写目的

本文档用于描述边缘计算单元（以下简称边缘盒）与上位机配置软件（以下简称上位机）之间进行的数据交互通信协议。

通信方式

边缘盒作为服务端，上位机作为客户端，采用TCP/IP协议的socket连接，端口号默认为6000，数据包采用字节的二进制数据传输。

数据包

数据包由包头和消息内容组成,包头固定**10个字节**，其内容如下：

**标志：** 包的前导字符，固定`CYRC`；

**负载长度：** 负载字节数（不包括包头的长度）；

**协议版本号：** 标识通信协议的版本，初版值为`0x10`；

**包类型：** 标识数据包的操作类型，具体见下表：

<table> 
 <thead> 
  <tr> 
   <th>取值</th> 
   <th>含义</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>查询</td> 
   <td>上位机发送查询消息。</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>设置</td> 
   <td>上位机发送设置消息。</td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td>查询应答</td> 
   <td>边缘盒对查询请求的应答。</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>设置应答</td> 
   <td>边缘盒对设置请求的应答。</td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td>订阅</td> 
   <td>上位机发送给边缘盒订阅数据主动上报请求。</td> 
  </tr> 
  <tr> 
   <td>6</td> 
   <td>主动上报</td> 
   <td>边缘盒主动向上位机发送数据。</td> 
  </tr> 
  <tr> 
   <td>7</td> 
   <td>心跳</td> 
   <td>上位机发送心跳消息</td> 
  </tr> 
  <tr> 
   <td>8</td> 
   <td>心跳应答</td> 
   <td>边缘盒对心跳消息的应答</td> 
  </tr> 
  <tr> 
   <td>其他</td> 
   <td>保留</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**校验位：** 负载数据所有字节之和；

**Reserve：** 预留，值填0；

包体负载（消息内容）表示具体的数据对象，其内容如下：

对于查询、心跳等包类型，包体负载（消息内容）只需要对象标识，对象数据内容省略。

**对象标识：** 标识数据表的操作对象，具体如下：

<table> 
 <thead> 
  <tr> 
   <th>取值</th> 
   <th>含义</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>0</td> 
   <td>心跳</td> 
   <td>上位机连接后间隔时间发送心跳消息给边缘盒。</td> 
  </tr> 
 </tbody> 
</table>

**具体的协议内容就不做展示了，下面就开始服务的编写。**

#### 服务开发 ####

> 这里我们开发一个上位机的配置软件（客户端），我们首先要来分析，怎么对数据包进行编解码，其实工作中，这个也是服务开发的核心所在，也是难点所在。

编写消息类

public class MyProtocol
    {
        /**
         * 消息的开头的信息标志
         */
        private String head = "CYRC";
        /**
         * 消息的长度
         */
        private int contentLength;
        /**
         * 消息的内容
         */
        private byte[] content;
    
        public MyProtocol(int contentLength, byte[] content)
        {
            this.contentLength = contentLength;
            this.content = content;
        }
    
        public String getHead()
        {
            return head;
        }
    
        public void setHead(String head)
        {
            this.head = head;
        }
    
        public int getContentLength()
        {
            return contentLength;
        }
    
        public void setContentLength(int contentLength)
        {
            this.contentLength = contentLength;
        }
    
        public byte[] getContent()
        {
            return content;
        }
    
        public void setContent(byte[] content)
        {
            this.content = content;
        }
    
        public String byteToHex(byte[] bytes, int cnt)
        {
            String strHex;
            StringBuilder sb = new StringBuilder();
            for (int n = 0; n < cnt; n++)
            {
                strHex = Integer.toHexString(bytes[n] & 0xFF);
                sb.append((strHex.length() == 1) ? "0" + strHex : strHex);
                sb.append(" ");
            }
            return sb.toString().trim();
        }
    
        @Override
        public String toString()
        {
            return "MyProtocol [head=" + head + ", contentLength="
                    + contentLength + ", content=" + byteToHex(content, contentLength) + "]";
        }
    }
    复制代码

MyDecoder解码器

@Slf4j
    public class MyDecoder extends ByteToMessageDecoder
    {
        /**
         * <pre>
         * 协议开始的标准head_data，CYRC，占据4个字节.
         * 表示数据的长度contentLength，占据2个字节.
         * </pre>
         */
        public final int BASE_LENGTH = 10;
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception
        {
            if (buffer.readableBytes() >= BASE_LENGTH)
            {
                if (buffer.readableBytes() > 2048)
                {
                    buffer.skipBytes(buffer.readableBytes());
                }
    
                // 记录包头开始的index
                int beginReader;
                //CYRC   43 59 52 43
                while (true)
                {
                    // 获取包头开始的index
                    beginReader = buffer.readerIndex();
                    // 标记包头开始的index
                    buffer.markReaderIndex();
                    // 读到了协议的开始标志，结束while循环
                    int head1 = buffer.readUnsignedShort();
                    int head2 = buffer.readUnsignedShort();
                    if (head1 == 17241 && head2 == 21059)
                    {
                        break;
                    }
    
                    // 未读到包头，略过一个字节
                    // 每次略过，一个字节，去读取，包头信息的开始标记
                    buffer.resetReaderIndex();
                    buffer.readByte();
                    // 当略过，一个字节之后，数据包的长度，又变得不满足
                    // 此时，应该结束。等待后面的数据到达
                    if (buffer.readableBytes() < BASE_LENGTH)
                    {
                        return;
                    }
                }
    
                // 消息的长度
                int length = buffer.readUnsignedShort() + 4;
                // 判断请求数据包数据是否到齐
                if (buffer.readableBytes() < length)
                {
                    // 还原读指针
                    buffer.readerIndex(beginReader);
                    return;
                }
    
                // 读取data数据
                byte[] data = new byte[length];
                buffer.readBytes(data);
    
                MyProtocol protocol = new MyProtocol(data.length, data);
                out.add(protocol);
            }
        }
    }
    复制代码

MyEncoder编码器

public class MyEncoder extends MessageToByteEncoder<MyProtocol>
    {
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, MyProtocol myProtocol, ByteBuf out) throws Exception
        {
            // 1.写入消息的开头的信息标志(CYCR)
            out.writeBytes(myProtocol.getHead().getBytes());
            // 2.写入消息的长度(负载长度)
            out.writeShort(myProtocol.getContentLength() - 4);
            // 3.写入消息的内容(byte[]类型)
            out.writeBytes(myProtocol.getContent());
        }
    }
    复制代码

自定义ChannelInboundHandlerAdapter

@Slf4j
    public class BootNettyClientChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter
    {
        public BootNettyClientChannelInboundHandlerAdapter()
        {
    
        }
    
        /**
         * 从服务端收到新的数据时，这个方法会在收到消息时被调用
         *
         * @param ctx
         * @param msg
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
        {
            MyProtocol protocol = (MyProtocol) msg;
            log.info("接收到服务端的消息:" + protocol);
        }
    
        /**
         * 从服务端收到新的数据、读取完成时调用
         *
         * @param ctx
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws IOException
        {
            ctx.flush();
        }
    
        /**
         * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
         *
         * @param ctx
         * @param cause
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws InterruptedException
        {
            log.error("exceptionCaught：{}", cause.getMessage());
            ctx.close();//抛出异常，断开与客户端的连接
        }
    
        /**
         * 客户端与服务端第一次建立连接时 执行
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception
        {
            super.channelActive(ctx);
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
            String clientIp = insocket.getAddress().getHostAddress();
            log.info("channelActive------TCP客户端新建连接------clientIp:{}", clientIp);
        }
    
        /**
         * 客户端与服务端 断连时 执行
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception
        {
            super.channelInactive(ctx);
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
            String clientIp = insocket.getAddress().getHostAddress();
            ctx.close(); //断开连接时，必须关闭，否则造成资源浪费
            log.info("channelInactive------TCP客户端断开连接----------clientIp:{}", clientIp);
        }
    }
    复制代码

BootNettyClient客户端

@Slf4j
    public class BootNettyClient
    {
        public void connect(String host,int port)
        {
            /**
             * 客户端的NIO线程组
             *
             */
            EventLoopGroup group = new NioEventLoopGroup();
            try
            {
                /**
                 * Bootstrap 是一个启动NIO服务的辅助启动类 客户端的
                 */
                Bootstrap bootstrap = new Bootstrap();
                /**
                 * 设置group
                 */
                bootstrap = bootstrap.group(group);
                /**
                 * 关联客户端通道
                 */
                bootstrap = bootstrap.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true);
                /**
                 * 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
                 */
                bootstrap = bootstrap.handler(new ChannelInitializer<SocketChannel>()
                {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception
                    {
                        ChannelPipeline pipeline = channel.pipeline();
                        // 添加自定义协议的编解码工具
                        pipeline.addLast(new MyDecoder());
                        pipeline.addLast(new MyEncoder());
                        /**
                         * 自定义ChannelInboundHandlerAdapter
                         */
                        pipeline.addLast(new BootNettyClientChannelInboundHandlerAdapter());
                    }
                });
                /**
                 * 连接服务端
                 */
                ChannelFuture f = bootstrap.connect(host, port).sync();
                log.info("TCP客户端连接成功, 地址是： " + host + ":" + port);
                /**
                 * 等待连接端口关闭
                 */
                f.channel().closeFuture().sync();
            }
            catch (Exception e)
            {
                log.error("启动netty client失败：", e);
            }
            finally
            {
                /**
                 * 退出，释放资源
                 */
                group.shutdownGracefully();
            }
        }
    }
    复制代码

NettyClientApplication程序启动类

@SpringBootApplication
    public class NettyClientApplication implements CommandLineRunner {
    
    	public static void main(String[] args) {
    		SpringApplication.run(NettyClientApplication.class, args);
    	}
    
    	@Override
    	public void run(String... args) throws Exception {
    		new BootNettyClient().connect("172.16.1.100", 6000);
    	}
    }
    复制代码

测试

> 利用网络调试助手工具，开启一个服务端，模拟发送数据

![format_png][]

> 发送一个完整的包（43 59 52 43 00 01 10 02 00 00 0f），如下图，客户端完整接收数据。

![format_png 1][]

> 半包测试数据（43 59 52 43 00 01 10 02 00），无日志打印，说明客户端没有接收该不完整数据。

> 粘包数据测试，两个包一起发送（43 59 52 43 00 01 10 02 00 00 0f 43 59 52 43 00 01 10 02 00 00 0f），如下图，客户端同时接收到两条数据。

![format_png 2][]

> 粘包数据测试，一个半包发送（43 59 52 43 00 01 10 02 00 00 0f 43 59 52 43 00 01 10 02），如下图，可以看出，只接收到前面完整包的数据，后面的半包数据被忽略。

![format_png 3][]

业务代码编写

> 业务代码，无非就是将收到的数据进行一些逻辑处理，数据的解析。编写一个接收消息处理类即可。示例如下

通信参数对象:

<table> 
 <thead> 
  <tr> 
   <th>序号</th> 
   <th>名称</th> 
   <th>字节数</th> 
   <th>取值范围</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>对象标识</td> 
   <td>1</td> 
   <td>3</td> 
   <td>对象标识号：3</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>IP地址</td> 
   <td>4</td> 
   <td></td> 
   <td>每个字节表示一段地址值（A.B.C.D,第一字节对应A，依次类推）</td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td>端口</td> 
   <td>2</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>标志</td> 
   <td>1</td> 
   <td>[0-1]</td> 
   <td>0:网络通信，1： 485通信（端口赋波特率，IP赋值0）</td> 
  </tr> 
 </tbody> 
</table>

@Slf4j
    public class ClientService {
    
      /** 接收边缘盒子消息 */
      public void receiveData(MyProtocol myProtocol) {
        try {
          byte[] data = myProtocol.getContent();
          int type = data[1];
          int objId = data[4];
    
          // 心跳应答
          if (type == PackageTypeConstant.HEART_BEAT_REPLY) {
            log.info("--------收到心跳回复----------");
          }
          // 查询应答
          else if (type == PackageTypeConstant.QUERY_RESULT) {
            switch (objId) {
              case ObjectIdConstant.SIGNAL:
                {
                  reSignalParameter(data); // 接收通信参数
                  break;
                }
        
      					//.....
         
              default:
                {
                  break;
                }
            }
          }
        } catch (Exception e) {
          log.error("错误的消息指令..", e);
        }
      }
    
      private void reSignalParameter(byte[] data) {
        EventCenterService.getInstance()
            .submitEvent(
                new IEvent() {
                  @Override
                  public void execute() {
                    try {
                      SignalParameter sp = new SignalParameter();
    
                      int idx = 5;
                      byte temp1;
                      byte temp2;
                      int a = (data[idx++] & 0xFF);
                      int b = (data[idx++] & 0xFF);
                      int c = (data[idx++] & 0xFF);
                      int d = (data[idx++] & 0xFF);
                      String ip = a + "." + b + "." + c + "." + d;
    
                      temp1 = data[idx++];
                      temp2 = data[idx++];
                      // 端口号
                      int port = ((char) (temp1 & 0xFF) << 8) | (char) (temp2 & 0xFF);
                      int sign = data[idx];
    
                      sp.setIp(ip);
                      sp.setPort(port);
                      sp.setSign(sign);
    
                      DataConfig.signalParameterList.add(sp);
                    } catch (Exception e) {
                      log.error("通信参数解析出错：", e);
                    }
                  }
                });
      }
    }
    复制代码

### 后记 ###

工作中，利用netty开发网络通信服务，数据的编解码处理好了，后面的业务代码相对就很容易了。

本篇文章，是我在工作中的一些实战经验，希望对netty感兴趣的小伙伴有点帮助。关于netty的原理这篇文章就不做过多介绍了，前面的文章也讲了很多，后面时间主要讲讲netty实际的运用。

任何一门技术，学会了和实际运用都是两码事，切记眼高手低，程序员从来不适合语言上的巨人，行动上的小人。看到这里，**点赞，关注不迷路**～～

[Netty]: https://juejin.cn/column/6968785053258088479
[format_png]: https://img-blog.csdnimg.cn/img_convert/0faeb95db58dd770a2b25b78a614a758.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/77221c57f1aba052ef8ae8f72b8e5190.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/7f3d71368dcecb44f51fcb00b954c008.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/6d69354552e82ca947817ed99def37fd.webp?x-oss-process=image/format,png