HSF框架流程设计及原理.pdf

简介：轻量封装 Ibatis3因为本人在国内最大的电子商务公司工作期间，深感一个好的分表分库框架可以大大提高系统的承载能力及系统的灵活性，而一个不好的分表分库方案，则让系统在大数据量处理的时候非常郁闷所以， 在根据笔者在从事电子商务开发的这几年中，对各个应用场景而开发的一个轻量封装 Ibatis3 的一个分表分库框架笔者工作的这几年之中，总结并开发了如下几个框架： summercool（Web 框架，已经应用于某国内大型网络公司的等重要应用） 、 summercool-hsf （基于 Netty实现的 RPC框架， 已经应用国内某移动互联网公司） 、 summercool-ddl（基于 Mybaits 的分表分库框架，已经应用国内某移动互联网公司） ；相继缓存方案、和消息系统解决方案也会慢慢开源Summercool 框架做为笔者的第一个开源框架summercool-hsf：http://summercool- 1. . 什什么么是是HHS SF F 框框架架HSF 框架是一个高性能远程通信框架，底层基于 Netty 实现 TCP 通信，对上层进行封装，提供易于使用和高度可扩展能力名词解译：1）Channel：可以理解为一个通道，或者连接2）ChannelGroup：多个通道组合成为一个 ChannelGroup2 2. .HHS SF F 工工作作流流程程3 3. .消消息息协协议议设设计计消息协议这里是指对消息编码和解码的规范的一种定义，HSF 内置的消息协议采用如下结构：Length：以 4 个字节表示，是指 ID + Content 的长度。

ID：以 1 个字节表示，1 表示 Content 部分被压缩，0 表示未被压缩Content：真实的消息内容4 4. .处处理理器器Netty 框架原生提供一个处理器链对事件进行处理，每个处理器均实现 ChannelHandler 接口，ChannelHandler 是个空接口，拥有三个子接口：ChannelDownstreamHandler, ChannelUpstreamHandler 和LifeCycleAwareChannelHandler这里我们主要关注前两个接口，因为它们被用来处理读与写的消息事件主要分为三种：ChannelEvent、MessageEvent 和 ExceptionEvent，一旦这些事件被触发，它们将从处理器链的一端到另一端，被逐个处理器处理，注意，整个过程是单线程场景一般而言，ChannelEvent 和 ExceptionEvent 事件都是从底层被触发，因此，它们会被 ChannelUpstreamHandler 处理而 MessageEvent 则需要根据读与写方式的不同，分别从两个方向被 ChannelUpstreamHandler 和 ChannelDownstreamHandler 处理。

HSF 内置的编(解)码处理器、压缩(解压)处理器及序列化(反序列化)处理器等都是直接或间接实现 ChannelHandlerC Ch ha an nn ne el lD Do owwn ns st tr re ea ammHHa an nd dl le er rJ Ja av va a 代代码码1.p pu ub bl li ic c i in nt te er rf fa ac ce e ChannelDownstreamHandler e ex xt te en nd ds s ChannelHandler {2./**3.* Handles the specified downstream event.4.*5.* @param ctx the context object for this handler6.* @param ethe downstream event to process or intercept7.*/8.v vo oi id d handleDownstream(ChannelHandlerContext ctx, ChannelEvent e) t th hr ro owws s Exception; 9.}C Ch ha an nn ne el lU Up ps st tr re ea ammHHa an nd dl le er rJ Ja av va a 代代码码1.p pu ub bl li ic c i in nt te er rf fa ac ce e ChannelUpstreamHandler e ex xt te en nd ds s ChannelHandler {2./**3.* Handles the specified upstream event.4.*5.* @param ctx the context object for this handler6.* @param ethe upstream event to process or intercept7.*/8.v vo oi id d handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) t th hr ro owws s Exception; 9.}4 4. .1 1. .E En nc co od di in ng g 流流程程HSF 内置的 encoding 过程由三个 Handler 组合完成，流程如下：1) SerializeDownstreamHandlerJ Ja av va a 代代码码1./**2.* @Title: SerializeDownstreamHandler.java3.* @Package ty.channelhandler.downstream4.* @Description: 序列化5.* @author6.* @date 2011-9-16 下午 4:45:597.* @version V1.08.*/9.p pu ub bl li ic c c cl la as ss s SerializeDownstreamHandler i immp pl le emme en nt ts s ChannelDownstreamHandler {10.Logger logger = LoggerFactory.getLogger(getClass());11.p pr ri iv va at te e Serializer serializer = n ne eww KryoSerializer();12.p pu ub bl li ic c SerializeDownstreamHandler() {13.}14.p pu ub bl li ic c v vo oi id d handleDownstream(ChannelHandlerContext ctx, ChannelEvent e) t th hr ro owws s Exception {15.i if f (!(e i in ns st ta an nc ce eo of f MessageEvent)) {16.ctx.sendDownstream(e);17.r re et tu ur rn n;18.}19.MessageEvent event = (MessageEvent) e;20.Object originalMessage = event.getMessage();21.Object encodedMessage = originalMessage;22.i if f (!(originalMessage i in ns st ta an nc ce eo of f Heartbeat)) {23.encodedMessage = serializer.serialize(originalMessage);24.} e el ls se e {25.encodedMessage = Heartbeat.BYTES;26.}27.i if f (originalMessage == encodedMessage) {28.ctx.sendDownstream(e);29.} e el ls se e i if f (encodedMessage != n nu ul ll l) {30.write(ctx, e.getFuture(), encodedMessage, event.getRemoteAddress());31.}32.}33.p pu ub bl li ic c v vo oi id d setSerializer(Serializer serializer) {34.t th hi is s.serializer = serializer;35.} 36.}2）CompressionDownstreamHandlerJ Ja av va a 代代码码1./**2.* @Title: CompressionDownstreamHandler.java3.* @Package ty.channelhandler.downstream4.* @Description: 压缩处理器5.* @author6.* @date 2011-9-16 下午 4:45:597.* @version V1.08.*/9.p pu ub bl li ic c c cl la as ss s CompressionDownstreamHandler i immp pl le emme en nt ts s ChannelDownstreamHandler {10.p pr ri iv va at te e CompressionStrategy compressionStrategy = n ne eww ThresholdCompressionStrategy();11.p pu ub bl li ic c v vo oi id d handleDownstream(ChannelHandlerContext ctx, ChannelEvent e) t th hr ro owws s Exception {12.i if f (!(e i in ns st ta an nc ce eo of f MessageEvent)) {13.ctx.sendDownstream(e);14.r re et tu ur rn n;15.}16.MessageEvent event = (MessageEvent) e;17.Object originalMessage = event.getMessage();18.i if f (originalMessage i in ns st ta an nc ce eo of f b by yt te e[]) {19.CompressionResult compressionResult = compressionSpress((b by yt te e[]) originalMessage);20.b by yt te e[] resBuffer = compressionResult.getBuffer();21.i in nt t length = resBuffer.length;22.b by yt te e[] bytes = n ne eww b by yt te e[length + 1];23.bytes[0] = compressionResult.isCompressed() ? (b by yt te e) 1 : (b by y。