基于 Kotlin 实现一个简单的 TCP 自定义协议

一. 开发背景

想要成为一名优秀的Android开发，你需要一份完备的 知识体系，在这里，让我们一起成长为自己所想的那样~。

我们的项目需要开发一款智能硬件。它由 web 后台发送指令到一款桌面端应用程序，再由桌面程序来控制不同的硬件设备实现业务上的操作。从 Web 后台到桌面端是通过一个 WebSocket 长链接来进行维护，而桌面程序到各个硬件设备也是一个 TCP 长链接来维护的。

本文讲述的，其实是从桌面程序到各个硬件之间的通讯。

二. 自定义通讯协议

首先，需要设计一个通用的 TCP 网络协议。

网络协议结构如下

1. +--------------+---------------+------------+---------------+-----------+----------+ 
2.      | 魔数(4)       | version(1)    |序列化方式(1) | command(1)    |数据长度(4) |数据(n)    | 
3.      +--------------+---------------+------------+---------------+-----------+----------+ 

• 魔数：4字节，本项目中使用 20200803(这一天编写的日子)，为了防止该端口被意外调用，我们在收到报文后取前4个字节与魔数比对，如果不相同则直接拒绝并关闭连接。
• 版本号：1字节，仅表示协议的版本号，便于协议升级时使用
• 序列化方式：1字节，表示如何将 java 对象转化为二进制数据，以及如何反序列化。
• 指令：1字节，表示该消息的意图(如拍照、拍视频、心跳、app 升级等)。最多支持 2^8 种指令。
• 数据长度：4字节，表示该字段后数据部分的长度。最多支持 2^32 位。
• 数据：具体数据的内容。

根据上述所设计的网络协议，定义一个抽象类 Packet：

1. abstract class Packet { 
2.     var magic:Int? = MAGIC_NUMBER     // 魔数 
3.     var version:Byte = 1              // 版本号，当前协议的版本号为 1 
4.     abstract val serializeMethod:Byte // 序列化方式 
5.     abstract val command:Byte         // Watcher 跟 App 相互通讯的指令 
6. } 

有多少个指令就需要定义多少个 Packet，下面以心跳的 Packet 为例，定义一个 HeartBeatPacket：

1. data class HeartBeatPacket(var msg:String = "ping", 
2.                            override val serializeMethod: Byte = Serialize.JSON, 
3.                            override val command: Byte = Commands.HEART_BEAT) : Packet() { 
4. } 

HeartBeatPacket 是由 TCP 客户端发起，由 TCP 服务端接收并返回给客户端。

每个 Packet 类都包含了该 Packet 所使用的序列化方式。

1.  
2. interface Serialize { 
3.     companion object { 
4.         const val JSON: Byte = 0 
5.     }} 

每个 Packet 也包含了其对应的 command。下面是 Commands 是指令集，支持256个指令。

1.  
2. interface Commands { 
3.     companion object { 
4.          
5.         const val HEART_BEAT: Byte = 0 
6.          
7.         const val LOGIN: Byte = 1 
8.         ......   }} 

由于使用自定义的协议，必须要有对报文的 encode、decode，PacketManager 负责这些事情。

encode 时按照协议的结构进行组装报文，同理 decode 是其逆向的过程。

1.  
2. object PacketManager { 
3.     fun encode(packet: Packet):ByteBuf = encode(ByteBufAllocator.DEFAULT, packet) 
4.     fun encode(alloc:ByteBufAllocator, packet: Packet) = encode(alloc.ioBuffer(), packet) 
5.     fun encode(buf: ByteBuf, packet: Packet): ByteBuf { 
6.         val serializer = SerializerFactory.getSerializer(packet.serializeMethod) 
7.         val bytes: ByteArray = serializer.serialize(packet) 
8.         //组装报文:魔数（4字节）+ 版本号（1字节）+ 序列化方式（1字节）+ 指令（1字节）+ 数据长度（4字节）+ 数据（N字节） 
9.         buf.writeInt(MAGIC_NUMBER) 
10.         buf.writeByte(packet.version.toInt()) 
11.         buf.writeByte(packet.serializeMethod.toInt()) 
12.         buf.writeByte(packet.command.toInt()) 
13.         buf.writeInt(bytes.size) 
14.         buf.writeBytes(bytes) 
15.         return buf 
16.     } 
17.     fun decode(buf:ByteBuf): Packet { 
18.         buf.skipBytes(4) // 魔数由单独的 Handler 进行校验 
19.         buf.skipBytes(1) 
20.         val serializationMethod = buf.readByte() 
21.         val serializer = SerializerFactory.getSerializer(serializationMethod) 
22.         val command = buf.readByte() 
23.         val clazz = PacketFactory.getPacket(command) 
24.         val length = buf.readInt()  // 数据的长度 
25.         val bytes = ByteArray(length)   // 定义需要读取的字符数组 
26.         buf.readBytes(bytes) 
27.         return serializer.deserialize(clazz, bytes) 
28.     } 
29. } 

三. TCP 服务端

启动 TCP 服务的方法

1. fun execute() { 
2.     boss = NioEventLoopGroup()        worker = NioEventLoopGroup()        val bootstrap = ServerBootstrap() 
3.     bootstrap.group(boss, worker).channel(NioServerSocketChannel::class.java) 
4.             .option(ChannelOption.SO_BACKLOG, 100) 
5.             .childOption(ChannelOption.SO_KEEPALIVE, true) 
6.             .childOption(ChannelOption.SO_REUSEADDR, true) 
7.             .childOption(ChannelOption.TCP_NODELAY, true) 
8.             .childHandler(object : ChannelInitializer<NioSocketChannel>() { 
9.                 @Throws(Exception::class) 
10.                 override fun initChannel(nioSocketChannel: NioSocketChannel) { 
11.                     val pipeline = nioSocketChannel.pipeline() 
12.                     pipeline.addLast(ServerIdleHandler())                        pipeline.addLast(MagicNumValidator())                        pipeline.addLast(PacketCodecHandler)                        pipeline.addLast(HeartBeatHandler)                        pipeline.addLast(ResponseHandler)                    }                })        val future: ChannelFuture = bootstrap.bind(TCP_PORT) 
13.     future.addListener(object : ChannelFutureListener { 
14.         @Throws(Exception::class) 
15.         override fun operationComplete(channelFuture: ChannelFuture) { 
16.             if (channelFuture.isSuccess) { 
17.                 logInfo(logger, "TCP Server is starting...") 
18.             } else { 
19.                 logError(logger,channelFuture.cause(),"TCP Server failed") 
20.             }            }        })    } 

其中，ServerIdleHandler： 表示 5 分钟内没有收到心跳，则断开连接。

1. class ServerIdleHandler : IdleStateHandler(0, 0, HERT_BEAT_TIME) { 
2.     private val logger: Logger = LoggerFactory.getLogger(ServerIdleHandler::class.java) 
3.     @Throws(Exception::class) 
4.     override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent) { 
5.         logInfo(logger) {            ctx.channel().close()            "$HERT_BEAT_TIME 秒内没有收到心跳，则断开连接" 
6.         }    }    companion object { 
7.         private const val HERT_BEAT_TIME = 300 
8.     }} 

MagicNumValidator：用于 TCP 报文的魔数校验。

1. class MagicNumValidator : LengthFieldBasedFrameDecoder(Int.MAX_VALUE, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH) { 
2.     private val logger: Logger = LoggerFactory.getLogger(this.javaClass) 
3.     @Throws(Exception::class) 
4.     override fun decode(ctx: ChannelHandlerContext, `in`: ByteBuf): Any? { 
5.         if (`in`.getInt(`in`.readerIndex()) !== MAGIC_NUMBER) { // 魔数校验不通过，则关闭连接 
6.             logInfo(logger,"魔数校验失败") 
7.             ctx.channel().close() 
8.             return null 
9.         } 
10.         return super.decode(ctx, `in`) 
11.     } 
12.     companion object { 
13.         private const val LENGTH_FIELD_OFFSET = 7 
14.         private const val LENGTH_FIELD_LENGTH = 4 
15.     } 
16. } 

PacketCodecHandler: 解析报文的 Handler。

PacketCodecHandler 继承自 ByteToMessageCodec ，它是用来处理 byte-to-message 和message-to-byte，便于解码字节消息成 POJO 或编码 POJO 消息成字节。

1. @ChannelHandler.Sharable 
2. object PacketCodecHandler : MessageToMessageCodec<ByteBuf, Packet>() {    override fun encode(ctx: ChannelHandlerContext, msg: Packet, list: MutableList<Any>) { 
3.         val byteBuf = ctx.channel().alloc().ioBuffer() 
4.         PacketManager.encode(byteBuf, msg)        list.add(byteBuf)    }    override fun decode(ctx: ChannelHandlerContext, msg: ByteBuf, list: MutableList<Any>) { 
5.         list.add(PacketManager.decode(msg));    }} 

HeartBeatHandler：心跳的 Handler，接收 TCP 客户端发来的"ping"，然后给客户端返回"pong"。

1. @ChannelHandler.Sharable 
2. object HeartBeatHandler : SimpleChannelInboundHandler<HeartBeatPacket>(){    private val logger: Logger = LoggerFactory.getLogger(this.javaClass) 
3.     override fun channelRead0(ctx: ChannelHandlerContext, msg: HeartBeatPacket) { 
4.         logInfo(logger,"收到心跳包：${GsonUtils.toJson(msg)}") 
5.         msg.msg = "pong" // 返回 pong 给到客户端 
6.         ctx.writeAndFlush(msg) 
7.     } 
8. } 

ResponseHandler：通用的处理接收 TCP 客户端发来指令的 Handler，可以根据对应的指令去查询对应的 Handler 并处理其命令。

1. object ResponseHandler: SimpleChannelInboundHandler<Packet>() { 
2.     private val logger: Logger = LoggerFactory.getLogger(this.javaClass) 
3.     private val handlerMap: ConcurrentHashMap<Byte, SimpleChannelInboundHandler<out Packet>> = ConcurrentHashMap() 
4.     init { 
5.         handlerMap[LOGIN] = LoginHandler        ......        handlerMap[ERROR] = ErrorHandler    }    override fun channelRead0(ctx: ChannelHandlerContext, msg: Packet) { 
6.         logInfo(logger,"收到客户端的指令: ${msg.command}") 
7.         val handler: SimpleChannelInboundHandler<out Packet>? = handlerMap[msg.command] 
8.         handler?.let {            logInfo(logger,"找到响应指令的 Handler: ${it.javaClass.simpleName}") 
9.             it.channelRead(ctx, msg)        } ?: logInfo(logger,"未找到响应指令的 Handler") 
10.     }    @Throws(Exception::class) 
11.     override fun channelInactive(ctx: ChannelHandlerContext) { 
12.         val insocket = ctx.channel().remoteAddress() as InetSocketAddress 
13.         val clientIP = insocket.address.hostAddress 
14.         val clientPort = insocket.port 
15.         logError(logger,"客户端掉线: $clientIP : $clientPort") 
16.         super.channelInactive(ctx) 
17.     }} 

四. TCP 客户端

模拟一个客户端的实现

1. val topLevelClass = object : Any() {}.javaClass.enclosingClass 
2. val logger: Logger = LoggerFactory.getLogger(topLevelClass)fun main() { 
3.     val worker = NioEventLoopGroup() 
4.     val bootstrap = Bootstrap() 
5.     bootstrap.group(worker).channel(NioSocketChannel::class.java) 
6.             .handler(object : ChannelInitializer<SocketChannel>() { 
7.                 @Throws(Exception::class) 
8.                 override fun initChannel(channel: SocketChannel) { 
9.                     channel.pipeline().addLast(PacketCodecHandler)                    channel.pipeline().addLast(ClientIdleHandler())                    channel.pipeline().addLast(ClientLogin())                }            })    val future: ChannelFuture = bootstrap.connect("127.0.0.1", TCP_PORT).addListener(object : ChannelFutureListener { 
10.         @Throws(Exception::class) 
11.         override fun operationComplete(channelFuture: ChannelFuture) { 
12.             if (channelFuture.isSuccess()) { 
13.                 logInfo(logger,"connect to server success!") 
14.             } else { 
15.                 logger.info("failed to connect the server! ") 
16.                 System.exit(0) 
17.             }        }    })    try { 
18.         future.channel().closeFuture().sync()        logInfo(logger,"与服务端断开连接！") 
19.     } catch (e: InterruptedException) { 
20.         e.printStackTrace()    }} 

其中，PacketCodecHandler 跟服务端使用的解析报文的 Handler 是一样的。

ClientIdleHandler：客户端实现心跳，每隔 30 秒发送一次心跳。

1. class ClientIdleHandler : IdleStateHandler(0, 0, HEART_BEAT_TIME) { 
2.     private val logger = LoggerFactory.getLogger(ClientIdleHandler::class.java) 
3.     @Throws(Exception::class) 
4.     override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent?) { 
5.         logInfo(logger,"发送心跳....") 
6.         ctx.writeAndFlush(HeartBeatPacket())    }    companion object { 
7.         private const val HEART_BEAT_TIME = 30 
8.     }} 

ClientLogin：登录服务端的 Handler。

1. @ChannelHandler.Sharable 
2. class ClientLogin: ChannelInboundHandlerAdapter() {    private val logger: Logger = LoggerFactory.getLogger(this.javaClass) 
3.     @Throws(Exception::class) 
4.     override fun channelActive(ctx: ChannelHandlerContext) { 
5.         val packet: LoginPacket = LoginPacket() 
6.         logInfo(logger,"packet = ${GsonUtils.toJson(packet)}") 
7.         val byteBuf = PacketManager.encode(packet) 
8.         ctx.channel().writeAndFlush(byteBuf)    }} 

五. 总结

这次，我开发的桌面端程序其实逻辑并不复杂，只需接收 Web 后台的指令，然后跟各个设备进行交互。

接收到 Web 端的指令后，通过 Guava 的 EventBus 将指令通过 TCP 发送给各个设备，发送时需要转化成对应的 Packet。因此，核心的模块就是这个 TCP 自定义的协议。