本文通过一个实例来讲解如何使用 Netty 框架来开发网络协议服务器，项目使用 Gradle 工具来构建和运行，并且支持 Docker 部署。项目代码已在 GitHub 开源，JW Netty Demo。
Netty 简介
Netty 是一个异步、事件驱动的网络应用框架，使用它可以快速开发出可维护良好的、高性能的网络协议服务器。它大幅简化和流程化了网络编程，比如 TCP 和 UDP 套接字服务器开发。难能可贵的是，在保证快速和易用性的同时，使用 Netty 开发的应用并没有丧失可维护性和性能。
上图是来自于官网的 Netty 架构图，可以看到整体结构非常清晰，每一层各司其职。
项目介绍
本项目实现了一个用来接收和存储传感器数据的 TCP 网络协议服务器。这些数据由连接在硬件设备上的传感器采集，然后由硬件设备上报到服务器。一个硬件设备可以连接多个传感器，每次上报硬件设备会把所有传感器的数据一并上报。服务端需要计算所有传感器数据的平均值，并保存起来。这是一个典型的物联网数据采集场景。
协议
考虑到物联网环境硬件设备性能不高、网络带宽较小且稳定性不够，我们需要尽可能降低协议编解码开销、减少报文大小。因此最好使用二进制协议，而不是 JSON 这样的文本格式。
本项目采用的二进制协议如下：
| 起始位(2 bytes) | 报文总长(2 bytes) | 协议版本(1 byte) | 设备号(4 bytes) | 时间戳(4 bytes) | 数据项长度(1 byte) | 数据项值 | 更多数据项... | 校验和(2 bytes) |
- 所有段都是无符号整数，字节顺序为网络字节序
- 起始位固定为
0x55 0xaa，用来防止网络传输过程中数据错乱 - 报文总长包含了起始位和最后的校验和
- 数据项值占用的字节数由其前面一个段的值决定
- 数据项可以有很多个，只要报文总长不超过 65535 就行
代码解读
目录结构
1. 2├── Dockerfile # Docker 镜像构建配置文件 3├── README.md 4├── bin 5├── build 6├── build.gradle # Gradle 构建配置文件 7├── docker-compose.yml # Docker Compose 配置文件 8├── gradle # Gradle 目录 9├── gradlew # Gradle 包装脚本，通过它来执行构建任务 10├── gradlew.bat # Windows 平台下的包装脚本 11├── settings.gradle # Gradle 设置 12└── src 13 ├── main 14 │ ├── java 15 │ │ └── net 16 │ │ └── jaggerwang 17 │ │ └── jwnettydemo 18 │ │ ├── Main.java # 应用入口 19 │ │ ├── config # 配置 20 │ │ │ ├── ApplicationConfig.java 21 │ │ │ └── MongoDBConfig.java 22 │ │ ├── decoder # 报文解码器 23 │ │ │ ├── Message.java 24 │ │ │ └── MessageDecoder.java 25 │ │ ├── handler # 报文处理器 26 │ │ │ └── MessageHandler.java 27 │ │ └── saver # 报文保存器 28 │ │ ├── Metric.java 29 │ │ └── MetricSaver.java 30 │ └── resources 31 │ ├── application.properties # 应用配置 32 │ └── log4j2.xml # 日志配置 33 └── test 34 └── java 35 └── net 36 └── jaggerwang 37 └── jwnettydemo 38 ├── decoder # 报文解码单元测试 39 │ └── MessageDecoderTests.java 40 └── saver # 报文保存单元测试 41 └── MetricSaverTests.java
可以看到，项目结构为典型的 Java 项目结构。
配置文件
通过配置文件，使得同一套代码可以在不同环境中都能运行。为了不为每套环境都维护一套配置文件，本项目将依赖环境的配置抽取了出来，使得运行时可以通过环境变量来覆盖指定配置项。这样各个环境的配置文件就保持了一致，可以共用一份。
配置文件有两个，一个是应用配置，一个是日志配置。日志组件用的是 Log4j 2，配置文件里环境变量的嵌入格式遵循的是 Log4j 风格，包括默认值的指定方式。
应用配置文件内容如下：
1# Path 2path.app=${PATH_APP:-/Users/jagger/projects/jw/jw-netty-demo} 3path.data=${PATH_DATA:-/Users/jagger/data/jw/jw-netty-demo} 4 5# Server 6server.port=${SERVER_PORT:-8080} 7 8# MongoDB 9mongodb.uri=${MONGODB_URI:-mongodb://localhost:27017/} 10mongodb.db=${MONGODB_DB:-jw_netty_demo}
核心代码
整个报文处理过程分两步，第一步解码，由类 MessageDecoder 负责，第二步保存解码出来的 Message，由类 MessageHandler 负责。下面我们重点来看这两个类的实现。
MessageDecoder 类
1package net.jaggerwang.jwnettydemo.decoder; 2 3import io.netty.buffer.ByteBuf; 4import io.netty.channel.ChannelHandlerContext; 5import io.netty.handler.codec.ReplayingDecoder; 6import org.apache.logging.log4j.LogManager; 7import org.apache.logging.log4j.Logger; 8 9import java.util.List; 10 11public class MessageDecoder extends ReplayingDecoder { 12 13 private static final Logger logger = LogManager.getLogger(); 14 15 public static final int HEADER_LENGTH = 13; 16 17 @Override 18 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { 19 Message msg = new Message(); 20 21 msg.setStartFirst(in.readUnsignedByte()); 22 msg.setStartSecond(in.readUnsignedByte()); 23 logger.debug("decode start ok: {} {}", msg.getStartFirst(), msg.getStartSecond()); 24 25 msg.setLength(in.readUnsignedShort()); 26 logger.debug("decode length ok: {}", msg.getLength()); 27 28 msg.setVersion(in.readUnsignedByte()); 29 logger.debug("decode version ok: {}", msg.getVersion()); 30 31 msg.setDeviceNo(in.readUnsignedInt()); 32 logger.debug("decode device no ok: {}", msg.getDeviceNo()); 33 34 msg.setTime(in.readUnsignedInt()); 35 logger.debug("decode time ok: {}", msg.getTime()); 36 37 int pos = HEADER_LENGTH; 38 while (pos < msg.getLength() - 2) { 39 short len = in.readUnsignedByte(); 40 pos += 1; 41 42 long data; 43 switch (len) { 44 case 1: 45 data = (long) in.readUnsignedByte(); 46 pos += 1; 47 break; 48 case 2: 49 data = (long) in.readUnsignedShort(); 50 pos += 2; 51 break; 52 case 4: 53 data = in.readUnsignedInt(); 54 pos += 4; 55 break; 56 default: 57 logger.error("unsupported data length: {}", len); 58 continue; 59 } 60 61 msg.getDatas().add(data); 62 logger.debug("decode one data ok: {}", data); 63 } 64 65 msg.setChecksum(in.readUnsignedShort()); 66 logger.debug("decode checksum ok: {}", msg.getChecksum()); 67 68 out.add(msg); 69 } 70}
MessageDecoder 类继承自 Netty 的 ReplayingDecoder 类，该类解决了网络报文接收不完整的问题。从网络中接收报文时，不能保证一次接收到整个报文，很可能只是其中一部分。如果在应用里来检测报文完整性，在接收到完整报文时才解码，会比较麻烦。通过继承 ReplayingDecoder 类，实现它的 decode 方法，应用就可以假设报文是完整的。实际处理过程中，如果解码时发现报文不完整，decode 会抛出异常，ReplayingDecoder 捕获该异常后将解码位置归零，下次再重头开始。
MessageHandler 类
1package net.jaggerwang.jwnettydemo.handler; 2 3import io.netty.channel.ChannelHandler.Sharable; 4import io.netty.channel.ChannelHandlerContext; 5import io.netty.channel.ChannelInboundHandlerAdapter; 6import net.jaggerwang.jwnettydemo.decoder.Message; 7import net.jaggerwang.jwnettydemo.saver.Metric; 8import net.jaggerwang.jwnettydemo.saver.MetricSaver; 9import org.apache.logging.log4j.LogManager; 10import org.apache.logging.log4j.Logger; 11 12@Sharable 13public class MessageHandler extends ChannelInboundHandlerAdapter { 14 15 private static final Logger logger = LogManager.getLogger(); 16 17 @Override 18 public void channelRead(ChannelHandlerContext ctx, Object msg) { 19 Message message = (Message) msg; 20 if (!message.isValid()) { 21 logger.error("invalid message: {}", message); 22 ctx.close(); 23 return; 24 } 25 logger.debug("received message: {}", message); 26 27 MetricSaver saver = new MetricSaver(); 28 Metric metric = new Metric(message); 29 saver.save(metric); 30 logger.debug("saved metric: {}", metric); 31 } 32 33 @Override 34 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 35 // Close the connection when an exception is raised. 36 cause.printStackTrace(); 37 ctx.close(); 38 } 39}
MessageHandler 在 Netty Channel 处理管道里位于 MessageDecoder 之后，因此它接收到的是已经解码出来的 Message 对象。它不用关心协议细节，只需要简单地把 Message 对象保存到数据库里就可以。
开发
构建工具
本项目使用 Gradle 工具来构建和运行，但是不需要额外去安装。只要使用 ./gradlew <task> 执行任意任务，如果没有检测到 Gradle 工具，就会自动下载并安装到项目下。
本项目的 build.gradle 配置文件如下：
1apply plugin : 'application' 2 3mainClassName = 'net.jaggerwang.jwnettydemo.Main' 4 5dependencies { 6 compile 'io.netty:netty-all:4.1.21.Final' 7 compile 'org.apache.logging.log4j:log4j-api:2.10.0' 8 compile 'org.apache.logging.log4j:log4j-core:2.10.0' 9 compile 'org.mongodb:mongodb-driver:3.6.3' 10 compile 'com.aliyun:hitsdb-client:0.0.5' 11 compile 'org.apache.kafka:kafka-clients:0.10.0.0' 12 compile 'org.apache.kafka:connect-json:0.10.0.0' 13 compile 'com.aliyun.openservices:ons-sasl-client:0.1' 14 compile 'com.fasterxml.jackson.core:jackson-core:2.9.4' 15 compile 'com.fasterxml.jackson.core:jackson-databind:2.9.4' 16 compile 'com.fasterxml.jackson.module:jackson-module-parameter-names:2.9.4' 17 compile 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.9.4' 18 compile 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.9.4' 19 compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0' 20 compile 'commons-codec:commons-codec:1.10' 21 22 compileOnly 'org.projectlombok:lombok:1.16.18' 23 24 testImplementation 'org.junit.jupiter:junit-jupiter-api:5.1.0' 25 testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.1.0' 26} 27 28repositories { 29 mavenCentral() 30} 31 32test { 33 useJUnitPlatform() 34}
运行
执行下面的命令来运行项目：
1$ ./gradlew run 2 Starting a Gradle Daemon (subsequent builds will be faster) 3 4 > Task :run 5 2018-四月-26 21:21:49 INFO net.jaggerwang.jwnettydemo.config.ApplicationConfig - load properties ok 6 SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". 7 SLF4J: Defaulting to no-operation (NOP) logger implementation 8 SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 9 2018-四月-26 21:21:50 INFO net.jaggerwang.jwnettydemo.Main - server started on port 8080 10 <=========----> 75% EXECUTING [17s] 11 > :run
运行起来的 Server 会在 8080 端口监听请求，按照协议组装报文后发往 localhost:8080 就可以上报数据。
测试
使用下面的命令来运行单元测试：
1$ ./gradlew test 2 Starting a Gradle Daemon (subsequent builds will be faster) 3 4 Deprecated Gradle features were used in this build, making it incompatible with Gradle 5.0. 5 See https://docs.gradle.org/4.6/userguide/command_line_interface.html#sec:command_line_warnings 6 7 BUILD SUCCESSFUL in 7s 8 4 actionable tasks: 4 up-to-date
本项目使用 JUnit 5 来编写和运行单元测试。目前有两个单元测试，一个是报文解码，另一个是报文保存。
打包
使用下面的命令来打包和运行：
1$ ./gradlew installDist 2 3 Deprecated Gradle features were used in this build, making it incompatible with Gradle 5.0. 4 See https://docs.gradle.org/4.6/userguide/command_line_interface.html#sec:command_line_warnings 5 6 BUILD SUCCESSFUL in 2s 7 5 actionable tasks: 3 executed, 2 up-to-date 8 9$ ./build/install/jw-netty-demo/bin/jw-netty-demo 10 2018-四月-26 22:23:22 INFO net.jaggerwang.jwnettydemo.config.ApplicationConfig - load properties ok 11 SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". 12 SLF4J: Defaulting to no-operation (NOP) logger implementation 13 SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 14 2018-四月-26 22:23:22 INFO net.jaggerwang.jwnettydemo.Main - server started on port 8080
打包好的应用会安装到目录 ./build/install/jw-netty-demo 下, 可以使用其下的 bin/jw-netty-demo 脚本来运行应用。
Docker 部署
构建镜像
执行下面的命令来构建镜像：
$ docker build -t jw-netty-demo .
构建镜像的配置文件如下：
1FROM java:8 2 3ENV APP_PATH=/app 4ENV DATA_PATH=/data 5 6WORKDIR $APP_PATH 7 8COPY . . 9RUN ./gradlew installDist 10 11VOLUME $DATA_PATH 12 13EXPOSE 8080 14 15CMD ./build/install/jw-netty-demo/bin/jw-netty-demo
在容器里运行服务
执行下面的命令来运行项目所有服务：
$ docker-compose up
Docker Compose 配置文件如下：
1version: "2" 2services: 3 app: 4 image: jw-netty-demo:latest 5 environment: 6 PATH_APP: /app 7 PATH_DATA: /data 8 MONGODB_URI: mongodb:27017 9 ports: 10 - 19900:8080 11 volumes: 12 - ~/data/jw-netty-demo/app:/data 13 mongodb: 14 image: mongo:3 15 volumes: 16 - ~/data/jw-netty-demo/mongodb:/data/db
Docker Compose 会同时启动 app 服务，及其依赖的 mongodb 服务。
参考资料
本文转自 https://blog.jaggerwang.net/netty-high-performance-protocol-server-develop/，如有侵权，请联系删除。