基于 -Kotlin-+-Netty- 实现一个简单的 -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.toI
nt())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 = 7private const val LENGTH_FIELD_LENGTH = 4}}
PacketCodecHandler: 解析报文的 Handler。
PacketCodecHandler 继承自 ByteToMessageCodec ,它是用来处理 byte-to-message 和 message-to-byte,便于解码字节消息成 POJO 或编码 POJO 消息成字节。
@ChannelHandler.Sharableobject 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));}
评论