基于 Kotlin 實現(xiàn)一個簡單的 TCP 自定義協(xié)議
作者：Android架構(gòu)解析
想要成為一名優(yōu)秀的Android開發(fā)，你需要一份完備的 知識體系，在這里，讓我們一起成長為自己所想的那樣~。
一. 開發(fā)背景
想要成為一名優(yōu)秀的Android開發(fā)，你需要一份完備的 知識體系，在這里，讓我們一起成長為自己所想的那樣~。
我們的項目需要開發(fā)一款智能硬件。它由 Web 后臺發(fā)送指令到一款桌面端應用程序，再由桌面程序來控制不同的硬件設備實現(xiàn)業(yè)務上的操作。從 Web 后臺到桌面端是通過一個 WebSocket 長鏈接來進行維護，而桌面程序到各個硬件設備也是一個 TCP 長鏈接來維護的。
本文講述的，其實是從桌面程序到各個硬件之間的通訊。
二. 自定義通訊協(xié)議
首先，需要設計一個通用的 TCP 網(wǎng)絡協(xié)議。
網(wǎng)絡協(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ù)上述所設計的網(wǎng)絡協(xié)議，定義一個抽象類 Packet：
- abstract class Packet {
- var magic:Int? = MAGIC_NUMBER // 魔數(shù)
- var version:Byte = 1 // 版本號，當前協(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 服務端接收并返回給客戶端。
每個 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
- ...... }}
由于使用自定義的協(xié)議，必須要有對報文的 encode、decode，PacketManager 負責這些事情。
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 服務端
啟動 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 分鐘內(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ù)對應的指令去查詢對應的 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 客戶端
模擬一個客戶端的實現(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,"與服務端斷開連接！")
- } catch (e: InterruptedException) {
- e.printStackTrace() }}
其中，PacketCodecHandler 跟服務端使用的解析報文的 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：登錄服務端的 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ā)的桌面端程序其實邏輯并不復雜，只需接收 Web 后臺的指令，然后跟各個設備進行交互。
接收到 Web 端的指令后，通過 Guava 的 EventBus 將指令通過 TCP 發(fā)送給各個設備，發(fā)送時需要轉(zhuǎn)化成對應的 Packet。因此，核心的模塊就是這個 TCP 自定義的協(xié)議。
責任編輯：未麗燕
來源：
今日頭條
