Tomcat集群高可用

　　Tomcat集群高可用集群架构总览DNSRoundRobinLoadBalancerCluster1Cluster2Tomcat1Tomcat2Tomcat3Tomcat4组件ServerServiceEngineClusterHostClusterContext（1N）ManagerDeltaManagerBackupManagerChannelInterceptor1。。InterceptorNReceiverSenderMembershipValveReplicationValveJvmRouteBinderValveLifecycleListenerClusterListenerClusterSessionListenerDeployerFarmWarDeployer工作原理1。TomcatA启动TomcatA使用标准的启动顺序启动，当Host被创建时，会有一个cluster对象跟其关联。当contexts在解析过程中发现web。xml中含有distributable标签，Tomcat会让Cluster（SimpleTcpCluster）创建一个Manager（默认DeltaManager）用于备份。如果是集群模式，会启动一个membershipservice（multicast）和一个replicationservice（tcpunicast）2。TomcatB启动（等待TomcatA启动完成）TomcatB跟TomcatA以相同的顺序启动，但会跟TomcatA互相建个一个关系（TomcatA，TomcatB），在启动前TomcatB会向TomcatA请求Session，如果60秒未响应已超时处理，记录一条异常日志并继续启动。推荐TomcatA和TomcatB以相同的配置启动3。TomcatA收到一个请求，创建SessionS1TomcatA收到一个请求，请求处理方式跟非集群方式一样。区别点是处理完后在响应给客户端之前ReplicationValve会通过TCP方式将Session同步给TomcatB。为了性能考虑，会有一个useDirtyFlag参数用于备份Session，只有达到指定次数才会备份。4。TomcatA崩溃如果TomcatA崩溃了，TomcatB会将其从成员列表中移除。TomcatA不会在引起TomcatB的任何变化，负载均衡会将所有到TomcatA的请求重定向到TomcatB5。TomcatB收到TomcatA同步过来的SessionS1TomcatB跟处理其他请求一样6。TomcatA恢复启动跟步骤1，2一样，完全启动前会加入Cluster，向TomcatB请求同步所有Session7。TomcatA收到一个请求，来自S1的客户端的无效（Session过期）调用请求完成后，TomcatA向TomcatB发送过期事件，TomcatB将S1移除8。TomcatB收到一个请求，创建SessionS2跟第3步一样9。TomcatA中SessionS2超时由于一直没有收到任何关于S2的请求如果TomcatA没有收到任何请求时，不会将失效的S2备份给其他成员集群配置Tomcat官方默认配置ClusterclassNameorg。apache。catalina。ha。tcp。SimpleTcpClusterchannelSendOptions8ManagerclassNameorg。apache。catalina。ha。session。DeltaManagerexpireSessionsOnShutdownfalsenotifyListenersOnReplicationtrueChannelclassNameorg。apache。catalina。tribes。group。GroupChannelMembershipclassNameorg。apache。catalina。tribes。membership。McastServiceaddress228。0。0。4port45564frequency500dropTime3000ReceiverclassNameorg。apache。catalina。tribes。transport。nio。NioReceiveraddressautoport4000autoBind100selectorTimeout5000maxThreads6SenderclassNameorg。apache。catalina。tribes。transport。ReplicationTransmitterTransportclassNameorg。apache。catalina。tribes。transport。nio。PooledParallelSenderSenderInterceptorclassNameorg。apache。catalina。tribes。group。interceptors。TcpFailureDetectorInterceptorclassNameorg。apache。catalina。tribes。group。interceptors。MessageDispatchInterceptorInterceptorclassNameorg。apache。catalina。tribes。group。interceptors。TcpPingInterceptorChannelValveclassNameorg。apache。catalina。ha。tcp。ReplicationValvefilterValveclassNameorg。apache。catalina。ha。session。JvmRouteBinderValveDeployerclassNameorg。apache。catalina。ha。deploy。FarmWarDeployertempDirtmpwartempdeployDirtmpwardeploywatchDirtmpwarlistenwatchEnabledfalseClusterListenerclassNameorg。apache。catalina。ha。session。ClusterSessionListenerCluster启用组播（Windows默认支持，Linux系统需手动开启）sudorouteaddnet224。0。0。0netmask240。0。0。0devwlp1s0sudorouteaddnet224。0。0。0netmask240。0。0。0devlosudorouteaddnet228。0。0。4netmask255。255。255。255devwlp1s0本地环回地址必须添加，不然没有效果sudorouteaddnet228。0。0。4netmask255。255。255。255devlo集群模型ClusterpublicStringgetClusterName（）；publicvoidsetClusterName（StringclusterName）；Session管理publicManagercreateManager（Stringname）；publicvoidregisterManager（Managermanager）；publicvoidremoveManager（Managermanager）；后台进程，用于热部署监控和心跳检测publicvoidbackgroundProcess（）；Channel添加拦截器publicvoidaddInterceptor（ChannelInterceptorinterceptor）；发送消息publicUniqueIdsend（Member〔〕destination，Serializablemsg，intoptions）throwsChannelException；publicUniqueIdsend（Member〔〕destination，Serializablemsg，intoptions，ErrorHandlerhandler）throwsChannelException；心跳检测publicvoidheartbeat（）；publicvoidsetHeartbeat（booleanenable）；成员回调publicvoidaddMembershipListener（MembershipListenerlistener）；publicvoidaddChannelListener（ChannelListenerlistener）；publicvoidremoveMembershipListener（MembershipListenerlistener）；publicvoidremoveChannelListener（ChannelListenerlistener）；publicbooleanhasMembers（）；publicMember〔〕getMembers（）；publicMembergetLocalMember（booleanincAlive）；ChannelReceiverpublicvoidstart（）throwsIOException；publicvoidstop（）；publicStringgetHost（）；publicintgetPort（）；publicintgetSecurePort（）；publicintgetUdpPort（）；publicvoidheartbeat（）；publicvoidsetMessageListener（MessageListenerlistener）；publicMessageListenergetMessageListener（）；publicChannelgetChannel（）；publicvoidsetChannel（Channelchannel）；ChannelSenderpublicvoidadd（Membermember）；publicvoidremove（Membermember）；publicvoidstart（）throwsIOException；publicvoidstop（）；publicvoidheartbeat（）；publicvoidsendMessage（ChannelMessagemessage，Member〔〕destination）throwsChannelException；publicChannelgetChannel（）；publicvoidsetChannel（Channelchannel）；MemberpublicStringgetName（）；publicbyte〔〕getHost（）；publicintgetPort（）；publicintgetSecurePort（）；publicintgetUdpPort（）；publiclonggetMemberAliveTime（）；publicvoidsetMemberAliveTime（longmemberAliveTime）；publicbooleanisReady（）；publicbooleanisSuspect（）；publicbooleanisFailing（）；publicbyte〔〕getUniqueId（）；publicbyte〔〕getPayload（）；publicvoidsetPayload（byte〔〕payload）；publicbyte〔〕getCommand（）；publicvoidsetCommand（byte〔〕command）；publicbyte〔〕getDomain（）；publicbyte〔〕getData（booleangetalive）；publicbyte〔〕getData（booleangetalive，booleanreset）；publicintgetDataLength（）；当前启动的TomcatpublicbooleanisLocal（）；publicvoidsetLocal（booleanlocal）；ChannelInterceptor
　　发送消息从前往后，接收消息回调从后往前publicintgetOptionFlag（）；publicvoidsetOptionFlag（intflag）；publicvoidsetNext（ChannelInterceptornext）；publicChannelInterceptorgetNext（）；publicvoidsetPrevious（ChannelInterceptorprevious）；publicChannelInterceptorgetPrevious（）；publicvoidsendMessage（Member〔〕destination，ChannelMessagemsg，InterceptorPayloadpayload）throwsChannelException；publicvoidmessageReceived（ChannelMessagedata）；publicvoidheartbeat（）；publicbooleanhasMembers（）；publicMember〔〕getMembers（）；publicMembergetLocalMember（booleanincAliveTime）；publicMembergetMember（Membermbr）；publicvoidstart（intsvc）throwsChannelException；publicvoidstop（intsvc）throwsChannelException；publicvoidfireInterceptorEvent（InterceptorEventevent）；publicChannelgetChannel（）；publicvoidsetChannel（Channelchannel）；MembershipServicepublicvoidsetProperties（java。util。Propertiesproperties）；publicjava。util。PropertiesgetProperties（）；publicvoidstart（）throwsjava。lang。Exception；publicvoidstart（intlevel）throwsjava。lang。Exception；publicvoidstop（intlevel）；publicbooleanhasMembers（）；publicMembergetMember（Membermbr）；publicMember〔〕getMembers（）；publicMembergetLocalMember（booleanincAliveTime）；publicString〔〕getMembersByName（）；publicMemberfindMemberByName（Stringname）；publicvoidsetLocalMemberProperties（StringlistenHost，intlistenPort，intsecurePort，intudpPort）；publicvoidsetMembershipListener（MembershipListenerlistener）；publicvoidremoveMembershipListener（）；publicvoidsetPayload（byte〔〕payload）；publicvoidsetDomain（byte〔〕domain）；publicvoidbroadcast（ChannelMessagemessage）throwsChannelException；publicChannelgetChannel（）；publicvoidsetChannel（Channelchannel）；publicMembershipProvidergetMembershipProvider（）；实现原理Tomcat集群启动入口
　　tomcat各功能模块统一实现接口Lifecycle，模块的生命周期由顶层统一控制，各模块的启动从上到下启动各服务。ContainerBasestartInternalOverrideprotectedsynchronizedvoidstartInternal（）throwsLifecycleException｛。。。。。。cluster为server。xml中配置的SimpleTcpClusterClusterclassNameorg。apache。catalina。ha。tcp。SimpleTcpClusterchannelSendOptions8。。。ClusterClusterclustergetClusterInternal（）；if（clusterinstanceofLifecycle）｛（（Lifecycle）cluster）。start（）；｝。。。。。。｝Session同步入口ReplicationValvepublicvoidinvoke（Requestrequest，Responseresponse）throwsIOException，ServletException｛。。。。。。。处理请求getNext（）。invoke（request，response）；处理完成后响应用户前同步Sessionif（context！nullcluster！nullcontext。getManager（）instanceofClusterManager）｛ClusterManagerclusterManager（ClusterManager）context。getManager（）；valveclustercanaccessmanagerotherclusterhandlereplicationathostlevelhopefully！if（cluster。getManager（clusterManager。getName（））null）｛return；｝if（cluster。hasMembers（））｛sendReplicationMessage（request，totalstart，isCrossContext，clusterManager）；｝else｛resetReplicationRequest（request，isCrossContext）；｝｝｝finally｛。。。。。｝｝protectedvoidsendReplicationMessage（Requestrequest，longtotalstart，booleanisCrossContext，ClusterManagerclusterManager）｛try｛sendinvalidsessionsDeltaManagerreturnsString〔0〕if（！（clusterManagerinstanceofDeltaManager））｛sendInvalidSessions（clusterManager）；｝sendreplicationsendSessionReplicationMessage（request，clusterManager）；if（isCrossContext）｛sendCrossContextSession（）；｝｝catch（Exceptionx）｛FIXMEwehavealotofsends，butthetroublewithonenodestopsthecorrectreplicationtoothernodes！log。error（sm。getString（ReplicationValve。send。failure），x）；｝finally｛｝｝通信协议心跳检测
　　集群间会话的复制采用类似RPC的方式，直接序列化整个类发送给集群成员TRIBESB10bytespackagelength4bytesalive8bytesport4bytessecureport4bytesudpport4byteshostlength1bytehost1bytescommandlength4bytescommandcommandlengthbytesdlen4bytesdomaindlenbytesuniqueId16bytespayloadlength4bytespayloadplenbytesTRIBESE10bytes组件介绍SimpleTcpCluster启用集群protectedvoidstartInternal（）throwsLifecycleException｛try｛checkDefaults（）；registerClusterValve（）；channel。addMembershipListener（this）；channel。addChannelListener（this）；channel。setName（getClusterName（）Channel）；channelStartOptions15，即默认启用四个组件channel。start（channelStartOptions）；if（clusterDeployer！null）｛clusterDeployer。start（）；｝registerMember（channel。getLocalMember（false））；｝catch（Exceptionx）｛log。error（sm。getString（simpleTcpCluster。startUnable），x）；thrownewLifecycleException（x）；｝｝心跳检测publicvoidbackgroundProcess（）｛if（clusterDeployer！null）｛clusterDeployer。backgroundProcess（）；｝sendaheartbeatthroughthechannelif（isHeartbeatBackgroundEnabled（）channel！null）｛channel。heartbeat（）；｝periodiceventfireLifecycleEvent（Lifecycle。PERIODICEVENT，null）；｝消息回调publicvoidmessageReceived（ClusterMessagemessage）｛invokeallthelistenersbooleanacceptedfalse；if（message！null）｛for（ClusterListenerlistener：clusterListeners）｛if（listener。accept（message））｛acceptedtrue；listener。messageReceived（message）；｝｝。。。。。｝ChannelCoordinator
　　位于Interceptor最后一个，消息发送和消息接收由他进行协调处理ChannelListener1。。ChannelListenerNMembershipListener1。。MembershipListenerN〔ApplicationLayer〕ChannelChannelInterceptor1〔Channelstack〕ChannelInterceptorNCoordinator（implementsMessageListener，MembershipListener，ChannelInterceptor）MembershipServiceChannelSenderChannelReceiver〔IOlayer〕svc默认15protectedsynchronizedvoidinternalStart（intsvc）throwsChannelException｛try｛booleanvalidfalse；makesurewedontpassdownanyflagsthatareunrelatedtothebottomlayersvcsvcChannel。DEFAULT；。。。。。。muststartthereceiverfirstsothatwecancoordinatetheportitlistenstowiththelocalmembershipsettingsNioReceiver，通信协议为TCP，用于同步Sessionif（Channel。SNDRXSEQ（svcChannel。SNDRXSEQ））｛clusterReceiver。setMessageListener（this）；clusterReceiver。setChannel（getChannel（））；clusterReceiver。start（）；synchronize，bigtimeFIXMEMemberlocalMembergetChannel（）。getLocalMember（false）；if（localMemberinstanceofStaticMember）｛staticmemberStaticMemberstaticMember（StaticMember）localMember；staticMember。setHost（getClusterReceiver（）。getHost（））；staticMember。setPort（getClusterReceiver（）。getPort（））；staticMember。setSecurePort（getClusterReceiver（）。getSecurePort（））；｝else｛multicastmembermembershipService。setLocalMemberProperties（getClusterReceiver（）。getHost（），getClusterReceiver（）。getPort（），getClusterReceiver（）。getSecurePort（），getClusterReceiver（）。getUdpPort（））；｝validtrue；｝NioSender，通信协议为TCP，用于发送Session信息if（Channel。SNDTXSEQ（svcChannel。SNDTXSEQ））｛clusterSender。setChannel（getChannel（））；clusterSender。start（）；validtrue；｝集群成员间的消息接收入口，采用组播方式通信协议为UDP接收心跳检测信息，Ping信息if（Channel。MBRRXSEQ（svcChannel。MBRRXSEQ））｛membershipService。setMembershipListener（this）；membershipService。setChannel（getChannel（））；if（membershipServiceinstanceofMcastService）｛（（McastService）membershipService）。setMessageListener（this）；｝membershipService。start（MembershipService。MBRRX）；validtrue；｝集群成员间的消息发送入口，采用组播方式通信协议为UDPif（Channel。MBRTXSEQ（svcChannel。MBRTXSEQ））｛membershipService。setChannel（getChannel（））；membershipService。start（MembershipService。MBRTX）；validtrue；｝。。。。。。｝catch（ChannelExceptioncx）｛throwcx；｝catch（Exceptionx）｛thrownewChannelException（x）；｝｝McastServicepublicvoidstart（intlevel）throwsjava。lang。Exception｛。。。。。if（impl！null）｛impl。start（level）；return；｝StringhostgetProperties（）。getProperty（tcpListenHost）；intportInteger。parseInt（getProperties（）。getProperty（tcpListenPort））；intsecurePortInteger。parseInt（getProperties（）。getProperty（tcpSecurePort））；intudpPortInteger。parseInt（getProperties（）。getProperty（udpListenPort））；if（localMembernull）｛localMembernewMemberImpl（host，port，100）；localMember。setUniqueId（UUIDGenerator。randomUUID（true））；localMember。setLocal（true）；｝else｛localMember。setHostname（host）；localMember。setPort（port）；localMember。setMemberAliveTime（100）；｝localMember。setSecurePort（securePort）；localMember。setUdpPort（udpPort）；。。。。。。implnewMcastServiceImpl（localMember，Long。parseLong（properties。getProperty（mcastFrequency）），Long。parseLong（properties。getProperty（memberDropTime）），Integer。parseInt（properties。getProperty（mcastPort）），bind，java。net。InetAddress。getByName（properties。getProperty（mcastAddress）），ttl，soTimeout，this，this，Boolean。parseBoolean（properties。getProperty（localLoopbackDisabled）））；。。。。。。impl。start（level）；。。。。。。｝McastServiceImpl实例化时初始化动作publicvoidinit（）throwsIOException｛setupSocket（）；发送包sendPacketnewDatagramPacket（newbyte〔MAXPACKETSIZE〕，MAXPACKETSIZE）；sendPacket。setAddress（address）；sendPacket。setPort（port）；接收包receivePacketnewDatagramPacket（newbyte〔MAXPACKETSIZE〕，MAXPACKETSIZE）；receivePacket。setAddress（address）；receivePacket。setPort（port）；member。setCommand（newbyte〔0〕）；if（membershipnull）｛membershipnewMembership（member）；｝｝protectedvoidsetupSocket（）throwsIOException｛if（mcastBindAddress！null）｛try｛socketnewMulticastSocket（newInetSocketAddress（address，port））；｝catch（BindExceptione）｛socketnewMulticastSocket（port）；｝｝else｛socketnewMulticastSocket（port）；｝。。。。。。｝启动入口publicsynchronizedvoidstart（intlevel）throwsIOException｛booleanvalidfalse；if（（levelChannel。MBRRXSEQ）Channel。MBRRXSEQ）｛if（receiver！null）｛thrownewIllegalStateException（sm。getString（mcastServiceImpl。receive。running））；｝try｛if（sendernull）｛将地址加入组播组广播与组播的区别是：广播向组里面的每个成员发送消息，组播同时向每个成员发送消息socket。joinGroup（address）；｝｝catch（IOExceptioniox）｛log。error（sm。getString（mcastServiceImpl。unable。join））；throwiox；｝doRunReceivertrue；接收线程receivernewReceiverThread（）；receiver。setDaemon（true）；receiver。start（）；validtrue；｝if（（levelChannel。MBRTXSEQ）Channel。MBRTXSEQ）｛if（sender！null）｛thrownewIllegalStateException（sm。getString（mcastServiceImpl。send。running））；｝if（receivernull）｛socket。joinGroup（address）；｝makesureatleastonepacketgetsoutthere手动发送一个send（false）；doRunSendertrue；发送线程sendernewSenderThread（sendFrequency）；sender。setDaemon（true）；sender。start（）；validtrue；｝pause，onceortwice线程休眠一个周期，该周期可配置waitForMembers（level）；｝GroupChannelprotectedfinalChannelCoordinatorcoordinatornewChannelCoordinator（）；消息回调处理publicvoidmessageReceived（ChannelMessagemsg）｛if（msgnull）｛return；｝try｛Serializablefwdnull；if（（msg。getOptions（）SENDOPTIONSBYTEMESSAGE）SENDOPTIONSBYTEMESSAGE）｛fwdnewByteMessage（msg。getMessage（）。getBytes（））；｝else｛try｛反序列化，主要内容为（地址，消息内容，消息唯一ID，时间戳）fwdXByteBuffer。deserialize（msg。getMessage（）。getBytesDirect（），0，msg。getMessage（）。getLength（））；｝catch（Exceptionsx）｛log。error（sm。getString（groupChannel。unable。deserialize，msg），sx）；return；｝｝gettheactualmemberwiththecorrectalivetimeMembersourcemsg。getAddress（）；booleanrxfalse；booleandeliveredfalse；for（ChannelListenerchannelListener：channelListeners）｛if（channelListener！nullchannelListener。accept（fwd，source））｛最终由SimpleTcpCluster处理channelListener。messageReceived（fwd，source）；deliveredtrue；ifthemessagewasacceptedbyanRPCchannel，thatchannelisresponsibleforreturningthereply，otherwisewesendanabsencereplyif（channelListenerinstanceofRpcChannel）｛rxtrue；｝｝｝｝catch（Exceptionx）｛thiscouldbethechannellistenerthrowinganexception，weshouldlogitasawarning。if（log。isWarnEnabled（））｛log。warn（sm。getString（groupChannel。receiving。error），x）；｝thrownewRemoteProcessException（sm。getString（groupChannel。receiving。error），x）；｝｝NioReceiver
　　Tcp协议接收成员消息绑定地址与端口protectedvoidbind（）throwsIOException｛serverChannelServerSocketChannel。open（）；ServerSocketserverSocketserverChannel。socket（）；this。selector。set（Selector。open（））；根据server。xml中Receiver标签中配置的4000端口开始检测可用的数据，最大到4100bind（serverSocket，getPort（），getAutoBind（））；serverChannel。configureBlocking（false）；serverChannel。register（this。selector。get（），SelectionKey。OPACCEPT）；if（this。getUdpPort（）0）｛datagramChannelDatagramChannel。open（）；configureDatagraChannel（）；bindUdp（datagramChannel。socket（），getUdpPort（），getAutoBind（））；｝｝监听端口protectedvoidlisten（）throwsException｛setListen（true）；Selectorselectorthis。selector。get（）；if（selector！nulldatagramChannel！null）｛ObjectReaderoreadernewObjectReader（MAXUDPSIZE）；maxsizeforadatagrampacketregisterChannel（selector，datagramChannel，SelectionKey。OPREAD，oreader）；｝while（doListen（）selector！null）｛thismayblockforalongtime，uponreturntheselectedsetcontainskeysofthereadychannelstry｛events（）；超时检测socketTimeouts（）；intnselector。select（getSelectorTimeout（））；if（n0）｛continue；nothingtodo｝getaniteratoroverthesetofselectedkeysIteratorSelectionKeyitselector。selectedKeys（）。iterator（）；lookateachkeyintheselectedsetwhile（it！nullit。hasNext（））｛SelectionKeykeyit。next（）；Isanewconnectioncomingin？if（key。isAcceptable（））｛ServerSocketChannelserver（ServerSocketChannel）key。channel（）；SocketChannelchannelserver。accept（）；接收缓冲区大小channel。socket（）。setReceiveBufferSize（getRxBufSize（））；发送缓冲区大小channel。socket（）。setSendBufferSize（getTxBufSize（））；是否启用Nagle算法Nagle算法：解决网络拥塞，如果搞频率只发送一个字节的数据，实际上发送的是41个字节的数据包，其中包含40个字节的Tcp协议头Nagle算法原理是在未接收到ACK之前将要发送的数据存入缓冲区，知道收到ACK或者缓冲的数据到达一定大小的数据再发送将小块的数据收集成大块数据后在发送channel。socket（）。setTcpNoDelay（getTcpNoDelay（））；长时间未收到ACK由操作系统发送一个心跳检测包channel。socket（）。setKeepAlive（getSoKeepAlive（））；套接字上接收的所有TCP紧急数据（OutOfBound，带外数据）都将通过套接字输入流接收，默认丢弃紧急数据channel。socket（）。setOOBInline（getOoBInline（））；地址重用channel。socket（）。setReuseAddress（getSoReuseAddress（））；用来控制Socket关闭Close（）方法的行为，执行Close（）方位会立即返回，但是底层的Socket会延迟一段时间，将缓冲区中的数据发送给对方channel。socket（）。setSoLinger（getSoLingerOn（），getSoLingerTime（））；读超时channel。socket（）。setSoTimeout（getTimeout（））；ObjectattachnewObjectReader（channel）；registerChannel（selector，channel，SelectionKey。OPREAD，attach）；｝istheredatatoreadonthischannel？if（key。isReadable（））｛readDataFromSocket（key）；｝else｛key。interestOps（key。interestOps（）（SelectionKey。OPWRITE））；｝it。remove（）；｝｝catch（java。nio。channels。ClosedSelectorExceptioncse）｛ignoreisnormalatshutdownorstoplistensocket｝catch（java。nio。channels。CancelledKeyExceptionnx）｛log。warn（sm。getString（nioReceiver。clientDisconnect））；｝catch（Throwablet）｛ExceptionUtils。handleThrowable（t）；log。error（sm。getString（nioReceiver。requestError），t）；｝｝serverChannel。close（）；if（datagramChannel！null）｛try｛datagramChannel。close（）；｝catch（Exceptioniox）｛if（log。isDebugEnabled（））｛log。debug（Unabletoclosedatagramchannel。，iox）；｝｝datagramChannelnull；｝closeSelector（）；｝读取数据，将其放入任务池中处理protectedvoidreadDataFromSocket（SelectionKeykey）throwsException｛NioReplicationTasktask（NioReplicationTask）getTaskPool（）。getRxTask（）；if（tasknull）｛Nothreadstasksavailable，donothing，theselectionloopwillkeepcallingthismethoduntilathreadbecomesavailable，thethreadpoolitselfhasawaitingmechanismsowewillnotwaithere。if（log。isDebugEnabled（））｛log。debug（NoTcpReplicationThreadavailable）；｝｝else｛invokingthiswakesuptheworkerthreadthenreturnsaddtasktothreadpooltask。serviceChannel（key）；getExecutor（）。execute（task）；｝｝protectedvoiddrainChannel（finalSelectionKeykey，ObjectReaderreader）throwsException｛reader。access（）；ReadableByteChannelchannel（ReadableByteChannel）key。channel（）；intcount1；buffer。clear（）；makebufferemptySocketAddresssaddrnull；if（channelinstanceofSocketChannel）｛loopwhiledataavailable，channelisnonblockingwhile（（countchannel。read（buffer））0）｛buffer。flip（）；makebufferreadableif（buffer。hasArray（））｛reader。append（buffer。array（），0，count，false）；｝else｛reader。append（buffer，count，false）；｝buffer。clear（）；makebufferemptydowehaveatleastonepackage？if（reader。hasPackage（））｛break；｝｝｝elseif（channelinstanceofDatagramChannel）｛DatagramChanneldchannel（DatagramChannel）channel；saddrdchannel。receive（buffer）；buffer。flip（）；makebufferreadableif（buffer。hasArray（））｛reader。append（buffer。array（），0，buffer。limit（）buffer。position（），false）；｝else｛reader。append（buffer，buffer。limit（）buffer。position（），false）；｝buffer。clear（）；makebufferemptydidwegetapackagecountreader。hasPackage（）？1：1；｝intpkgcntreader。count（）；if（count0pkgcnt0）｛endofstream，andnomorepackagestoprocessremoteEof（key）；return；｝ChannelMessage〔〕msgspkgcnt0？ChannelData。EMPTYDATAARRAY：reader。execute（）；registerForRead（key，reader）；registertoreadnewdata，beforewesenditofftoavoiddeadlocksfor（ChannelMessagemsg：msgs）｛UsesendackhereifyouwanttoacktherequesttotheremoteserverbeforecompletingtherequestThisisconsideredanasynchronousrequestif（ChannelData。sendAckAsync（msg。getOptions（）））｛sendAck（key，（WritableByteChannel）channel，Constants。ACKCOMMAND，saddr）；｝try｛if（Logs。MESSAGES。isTraceEnabled（））｛try｛Logs。MESSAGES。trace（NioReplicationThreadReceivedmsg：newUniqueId（msg。getUniqueId（））atnewjava。sql。Timestamp（System。currentTimeMillis（）））；｝catch（Throwablet）｛｝｝核心逻辑，消息回调处理。由ChannelCoordinator协调处理getCallback（）。messageDataReceived（msg）；UsesendackhereifyouwanttherequesttocompleteonthisserverbeforesendingtheacktotheremoteserverThisisconsideredasynchronizedrequestif（ChannelData。sendAckSync（msg。getOptions（）））｛sendAck（key，（WritableByteChannel）channel，Constants。ACKCOMMAND，saddr）；｝｝catch（RemoteProcessExceptione）｛if（log。isDebugEnabled（））｛log。error（sm。getString（nioReplicationTask。process。clusterMsg。failed），e）；｝if（ChannelData。sendAckSync（msg。getOptions（）））｛sendAck（key，（WritableByteChannel）channel，Constants。FAILACKCOMMAND，saddr）；｝｝catch（Exceptione）｛log。error（sm。getString（nioReplicationTask。process。clusterMsg。failed），e）；if（ChannelData。sendAckSync（msg。getOptions（）））｛sendAck（key，（WritableByteChannel）channel，Constants。FAILACKCOMMAND，saddr）；｝｝if（getUseBufferPool（））｛BufferPool。getBufferPool（）。returnBuffer（msg。getMessage（））；msg。setMessage（null）；｝｝if（count0）｛remoteEof（key）；｝｝ChannelCoordinator由后向前处理，即Server。xml中配置的Interceptor标签，最前一个为GroupChannelOverridepublicvoidmessageReceived（ChannelMessagemsg）｛if（getPrevious（）！null）｛getPrevious（）。messageReceived（msg）；｝｝NioSender
　　Tcp协议，集群间的消息发送接口publicbooleanprocess（SelectionKeykey，booleanwaitForAck）throwsIOException｛intopskey。readyOps（）；key。interestOps（key。interestOps（）ops）；incasedisconnecthasbeencalledif（（！isConnected（））（！connecting））｛thrownewIOException（sm。getString（nioSender。sender。disconnected））；｝if（！key。isValid（））｛thrownewIOException（sm。getString（nioSender。key。inValid））；｝if（key。isConnectable（））｛if（socketChannel。finishConnect（））｛completeConnect（）；if（current！null）｛key。interestOps（key。interestOps（）SelectionKey。OPWRITE）；｝returnfalse；｝else｛waitfortheconnectiontofinishkey。interestOps（key。interestOps（）SelectionKey。OPCONNECT）；returnfalse；｝endif｝elseif（key。isWritable（））｛booleanwritecompletewrite（）；if（writecomplete）｛wearecompleted，shouldwereadanack？if（waitForAck）｛registertoreadtheackkey。interestOps（key。interestOps（）SelectionKey。OPREAD）；｝else｛ifnot，weareready，setMessagewillreregisterusforanotherwriteinterestdoahealthcheck，wehavenowayofverifyadisconnectedsocketsincewedontregisterforOPREADonwaitForAckfalseread（）；thiscausesoverheadsetRequestCount（getRequestCount（）1）；returntrue；｝｝else｛wearenotcomplete，letswritesomemorekey。interestOps（key。interestOps（）SelectionKey。OPWRITE）；｝endif｝elseif（key。isReadable（））｛booleanreadcompleteread（）；if（readcomplete）｛setRequestCount（getRequestCount（）1）；returntrue；｝else｛key。interestOps（key。interestOps（）SelectionKey。OPREAD）；｝endif｝else｛unknownstate，shouldneverhappenlog。warn（sm。getString（nioSender。unknown。state，Integer。toString（ops）））；thrownewIOException（sm。getString（nioSender。unknown。state，Integer。toString（ops）））；｝endifreturnfalse；｝消息发送protectedbooleanwrite（）throwsIOException｛if（（！isConnected（））（this。socketChannelnullthis。dataChannelnull））｛thrownewIOException（sm。getString（nioSender。not。connected））；｝if（current！null）｛if（remaining0）｛wehavewritteneverything，orwearestartinganewpackageprotectagainstbufferoverwriteintbyteswrittenisUdpBased（）？dataChannel。write（writebuf）：socketChannel。write（writebuf）；if（byteswritten1）｛thrownewEOFException（）；｝remainingbyteswritten；iftheentiremessagewaswrittenfromthebufferresetthepositioncounterif（remaining0）｛remaining0；｝｝return（remaining0）；｝nomessagetosend，wecanconsiderthatcompletereturntrue；｝ClusterManager
　　ClusterManager主要分为DeltaManager和BackupManager。
　　DeltaManager：会向所有活动的节点同步Session数据，如果节点过多（超过4个官方不推荐使用该模式）不推荐使用
　　BackupManager：向一个备份节点同步Session数据，默认取成员列表中的第一个DeltaManagerprotectedvoidmessageReceived（SessionMessagemsg，Membersender）｛ClassLoadercontextLoaderThread。currentThread（）。getContextClassLoader（）；try｛ClassLoader〔〕loadersgetClassLoaders（）；Thread。currentThread（）。setContextClassLoader（loaders〔0〕）；if（log。isDebugEnabled（））｛log。debug（sm。getString（deltaManager。receiveMessage。eventType，getName（），msg。getEventTypeString（），sender））；｝switch（msg。getEventType（））｛caseSessionMessage。EVTGETALLSESSIONS：handleGETALLSESSIONS（msg，sender）；break；caseSessionMessage。EVTALLSESSIONDATA：handleALLSESSIONDATA（msg，sender）；break；caseSessionMessage。EVTALLSESSIONTRANSFERCOMPLETE：handleALLSESSIONTRANSFERCOMPLETE（msg，sender）；break；caseSessionMessage。EVTSESSIONCREATED：handleSESSIONCREATED（msg，sender）；break；caseSessionMessage。EVTSESSIONEXPIRED：handleSESSIONEXPIRED（msg，sender）；break；caseSessionMessage。EVTSESSIONACCESSED：handleSESSIONACCESSED（msg，sender）；break；caseSessionMessage。EVTSESSIONDELTA：handleSESSIONDELTA（msg，sender）；break；caseSessionMessage。EVTCHANGESESSIONID：handleCHANGESESSIONID（msg，sender）；break；caseSessionMessage。EVTALLSESSIONNOCONTEXTMANAGER：handleALLSESSIONNOCONTEXTMANAGER（msg，sender）；break；default：wedidntrecognizethemessagetype，donothingbreak；｝switch｝catch（Exceptionx）｛log。error（sm。getString（deltaManager。receiveMessage。error，getName（）），x）；｝finally｛Thread。currentThread（）。setContextClassLoader（contextLoader）；｝｝主要看下创建Session事件protectedvoidhandleSESSIONCREATED（SessionMessagemsg，Membersender）｛counterReceiveEVTSESSIONCREATED；if（log。isDebugEnabled（））｛log。debug（sm。getString（deltaManager。receiveMessage。createNewSession，getName（），msg。getSessionID（）））；｝DeltaSessionsession（DeltaSession）createEmptySession（）；session。setValid（true）；session。setPrimarySession（false）；session。setCreationTime（msg。getTimestamp（））；usecontainermaxInactiveIntervalsothatsessionwillexpirecorrectlyincaseofprimarytransfersession。setMaxInactiveInterval（getContext（）。getSessionTimeout（）60，false）；session。access（）；最终执行sessions。put（session。getIdInternal（），session）；将其放入内存中统一管理session。setId（msg。getSessionID（），notifySessionListenersOnReplication）；session。endAccess（）；｝BackupManager
　　备份节点默认取所有节点中的第一个节点，采用Rpc协议进行通信protectedvoidsend（ClusterManagermanager，StringsessionId）｛BackupManager返回nullClusterMessagemsgmanager。requestCompleted（sessionId）；if（msg！nullcluster！null）｛cluster。send（msg）；if（doStatistics（））｛nrOfSendRequests；｝｝｝key为sessionId，complete默认falsepublicvoidreplicate（Objectkey，booleancomplete）｛MapEntryK，VentryinnerMap。get（key）；if（entrynull）｛return；｝if（！entry。isSerializable（））｛return；｝if（entry。isPrimary（）entry。getBackupNodes（）！nullentry。getBackupNodes（）。length0）｛checktoseeifweneedtoreplicatethisobjectisDirty（）completeisAccessReplicate（）ReplicatedMapEntryrentrynull；if（entry。getValue（）instanceofReplicatedMapEntry）｛rentry（ReplicatedMapEntry）entry。getValue（）；｝控制备份频率booleanisDirtyrentry！nullrentry。isDirty（）；booleanisAccessrentry！nullrentry。isAccessReplicate（）；booleanreplcompleteisDirtyisAccess；if（！repl）｛if（log。isTraceEnabled（））｛log。trace（Notreplicating：key，nochangemade）；｝return；｝checktoseeifthemessageisdiffableMapMessagemsgnull；if（rentry！nullrentry。isDiffable（）（isDirtycomplete））｛rentry。lock（）；try｛constructadiffmessagemsgnewMapMessage（mapContextName，getReplicateMessageType（），true，（Serializable）entry。getKey（），null，rentry。getDiff（），entry。getPrimary（），entry。getBackupNodes（））；rentry。resetDiff（）；｝catch（IOExceptionx）｛log。error（sm。getString（abstractReplicatedMap。unable。diffObject），x）；｝finally｛rentry。unlock（）；｝｝if（msgnullcomplete）｛constructacompletemsgnewMapMessage（mapContextName，getReplicateMessageType（），false，（Serializable）entry。getKey（），（Serializable）entry。getValue（），null，entry。getPrimary（），entry。getBackupNodes（））；｝if（msgnull）｛constructaaccessmessagemsgnewMapMessage（mapContextName，MapMessage。MSGACCESS，false，（Serializable）entry。getKey（），null，null，entry。getPrimary（），entry。getBackupNodes（））；｝try｛if（channel！nullentry。getBackupNodes（）！nullentry。getBackupNodes（）。length0）｛if（rentry！null）｛rentry。setLastTimeReplicated（System。currentTimeMillis（））；｝向备份节点发送当前Session信息，备份节点默认所有节点中的第一个节点channel。send（entry。getBackupNodes（），msg，channelSendOptions）；｝｝catch（ChannelExceptionx）｛log。error（sm。getString（abstractReplicatedMap。unable。replicate），x）；｝｝endif｝热部署
　　支持上传War包部署FarmWarDeployerpublicvoidbackgroundProcess（）｛if（started）｛if（watchEnabled）｛count（count1）processDeployFrequency；if（count0）｛定时校验watcher。check（）；｝｝removeInvalidFileFactories（）；｝｝WarWatcherpublicvoidcheck（）｛所有以。war结尾的文件File〔〕listwatchDir。listFiles（newWarFilter（））；if（listnull）｛log。warn（sm。getString（warWatcher。cantListWatchDir，watchDir））；listnewFile〔0〕；｝firstmakesureallthefilesarelistedinourcurrentstatusfor（Filefile：list）｛if（！file。exists（））｛log。warn（sm。getString（warWatcher。listedFileDoesNotExist，file，watchDir））；｝如果上传新war包就会在这儿加入管理addWarInfo（file）；｝CheckallthestatuscodesandupdatetheFarmDeployerfor（IteratorMap。EntryString，WarInfoicurrentStatus。entrySet（）。iterator（）；i。hasNext（）；）｛Map。EntryString，WarInfoentryi。next（）；WarInfoinfoentry。getValue（）；war包如果被变更过返回1war包如果被移除返回1war包如果新增则返回1intcheckinfo。check（）；if（check1）｛部署服务，通过JMX调用服务tryAddServiced进行部署listener。fileModified（info。getWar（））；｝elseif（check1）｛listener。fileRemoved（info。getWar（））；noneedtokeepinmemoryi。remove（）；｝｝｝部署服务publicvoidfileModified（FilenewWar）｛try｛FiledeployWarnewFile（getDeployDirFile（），newWar。getName（））；ContextNamecnnewContextName（deployWar。getName（），true）；if（deployWar。exists（）deployWar。lastModified（）newWar。lastModified（））｛return；｝installlocalif（tryAddServiced（cn。getName（）））｛try｛copy（newWar，deployWar）；｝finally｛removeServiced（cn。getName（））；｝check（cn。getName（））；｝通知集群中的其他成员部署新增的服务install（cn。getName（），deployWar）；｝catch（Exceptionx）｛log。error（sm。getString（farmWarDeployer。modInstallFail），x）；｝｝