Shulou(Shulou.com)06/02 Report--

This article mainly explains "how to achieve a simple TCP custom protocol based on Kotlin", the content of the article is simple and clear, easy to learn and understand, the following please follow the editor's ideas slowly in depth, together to study and learn "how to achieve a simple TCP custom protocol based on Kotlin"!

Custom communication protocol

First of all, we need to design a general TCP network protocol.

The network protocol structure is as follows

+-+ | number of demons (4) | version (1) | Serialization method (1) | command (1) | number According to length (4) | data (n) | +-+-+

Magic number: 4 bytes. 20200803 is used in this project (the day on which it was written). In order to prevent the port from being accidentally called, we compare the first 4 bytes with the magic number after receiving the message, and directly reject and close the connection if it is different.

Version number: 1 byte, which only represents the version number of the protocol, so it is easy to use when upgrading the protocol.

Serialization method: 1 byte, indicating how to convert Java objects to binary data and how to deserialize them.

Instruction: 1 byte, indicating the intention of the message (such as photo, video, heartbeat, App upgrade, etc.). A maximum of 2 ^ 8 instructions are supported.

Data length: 4 bytes, indicating the length of the data portion after the field. Up to 2 ^ 32 bits are supported.

Data: the content of specific data.

According to the network protocol designed above, an abstract class Packet is defined:

Abstract class Packet {var magic:Int? = MAGIC_NUMBER / / Magic number var version:Byte = 1 / / version number, the version number of the current protocol is 1 abstract val serializeMethod:Byte / / Serialization mode abstract val command:Byte / / Watcher instructions for communicating with App}

You need to define as many Packet as there are instructions. Take the Packet of heartbeat as an example, define a HeartBeatPacket:

Data class HeartBeatPacket (var msg:String = "ping", overrideval serializeMethod: Byte = Serialize.JSON, overrideval command: Byte = Commands.HEART_BEAT): Packet () {}

HeartBeatPacket is initiated by the TCP client, received by the TCP server and returned to the client.

Each Packet class contains the serialization method used by the Packet.

/ * * constant list of serialization methods * / interface Serialize {companion object {const val JSON: Byte = 0}}

Each Packet also contains its corresponding command. The following is the Commands instruction set, which supports 256 instructions.

/ * instruction set supports a total of 256 instructions from-128to127256 instructions * / interface Commands {companion object {/ * heartbeat package * / const val HEART_BEAT: Byte = 0 / * login (App needs to tell Watcher: cameraPosition location) * / const val LOGIN: Byte = 1. }}

Due to the use of custom protocols, there must be encode and decode,PacketManager responsible for these things.

Encode assembles messages according to the structure of the protocol, and decode is the reverse process.

/ * the management class of the message 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) / / Assembly message: magic number (4 bytes) + version number (1 byte) + serialization mode (1 byte) + instruction (1 byte) + data length (4 bytes) + data (N bytes) 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) / / Magic number is verified by separate 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 () / / length of data val bytes = ByteArray (length) / / defines the character array buf.readBytes (bytes) return serializer.deserialize (clazz) to be read Bytes)}}

TCP server

How to start the TCP service

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 () {@ 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: if no heartbeat is received within 5 minutes, the connection will be disconnected.

Class ServerIdleHandler: IdleStateHandler (0,0, HERT_BEAT_TIME) {privateval 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 did not receive a heartbeat within seconds. Then disconnect "}} companion object {private const val HERT_BEAT_TIME = 300}}

MagicNumValidator: magic number check for TCP messages.

Class MagicNumValidator: LengthFieldBasedFrameDecoder (Int.MAX_VALUE, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH) {privateval logger: Logger = LoggerFactory.getLogger (this.javaClass) @ Throws (Exception::class) override fun decode (ctx: ChannelHandlerContext, `in`: ByteBuf): Any? {if (`in`.getInt (`in`.readerIndex ())! = MAGIC_NUMBER) {/ / Magic number check failed Then close the connection logInfo (logger, "magic number check failed") 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: the Handler that parses the message.

PacketCodecHandler inherits from ByteToMessageCodec and is used to handle byte-to-message and message-to-byte, making it easy to decode byte messages into POJO or encode POJO messages into bytes.

@ ChannelHandler.Sharable object PacketCodecHandler: MessageToMessageCodec () {override fun encode (ctx: ChannelHandlerContext, msg: Packet, list: MutableList) {val byteBuf = ctx.channel (). Alloc (). IoBuffer () PacketManager.encode (byteBuf, msg) list.add (byteBuf)} override fun decode (ctx: ChannelHandlerContext, msg: ByteBuf, list: MutableList) {list.add (PacketManager.decode (msg));}}

HeartBeatHandler: the Handler of the heartbeat, receives "ping" from the TCP client, and then returns "pong" to the client.

ChannelHandler.Sharable object HeartBeatHandler: SimpleChannelInboundHandler () {privateval logger: Logger = LoggerFactory.getLogger (this.javaClass) override fun channelRead0 (ctx: ChannelHandlerContext, msg: HeartBeatPacket) {logInfo (logger, "receive heartbeat packet: ${GsonUtils.toJson (msg)}") msg.msg = "pong" / / return pong to client ctx.writeAndFlush (msg)}}

ResponseHandler: a general Handler that receives instructions from a TCP client. You can query the corresponding Handler and process its commands according to the corresponding instructions.

Object ResponseHandler: SimpleChannelInboundHandler () {privateval logger: Logger = LoggerFactory.getLogger (this.javaClass) privateval handlerMap: ConcurrentHashMap = ConcurrentHashMap () init {handlerMap [LOGIN] = LoginHandler. HandlerMap [ERROR] = ErrorHandler} override fun channelRead0 (ctx: ChannelHandlerContext, msg: Packet) {logInfo (logger, "receive client instruction: ${msg.command}") val handler: SimpleChannelInboundHandler? = handlerMap [msg.command] handler?.let {logInfo (logger, "find Handler: ${it.javaClass.simpleName}") it.channelRead (ctx, msg)}?: logInfo (logger "Handler not found responding to instruction") @ 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, "client drop: $clientIP: $clientPort") super.channelInactive (ctx)}}

TCP client

Simulate the implementation of a client

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 () {@ 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, "disconnect from the server!") } catch (e: InterruptedException) {e.printStackTrace ()}}

Among them, PacketCodecHandler is the same as the Handler used by the server to parse messages.

ClientIdleHandler: the client implements a heartbeat and sends a heartbeat every 30 seconds.

Class ClientIdleHandler: IdleStateHandler (0,0, HEART_BEAT_TIME) {privateval logger = LoggerFactory.getLogger (ClientIdleHandler::class.java) @ Throws (Exception::class) override fun channelIdle (ctx: ChannelHandlerContext, evt: IdleStateEvent?) {logInfo (logger, "send heartbeat....") Ctx.writeAndFlush (HeartBeatPacket ())} companion object {private const val HEART_BEAT_TIME = 30}}

ClientLogin: log in to the server's Handler.

ChannelHandler.Sharable class ClientLogin: ChannelInboundHandlerAdapter () {privateval 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)}} Thank you for reading The above is the content of "how to achieve a simple TCP custom protocol based on Kotlin". After the study of this article, I believe you have a deeper understanding of how to achieve a simple TCP custom protocol based on Kotlin. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.