一. 开发背景

想要成为一名优秀的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 自定义的协议。

现在都说互联网寒冬，其实只要自身技术能力够强，咱们就不怕！我这边专门针对Android开发工程师整理了一套【Android进阶学习视频】、【全套Android面试秘籍】、【Android知识点PDF】。如有需要获取资料文档的朋友，可以点击我的GitHub免费获取！

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

一. 开发背景

二. 自定义通讯协议

三. TCP 服务端

四. TCP 客户端

五. 总结

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

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

一. 开发背景

二. 自定义通讯协议

三. TCP 服务端

四. TCP 客户端

五. 总结

热门文章

最新文章

相关课程

相关电子书