編解碼粘包拆包心跳機制斷線自動重連

Netty涉及到編解碼的組件有Channel、ChannelHandler、ChannelPipe等，先大概了解下這幾個組件的作Netty涉及到編解碼的組件有Channel、ChannelHandler、ChannelPipe等，先大概了解下這幾個組件的作或ChaeondHaerAdapter），你就可以接收入站事件和數(shù)據(jù)，這些數(shù)據(jù)隨后會被你的應(yīng)用程序的業(yè)務(wù)邏輯處理。當(dāng)你要給連接的客戶端發(fā)送響應(yīng)時，也可以從ChaeoHaer沖刷數(shù)據(jù)。你的業(yè)務(wù)邏輯通常寫在一個或者多個ChaelIoaerChaetbouaer原理一樣，只不過它是用來處理出站數(shù)據(jù)的。Chaeipene提供了Chaelaer鏈的容器。出站的，即客戶端發(fā)送給服務(wù)端的數(shù)據(jù)會通過pieChannlbnHanler(Channlounnlr調(diào)用是從tai到head方向逐個調(diào)用每個handle的邏輯)，并被這些Haer入站只調(diào)用ppee里的ChannelInbounandler(ChannelInbounandler從hea到tai方向逐個調(diào)用每個hanler的邏輯)。當(dāng)你通過Netty發(fā)送或者接受一個消息的時候，就將會發(fā)生一次數(shù)據(jù)轉(zhuǎn)換。入站消息會被解碼：從字節(jié)轉(zhuǎn)換為另一種格式（比如java象）；如果是出站消息，它會被編碼成字節(jié)提供了一系列實用的編碼解碼器，他們都實現(xiàn)了ChannelInboundHadnler或者ChannelOutboundHandler接口。在這些channelRead方法已經(jīng)被重寫了。以入站為例，對于每個從入站Channel讀取的消息，這個方法會被調(diào)用。隨后，它將調(diào)用由已知解碼所提供的decode()方法進行解碼，并將已經(jīng)解碼的字節(jié)轉(zhuǎn)發(fā)給ChannelPipeline中的下一個ChannelInboundHandler提供了很多編解碼器，比如編解碼字符串的StringEncoder和StringDecoder，編解碼對象的ObjectEncoder和等如果要實現(xiàn)高效的編解碼可以用protobuf，但是protobuf需要維護大量的proto文件比較麻煩，現(xiàn)在一般可以使用protostuff。protostuff是一個基于protobuf實現(xiàn)的序列化方法，它較于protobufo文件來實現(xiàn)序列化。使用它也非常簡單，代碼如下：引入依protostuff1importtostuff.LinkedBuffer;importtostuff.Schema;import23451><version>1.0.10<version>1.0.10<version>1.0.10importimport89*protostuff序列化工具類，基于protobufpublicclassProtostuffUtilprivatestatic,cachedSchema=new,privatestatic<T>Schema<T>getSchema(Class<T>{importimport89*protostuff序列化工具類，基于protobufpublicclassProtostuffUtilprivatestatic,cachedSchema=new,privatestatic<T>Schema<T>getSchema(Class<T>{Schema<T>schema=(Schema<T>)ifnull)schema=ifnull)}}return}**@parampublicstatic<T>byte[]serializer(T{Class<T>clazz=(Class<T>)LinkedBufferbuffer=LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);try{Schema<T>schema=returnProtostuffIOUtil.toByteArray(obj,schema,}catch(Exceptione)thrownewIllegalStateException(e.getMessage(),}{}}*@param@parampublicstatic<T>Tdeserializer(byte[]data,Class<T>clazz)tryTobj=Schema<T>schema=returnobj;}catch(Exceptione)thrownewIllegalStateException(e.getMessage(),}}publicstaticvoidmain(String[]args)參見項目示例ty.codec包下代TCP是一個流協(xié)議，就是沒有界限參見項目示例ty.codec包下代TCP是一個流協(xié)議，就是沒有界限的一長串二進制數(shù)據(jù)。TCP作為傳輸層協(xié)議并不不了解上層業(yè)務(wù)數(shù)據(jù)的具體含義，它會根據(jù)TCP緩沖的實際情況進行數(shù)據(jù)包的劃分，所以在業(yè)務(wù)上認(rèn)為是一個完整的包，可能會被CP拆分成多個包進行發(fā)送，也有可能把多個小的包封裝成一個大的數(shù)據(jù)包發(fā)送，這就是所謂的C面向流的通信是無消息保護邊界的。如下圖所示，client發(fā)了兩個數(shù)據(jù)包D1和D2，但是server端可能會收到如下幾種情況的數(shù)據(jù)消息定長度，傳輸?shù)臄?shù)據(jù)大小固定長度，例如每段的長度固定為100字節(jié)，如果不夠空位補空在數(shù)據(jù)包尾部添加特殊分隔符，比如下劃線，中劃線等，這種方法簡單易行，但選擇分隔符的時候一定要注意每條數(shù)據(jù)的內(nèi)部一定不出現(xiàn)分隔符3）送長度：發(fā)送每條數(shù)據(jù)的時候，將數(shù)據(jù)的長度一并發(fā)送，比如可以選擇每條數(shù)據(jù)的前4LineBasedFrameDecoder（回車換行分包）自定義長度分包解碼器，參見項目示例ty.split包下所謂心跳TCP連接中戶端和服務(wù)器之間定期發(fā)送的一種特殊的數(shù)據(jù)包知對方自己還在線確保TCP接的有效性.在Netty中,實現(xiàn)心跳機制的關(guān)鍵是IdleStateHandler,看下它的構(gòu)造器：這里解釋下三個參數(shù)的含義readerIdleTimeSeconds超在指定的時間間隔內(nèi)沒有從Channel數(shù)據(jù)時READER_IDLEIdleStateEvent事件writerIdleTimeSeconds超時當(dāng)在指定的時間間隔內(nèi)沒有數(shù)據(jù)寫入到Channel觸發(fā)一個WRITER_IDLEIdleStateEvent事件allIdleTimeSeconds/寫超時當(dāng)在指定的時間間隔內(nèi)沒有讀或?qū)懖僮鲿r發(fā)一個ALL_IDLEIdleStateEvent件.要實現(xiàn)Netty服務(wù)端心跳檢測機制需要在服務(wù)器端的ChannelInitializer中加入如下的初步地看下IdleStateHandler源碼，先看下IdleStateHandler中的channelRead方法1pipeline.addLast(newIdleStateHandler(3,0,0,1IdleStateHandler(booleanobserveOutput,longreaderIdleTime,longwriterIdleTime,longallIdleTime,TimeUnitpublicIdleStateHandler(intreaderIdleTimeSeconds,intwriterIdleTimeSeconds,intallIdleTimeSeconds)367byte[]userBytesUseruser=ProtostuffUtil.deserializer(userBytes,7071紅框代碼其實表示該方法只是進行了透傳，不做任何業(yè)務(wù)邏輯處理，讓haelPipe紅框代碼其實表示該方法只是進行了透傳，不做任何業(yè)務(wù)邏輯處理，讓haelPipe中的下一個haer處理hanead我們再看看haete這里有個initialize的方法，這是IdleStateHandler的精髓，接著探這邊會觸發(fā)一個Task，ReaderIdleTimeoutTask，這個task里的run方法源碼是這樣第一個紅框代碼是用當(dāng)前時間減去最后一次haead方法調(diào)用的時間，假如這個結(jié)果是6s，說明最后一次調(diào)用ha第一個紅框代碼是用當(dāng)前時間減去最后一次haead方法調(diào)用的時間，假如這個結(jié)果是6s，說明最后一次調(diào)用haead已經(jīng)是的userEventTriggered方法如果沒有超時則不觸發(fā)userEventTriggered方法Netty心跳檢測代碼示例publicclassHeartBeatServer3publicstaticvoidmain(String[]args)throwsExceptionEventLoopGroupboss=newEventLoopGroupworker=newtryServerBootstrapbootstrap=newprotectedvoidinitChannel(SocketChannelch)throwsExceptionChannelPipelinepipeline=pipeline.addLast("decoder",newpipeline.addLast("encoder",newpipeline.addLast(newIdleStateHandler(3,0,0,pipeline.addLast(new2223System.out.println("nettyserverstart。。ChannelFuturefuture=}catch(Exceptione)}finallypublicclassHeartBeatClientpublicclassHeartBeatClientpublicstaticvoidmain(String[]args)throwsExceptionEventLoopGroupeventLoopGroup=newtryBootstrapbootstrap=newpublicclassHeartBeatServerHandlerextendsSimpleChannelInboundHandler<String>34intreadIdleTimes=5protectedvoidchannelRead0(ChannelHandlerContextctx,Strings)throwsExceptionSystem.out.println("======>[server]messagereceived:"+if("HeartbeatPacket".equals(s))}elseSystem.out.println(1314publicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsExceptionIdleStateEventevent=(IdleStateEvent)StringeventType=switch(event.state())caseeventType"讀空閑readIdleTimes讀空閑的計數(shù)加caseeventType"寫空閑caseeventType"讀寫空閑34System.out.println(ctx.channel().remoteAddress"超時事件：if(readIdleTimes>3)"[4041publicvoidchannelActive(ChannelHandlerContextctx)throwsExceptionSystem.err.println("==="+ctx.channel().remoteAddress()+"isactive464731323334參見代碼12鏈接：http:protectedvoidinitChannel(SocketChannelch)throwsExceptionChannelPipelinepipeline=pipeline.addLast("decoder",newpipeline.addLast("encoder",newpipeline.addLast(new1516System.out.println("nettyclientstart。。Channelchannel=bootstrap.connect("",Stringtext="HeartbeatRandomrandom=newwhile(channel.isActive())intnum=Thread.sleep(num*2