当前位置:首页 > 人工智能

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

一. 开发背景

想要成为一名优秀的基于Android开发,你需要一份完备的实现 知识体系,在这里,个简让我们一起成长为自己所想的单的定义那样~。

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

本文讲述的协议,其实是基于从桌面程序到各个硬件之间的通讯。

二. 自定义通讯协议

首先,实现需要设计一个通用的个简 TCP 网络协议。

网络协议结构如下

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

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

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

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

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

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

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

/**  * 序列化方式的常量列表  */ interface Serialize {      companion object {          const val JSON: Byte = 0     }} 

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

/**  * 指令集,支持从 -128 到 127 总共 256 个指令  */ interface Commands {      companion object {          /**          * 心跳包          */         const val HEART_BEAT: Byte = 0         /**          * 登录(App 需要告诉 Watcher :cameraPosition 的位置)          */         const val LOGIN: Byte = 1         ......   }} 

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

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

/**  * 报文的管理类,对报文进行 encode、decode  */ object PacketManager {      fun encode(packet: Packet):ByteBuf = encode(ByteBufAllocator.DEFAULT, packet)     fun encode(alloc:ByteBufAllocator, packet: Packet) = encode(alloc.ioBuffer(), packet)     fun encode(buf: ByteBuf, packet: Packet): ByteBuf {          val serializer = SerializerFactory.getSerializer(packet.serializeMethod)         val bytes: ByteArray = serializer.serialize(packet)         //组装报文:魔数(4字节)+ 版本号(1字节)+ 序列化方式(1字节)+ 指令(1字节)+ 数据长度(4字节)+ 数据(N字节)         buf.writeInt(MAGIC_NUMBER)         buf.writeByte(packet.version.toInt())         buf.writeByte(packet.serializeMethod.toInt())         buf.writeByte(packet.command.toInt())         buf.writeInt(bytes.size)         buf.writeBytes(bytes)         return buf     }     fun decode(buf:ByteBuf): Packet {          buf.skipBytes(4) // 魔数由单独的 Handler 进行校验         buf.skipBytes(1)         val serializationMethod = buf.readByte()         val serializer = SerializerFactory.getSerializer(serializationMethod)         val command = buf.readByte()         val clazz = PacketFactory.getPacket(command)         val length = buf.readInt()  // 数据的长度         val bytes = ByteArray(length)   // 定义需要读取的字符数组         buf.readBytes(bytes)         return serializer.deserialize(clazz, bytes)     } } 

三. TCP 服务端

启动 TCP 服务的方法

fun execute() {      boss = NioEventLoopGroup()        worker = NioEventLoopGroup()        val bootstrap = ServerBootstrap()     bootstrap.group(boss, worker).channel(NioServerSocketChannel::class.java)             .option(ChannelOption.SO_BACKLOG, 100)             .childOption(ChannelOption.SO_KEEPALIVE, true)             .childOption(ChannelOption.SO_REUSEADDR, true)             .childOption(ChannelOption.TCP_NODELAY, true)             .childHandler(object : ChannelInitializer<NioSocketChannel>() {                  @Throws(Exception::class)                 override fun initChannel(nioSocketChannel: NioSocketChannel) {                      val pipeline = nioSocketChannel.pipeline()                     pipeline.addLast(ServerIdleHandler())                        pipeline.addLast(MagicNumValidator())                        pipeline.addLast(PacketCodecHandler)                        pipeline.addLast(HeartBeatHandler)                        pipeline.addLast(ResponseHandler)                    }                })        val future: ChannelFuture = bootstrap.bind(TCP_PORT)     future.addListener(object : ChannelFutureListener {          @Throws(Exception::class)         override fun operationComplete(channelFuture: ChannelFuture) {              if (channelFuture.isSuccess) {                  logInfo(logger, "TCP Server is starting...")             } else {                  logError(logger,channelFuture.cause(),"TCP Server failed")             }            }        })    } 

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

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

MagicNumValidator:用于 TCP 报文的魔数校验。站群服务器

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

PacketCodecHandler: 解析报文的 Handler。

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

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

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

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

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

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

四. TCP 客户端

模拟一个客户端的实现

val topLevelClass = object : Any() { }.javaClass.enclosingClass val logger: Logger = LoggerFactory.getLogger(topLevelClass)fun main() {      val worker = NioEventLoopGroup()     val bootstrap = Bootstrap()     bootstrap.group(worker).channel(NioSocketChannel::class.java)             .handler(object : ChannelInitializer<SocketChannel>() {                  @Throws(Exception::class)                 override fun initChannel(channel: SocketChannel) {                      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 {          @Throws(Exception::class)         override fun operationComplete(channelFuture: ChannelFuture) {              if (channelFuture.isSuccess()) {                  logInfo(logger,"connect to server success!")             } else {                  logger.info("failed to connect the server! ")                 System.exit(0)             }        }    })    try {          future.channel().closeFuture().sync()        logInfo(logger,"与服务端断开连接!")     } catch (e: InterruptedException) {          e.printStackTrace()    }} 

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

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

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

ClientLogin:登录服务端的 Handler。

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

五. 总结

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

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

分享到:

滇ICP备2023006006号-16