https://github.com/xiaotutiger/miyue-music-service/tree/master/miyue-music-service
下面将分析手头上一个项目,运用的技术很全,值得学习,先做一个简单介绍,当然业务部分代码就不讲了。
整个工程采用maven来管理,主要的技术是spring+jedis+netty+disruptor.看这个组合,这个服务器端性能应该很不错。
这个工程又引发我对技术无限热爱 ,哈哈。
这
个工程,目前主要是针对一些基于json/xml/text格式的请求,同时也是支持标准手机请求的,当然,可以自定义一些其他格式或者pc端的请求,而
且针对不同URI,后面挂了不同的handler,这些可能都是一些web处理的基本思想,只是脱离了常规的web容器或者应用服务器。
xml工具采用xstram来处理,两个字,方便。
json工具采用jackson\不知道和业界出名的fastjson\gson\sf.json有何区别,待鉴定。
客
户端的请求,统一继承ClientRequestModel,经过编码统一转化为domainMessage,交由disruptor来处理,其实oop
里什么继承,实现,封装思想,大部分都在围绕一个东西在走,一句话,把看似各有棱角的东西如何转化为共同的东西,求同存异啊(比如,水,石头,空气等,如
果在这一层,我们没法统一用一个特征来表示,我们可以先把它转化为分子,那是不是可以用同一个东西来表示呢?如何高度抽象封装,这真是一门艺术)。
看这个工程对客户端请求,是如何一步步处理的,message->request->event 交由disruptor来处理,很美妙的思想。在了解这些之前,我们有必要深入学习一下disruptor,很特别的一个框架,宣言很牛逼,中文文档在这里(http://ifeve.com/dissecting-disruptor-whats-so-special/),E文好的同学请移步到这里(http://mechanitis.blogspot.com/2011/06/dissecting-disruptor-whats-so-special.html)
了解disruptor之前,先学习下ringbuffer是如何实现的?
1、ringbuffer的特别之处:
只有一个指针,没有尾指针,基于数组,且不会删除元素,元素会覆盖,充分利用缓存行,减少垃圾回收。
2、如何从ringbuffer读取数据:
------------------------------------------2013-9-9 补充-----------------------------------------------------
下面主要讲一下请求如何处理这块架构吧,其实架构这个东西,说简单一点,就是一种简单可扩展的实现方式,在某些程度上,不要太在意性能。
底层通信建立在netty之上,基本没做任何改动
Java代码
public class HttpServerPipelineFactory implements ChannelPipelineFactory {
private ChannelUpstreamHandler channelUpstreamHandler
public ChannelPipeline getPipeline() throws Exception {
// Create a default pipeline implementation.
ChannelPipeline pipeline = pipeline()
// Uncomment the following line if you want HTTPS
//SSLEngine engine = SecureChatSslContextFactory.getServerContext().createSSLEngine()
//engine.setUseClientMode(false)
//pipeline.addLast("ssl", new SslHandler(engine))
pipeline.addLast("decoder", new HttpRequestDecoder())
// Uncomment the following line if you don't want to handle HttpChunks.
pipeline.addLast("aggregator", new HttpChunkAggregator(1048576))
pipeline.addLast("encoder", new HttpResponseEncoder())
// Remove the following line if you don't want automatic content compression.
pipeline.addLast("deflater", new HttpContentCompressor())
//pipeline.addLast("handler", new HttpRequestHandler())
pipeline.addLast("handler", channelUpstreamHandler)
return pipeline
}
public void setChannelUpstreamHandler(ChannelUpstreamHandler channelUpstreamHandler) {
this.channelUpstreamHandler = channelUpstreamHandler
}
}
相关spring配置
Java代码
<bean id="httpServerPipelineFactory" class="com.yunchao.cm.network.http.HttpServerPipelineFactory">
<property name="channelUpstreamHandler" ref="httpRequestHandler"/>
</bean>
Java代码
<bean id="httpRequestHandler" class="com.yunchao.cm.network.http.HttpRequestHandler">
<property name="urlMaps">
<map>
<entry key="/payorder">
<ref bean="payOrderCodecFactory"/>
</entry>
<entry key="/question">
<ref bean="questionCodecFactory"/>
</entry>
<entry key="/sms">
<ref bean="smsCodecFactory"/>
</entry>
代码太多,不全部贴出来,后面整理一下放到我的github上去。
基如此,我们还是得定义一个handler,继承simpleChannelUpstreamHander,并重写了messageReceied方法,具体在这里。
Java代码
QueryStringDecoder queryStringDecoder = new QueryStringDecoder(request.getUri())
String url = queryStringDecoder.getPath()
CodecFactory codecFactory = urlMaps.get(url)
if (null == codecFactory) {
logger.error("unsupported url:{} request.", url)
//sendError(ctx, BAD_REQUEST)
e.getChannel().close()
return
}
//获取cmwap网络中的手机号码
String phone = PhoneUtils.getPhone(request.getHeader("x-up-calling-line-id"))
if (request.getMethod().equals(HttpMethod.POST)) {
ChannelBuffer content = request.getContent()
String postParams = content.toString(CharsetUtil.UTF_8)
logger.debug("request content:{}", postParams)
ClientRequestModel model = (ClientRequestModel) codecFactory.decode(postParams)
model.setProperty(model.MESSAGE_EVENT_KEY, e)
model.setProperty(model.HTTP_REQUEST_KEY, request)
model.setProperty(model.HTTP_PHONE_KEY, phone)
InetSocketAddress remoteAddress = (InetSocketAddress) e.getRemoteAddress()
model.setProperty(model.IP_KEY, remoteAddress.getAddress().getHostAddress())
logger.info("user request model:{}", model)
model.fireSelf()
Java代码
@Override
public DomainMessage fireSelf() {
DomainMessage em = new DomainMessage(this)
EventUtils.fireEvent(em, "alipayNotifyState")
return em
}
看到这里基本上能够清楚了,是如何把客户端请求包装成ClientRequestModel了,且后面涉及到处理的对象,全部继承它,在整个架构之
中,has a 优于 is
a,对于客户端netty的一些对象,也是存储在ClientRequestModel中,codec无非也是采用了xml/json/kv,如斯,实现
了字节与对象之间的转换。
除
此之外,突然想到刚来杭州工作的第一家公司,基于此,采用的架构师servlet充当服务器,因为这是一个公司内部的server,而不是一个平台,采用
的数据格式也比较单一,就是xml,但是采用的外部类库也是xstream来处理的,但是整个系统维持的日调用量也是在百万级别,运用的client则是
采用httpclient,对于不同请求后面挂的handler,是在容器启动时加载到内存中,其余也没有什么亮点了。
疯狂创客圈 Java 高并发【 亿级流量聊天室实战】实战系列 【 博客园总入口 】
架构师成长+面试必备之 高并发基础书籍 【 Netty Zookeeper Redis 高并发实战 】
很多项目,都需要基于 Websocket 协议做在线客服、在线推送、在线聊天,虽然 Tomcat 内置支持 Websocket 协议,但是由于 Tomcat 的吞吐量、连接数都很低,作为测试是可以的。 在生产环境,一定需要使用高吞吐量、高连接数的 Netty 服务器进行替代 。
之所以 Netty 性能高,因为其使用的是 Reactor 反应器模式。关于反应器模式原理,请参见 《Netty Zookeeper Redis 高并发实战》 一书。
聊天过程gif 演示:
聊天示意图:
Netty搭建的服务器基本上都是差不多的写法:
绑定主线程组和工作线程组,这部分对应架构图中的事件循环组。其原理,,请参见 《Netty Zookeeper Redis 高并发实战》 一书。
重点就是ChannelInitializer的配置,以异步的方式启动,最后是结束的时候关闭线程组。
下面是用websocket做聊天室的逻辑:
源码网址: Java 高并发研习社群 【 博客园 总入口 】
疯狂创客圈 经典图书 : 《Netty Zookeeper Redis 高并发实战》 面试必备 + 面试必备 + 面试必备
欢迎分享,转载请注明来源:夏雨云
评论列表(0条)