一. 開發(fā)背景

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

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

[[342699]]

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

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

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

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

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

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

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

abstract class Packet { 
    var magic:Int? = MAGIC_NUMBER     // 魔數(shù) 
    var version:Byte = 1              // 版本號(hào)，當(dāng)前協(xié)議的版本號(hào)為 1 
    abstract val serializeMethod:Byte // 序列化方式 
    abstract val command:Byte         // Watcher 跟 App 相互通訊的指令 
} 

有多少個(gè)指令就需要定義多少個(gè) Packet，下面以心跳的 Packet 為例，定義一個(gè) 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ù)端接收并返回給客戶端。

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

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

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

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

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

encode 時(shí)按照協(xié)議的結(jié)構(gòu)進(jìn)行組裝報(bào)文，同理 decode 是其逆向的過(guò)程。

/** 
 * 報(bào)文的管理類，對(duì)報(bào)文進(jìn)行 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) 
        //組裝報(bào)文:魔數(shù)（4字節(jié)）+ 版本號(hào)（1字節(jié)）+ 序列化方式（1字節(jié)）+ 指令（1字節(jié)）+ 數(shù)據(jù)長(zhǎng)度（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ù)由單獨(dú)的 Handler 進(jìn)行校驗(yàn) 
        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ù)的長(zhǎng)度 
        val bytes = ByteArray(length)   // 定義需要讀取的字符數(shù)組 
        buf.readBytes(bytes) 
        return serializer.deserialize(clazz, bytes) 
    } 
} 

三. TCP 服務(wù)端

啟動(dòng) 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)沒(mé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)沒(méi)有收到心跳，則斷開連接" 
        }    }    companion object { 
        private const val HERT_BEAT_TIME = 300 
    }} 

MagicNumValidator：用于 TCP 報(bào)文的魔數(shù)校驗(yàn)。

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ù)校驗(yàn)不通過(guò)，則關(guān)閉連接 
            logInfo(logger,"魔數(shù)校驗(yàn)失敗") 
            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: 解析報(bào)文的 Handler。

PacketCodecHandler 繼承自 ByteToMessageCodec ，它是用來(lái)處理 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ā)來(lái)的"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ā)來(lái)指令的 Handler，可以根據(jù)對(duì)應(yīng)的指令去查詢對(duì)應(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 客戶端

模擬一個(gè)客戶端的實(shí)現(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ù)端使用的解析報(bào)文的 Handler 是一樣的。

ClientIdleHandler：客戶端實(shí)現(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ā)的桌面端程序其實(shí)邏輯并不復(fù)雜，只需接收 Web 后臺(tái)的指令，然后跟各個(gè)設(shè)備進(jìn)行交互。

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