一. 開發背景

想要成為一名優秀的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() { @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.Sharableobject PacketCodecHandler : MessageToMessageCodec() { override fun encode(ctx: ChannelHandlerContext, msg: Packet, list: MutableList) { val byteBuf = ctx.channel().alloc().ioBuffer() PacketManager.encode(byteBuf, msg) list.add(byteBuf) } override fun decode(ctx: ChannelHandlerContext, msg: ByteBuf, list: MutableList) { list.add(PacketManager.decode(msg)); }}

HeartBeatHandler：心跳的 Handler，接收 TCP 客戶端發來的"ping"，然后給客戶端返回"pong"。

@ChannelHandler.Sharableobject HeartBeatHandler : SimpleChannelInboundHandler(){ 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() { private val logger: Logger = LoggerFactory.getLogger(this.javaClass) private val handlerMap: ConcurrentHashMap> = ConcurrentHashMap() init { handlerMap[LOGIN] = LoginHandler ...... handlerMap[ERROR] = ErrorHandler } override fun channelRead0(ctx: ChannelHandlerContext, msg: Packet) { logInfo(logger,"收到客戶端的指令: ${msg.command}") val handler: SimpleChannelInboundHandler? = 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.enclosingClassval logger: Logger = LoggerFactory.getLogger(topLevelClass)fun main() { val worker = NioEventLoopGroup() val bootstrap = Bootstrap() bootstrap.group(worker).channel(NioSocketChannel::class.java) .handler(object : ChannelInitializer() { @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.Sharableclass 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 自定義的協議。

總結

以上是生活随笔為你收集整理的tcp option 结构体_基于 Kotlin 实现一个简单的 TCP 自定义协议的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。