基于 Kotlin 實現(xiàn)一個簡單的 TCP 自定義協(xié)議

作者：Android架構(gòu)解析 2020-09-18 10:12:24

想要成為一名優(yōu)秀的Android開發(fā)，你需要一份完備的知識體系，在這里，讓我們一起成長為自己所想的那樣~。

一. 開發(fā)背景

想要成為一名優(yōu)秀的Android開發(fā)，你需要一份完備的知識體系，在這里，讓我們一起成長為自己所想的那樣~。

我們的項目需要開發(fā)一款智能硬件。它由 Web 后臺發(fā)送指令到一款桌面端應(yīng)用程序，再由桌面程序來控制不同的硬件設(shè)備實現(xiàn)業(yè)務(wù)上的操作。從 Web 后臺到桌面端是通過一個 WebSocket 長鏈接來進行維護，而桌面程序到各個硬件設(shè)備也是一個 TCP 長鏈接來維護的。

[[342699]]

本文講述的，其實是從桌面程序到各個硬件之間的通訊。

二. 自定義通訊協(xié)議

首先，需要設(shè)計一個通用的 TCP 網(wǎng)絡(luò)協(xié)議。

網(wǎng)絡(luò)協(xié)議結(jié)構(gòu)如下

+--------------+---------------+------------+---------------+-----------+----------+ 
     | 魔數(shù)(4)       | version(1)    |序列化方式(1) | command(1)    |數(shù)據(jù)長度(4) |數(shù)據(jù)(n)    | 
     +--------------+---------------+------------+---------------+-----------+----------+

魔數(shù)：4字節(jié)，本項目中使用 20200803(這一天編寫的日子)，為了防止該端口被意外調(diào)用，我們在收到報文后取前4個字節(jié)與魔數(shù)比對，如果不相同則直接拒絕并關(guān)閉連接。
版本號：1字節(jié)，僅表示協(xié)議的版本號，便于協(xié)議升級時使用
序列化方式：1字節(jié)，表示如何將 Java 對象轉(zhuǎn)化為二進制數(shù)據(jù)，以及如何反序列化。
指令：1字節(jié)，表示該消息的意圖(如拍照、拍視頻、心跳、App 升級等)。最多支持 2^8 種指令。
數(shù)據(jù)長度：4字節(jié)，表示該字段后數(shù)據(jù)部分的長度。最多支持 2^32 位。
數(shù)據(jù)：具體數(shù)據(jù)的內(nèi)容。

根據(jù)上述所設(shè)計的網(wǎng)絡(luò)協(xié)議，定義一個抽象類 Packet：

abstract class Packet { 
    var magic:Int? = MAGIC_NUMBER     // 魔數(shù) 
    var version:Byte = 1              // 版本號，當(dāng)前協(xié)議的版本號為 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 客戶端發(fā)起，由 TCP 服務(wù)端接收并返回給客戶端。

每個 Packet 類都包含了該 Packet 所使用的序列化方式。

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

每個 Packet 也包含了其對應(yīng)的 command。下面是 Commands 是指令集，支持256個指令。

/** 
 * 指令集，支持從 -128 到 127 總共 256 個指令 
 */ 
interface Commands { 
    companion object { 
        /** 
         * 心跳包 
         */ 
        const val HEART_BEAT: Byte = 0 
        /** 
         * 登錄（App 需要告訴 Watcher ：cameraPosition 的位置） 
         */ 
        const val LOGIN: Byte = 1 
        ......   }}

由于使用自定義的協(xié)議，必須要有對報文的 encode、decode，PacketManager 負責(zé)這些事情。

encode 時按照協(xié)議的結(jié)構(gòu)進行組裝報文，同理 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) 
        //組裝報文:魔數(shù)（4字節(jié)）+ 版本號（1字節(jié)）+ 序列化方式（1字節(jié)）+ 指令（1字節(jié)）+ 數(shù)據(jù)長度（4字節(jié)）+ 數(shù)據(jù)（N字節(jié)） 
        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) // 魔數(shù)由單獨的 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()  // 數(shù)據(jù)的長度 
        val bytes = ByteArray(length)   // 定義需要讀取的字符數(shù)組 
        buf.readBytes(bytes) 
        return serializer.deserialize(clazz, bytes) 
    } 
}

三. TCP 服務(wù)端

啟動 TCP 服務(wù)的方法

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 分鐘內(nèi)沒有收到心跳，則斷開連接。

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 秒內(nèi)沒有收到心跳，則斷開連接" 
        }    }    companion object { 
        private const val HERT_BEAT_TIME = 300 
    }}

MagicNumValidator：用于 TCP 報文的魔數(shù)校驗。

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) { // 魔數(shù)校驗不通過，則關(guān)閉連接 
            logInfo(logger,"魔數(shù)校驗失敗") 
            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，便于解碼字節(jié)消息成 POJO 或編碼 POJO 消息成字節(jié)。

@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 客戶端發(fā)來的"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 客戶端發(fā)來指令的 Handler，可以根據(jù)對應(yīng)的指令去查詢對應(yīng)的 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,"找到響應(yīng)指令的 Handler: ${it.javaClass.simpleName}") 
            it.channelRead(ctx, msg)        } ?: logInfo(logger,"未找到響應(yīng)指令的 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 客戶端

模擬一個客戶端的實現(xiàn)

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,"與服務(wù)端斷開連接！") 
    } catch (e: InterruptedException) { 
        e.printStackTrace()    }}

其中，PacketCodecHandler 跟服務(wù)端使用的解析報文的 Handler 是一樣的。

ClientIdleHandler：客戶端實現(xiàn)心跳，每隔 30 秒發(fā)送一次心跳。

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,"發(fā)送心跳....") 
        ctx.writeAndFlush(HeartBeatPacket())    }    companion object { 
        private const val HEART_BEAT_TIME = 30 
    }}

ClientLogin：登錄服務(wù)端的 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)    }}

五. 總結(jié)

這次，我開發(fā)的桌面端程序其實邏輯并不復(fù)雜，只需接收 Web 后臺的指令，然后跟各個設(shè)備進行交互。

接收到 Web 端的指令后，通過 Guava 的 EventBus 將指令通過 TCP 發(fā)送給各個設(shè)備，發(fā)送時需要轉(zhuǎn)化成對應(yīng)的 Packet。因此，核心的模塊就是這個 TCP 自定義的協(xié)議。

責(zé)任編輯：未麗燕來源：今日頭條