openfire源码解读之将cache和session对象移入redis以提升性能

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

原文：http://blog.csdn.net/jinzhencs/article/details/50522322

前言:

目前我们的openfire服务器只能支撑单机2W 集群4W.(估测在线用户数已经7.8W了)

内存25G,已经去除了好友关系(我们的场景是设备控制,消息推送,无需好友关系,)

在此情况下我们需要研究出单节点25G能够支撑5W甚至更多.

并且目前感觉openfire还是有些问题的,但是无法测出来(tsung完全测不出来,500M内存在线5.6W...)

那么修改openfire哪些地方来扩容呢？

———————————————————————罪恶的分割线—————————————————————

一.cache 缓存机制

位置:org.jivesoftware.util.cache

CacheFactory:里面定义了管理控制台可以看到的所有的cache及其他的cache,其默认设置,

CacheFactoryStrategy:定义了缓存工厂的策略模式,即当没有开启集群的时候适用默认的CacheFactory,当启动集群则适用ClusterCacheFactory

场景:

为了扩大容量支撑更多用户,我们需要把几个大的cache移入redis,让redis来管理cache.

问题提出:

根据这个cache,我们可以把这个cache移植到redis里面吗？如何实现呢？后面补上

———————————————————————罪恶的分割线—————————————————————

二.建立session

1.消息接收:
package org.jivesoftware.openfire.nio[ConnectionHandler] line165:messageReceived()

2.处理包:process packet
<org.jivesoftware.openfire.net>[SocketPacketWriteHandler]line:56---process

3.上线创建session
package org.jivesoftware.openfire.nio[ConnectionHandler]-sessionOpened()

4.下线关闭session
package org.jivesoftware.openfire.nio[ConnectionHandler]-sessionClosed()

5.当一个会话某段时间内没做任何动作，则被视为idle，每过一段时间，openfire会给这些idle的会话发送一个空包，即ping。若正常的连接，会收到返回的值，不做任何事情。若是死连接，则openfire会断开连接
org.jivesoftware.openfire.nio[ConnectionHandler]-sessionIdle()
org.jivesoftware.openfire.nio[ClientConnectionHandler]-sessionIdle()

ps：package org.jivesoftware.openfire.nio[NIOConnectionHandler] deliver即是ping发出去的包的地方，若发不出去，则关闭会话，二次发包，或者存储为离线消息
（若不要离线消息，此处可以去除代码）

6.package org.jivesoftware.openfire.net - StanzaHandler - process(Element doc)

sessionOpened这个方法里面可以看到设置了StanzaHandler去处理这个session.

而StanzaHandler里面的processMessage等可以看到是交由router去rout这个packet的.

跟进,发现它的实现方法

再跟进,这就是具体实现

———————————————————————罪恶的分割线—————————————————————

研究成果：

一.如何把cache写入redis.

cache是由DefaultCacheFactory创建或者ClusterCacheFactory创建.先说说原来集群的cache创建

位置：package org.jivesoftware.openfire.plugin.util.cache;

208行createCache()方法的return new ClusteredCache(name, hazelcast.getMap(name)); 这里返回一个cache实例

解释:

hazelcast.getMap(name)：如果这个name的map不存在,则会创建,此处相当于创建一个map/获得map.意即cache存在hazelcast的 IMap 里面.

然后map.put map.get map.remove即可实现一系列操作

想法及问题：

按这样来看把cache移到redis实在是非常简单,只需要把createCache的map换成redis的map即可了.然后操作改一下put换成rpush等等.

但是问题是,session和cache是什么关系,还有其他的东西有用到cache吗？

———————————————————————罪恶的分割线—————————————————————

2016-01-16

问题:session和cache是什么关系,还有其他的东西有用到cache吗？

昨天花了一天算是把session打通了,完整的创建流程,遗留问题是cache和session各自的作用,为什么用了cache还用sesion.

解答:(先贴上后面排版)

SessionManager保存了一份session的对象

然后登录的时候

package org.jivesoftware.openfire.handler - IQAuthHandler - 190行登录

把这个session传入,之后把session做成sessionInfo存入cache

算起来两个地方保存了session，1是sessionmanager 2是cache

之后sessionManager管理的又是LocalsessionManager和

转入到这里面去拿
1.
//Auth前保存的之后看auth完了把这个清空不? （最下面的总结可以看到清楚了的！猜想正确）
getPreAuthenticatedSessions(){
}

这是声明：即这里存的就是session对象了(保存的认证Auth之前的)

private Map<String, LocalClientSession> preAuthenticatedSessions = new ConcurrentHashMap<String, LocalClientSession>();

或者
2.
routingTable.getClientRoute(from);

位于
package org.jivesoftware.openfire.spi - RoutingTableImpl - 686行

这里拿的是package org.jivesoftware.openfire.spi - LocalRoutingTable

Map<String, RoutableChannelHandler> routes = new ConcurrentHashMap<String, RoutableChannelHandler>();

这里面存的是package org.jivesoftware.openfire.spi - RoutingTableImpl-153行

public boolean addClientRoute(JID route, LocalClientSession destination) {
boolean added;
boolean available = destination.getPresence().isAvailable();
localRoutingTable.addRoute(route.toString(), destination); //这存进去的ClientSession。
**********************
usersSessions.put(route.toBareJID(), Arrays.asList(route.toString())); //// 把session数据加入UserCache和UserSession

总结：

最主要就是这一段:完成了把session加入cache和map里面

package org.jivesoftware.openfire - SessionManager - 550 行

public void addSession(LocalClientSession session) {
// Add session to the routing table (routing table will know session is not available yet)
routingTable.addClientRoute(session.getAddress(), session);
// Remove the pre-Authenticated session but remember to use the temporary ID as the key
localSessionManager.getPreAuthenticatedSessions().remove(session.getStreamID().toString());
SessionEventDispatcher.EventType event = session.getAuthToken().isAnonymous() ?
SessionEventDispatcher.EventType.anonymous_session_created :
SessionEventDispatcher.EventType.session_created;
// Fire session created event.
SessionEventDispatcher.dispatchEvent(session, event);
if (ClusterManager.isClusteringStarted()) {
// Track information about the session and share it with other cluster nodes
sessionInfoCache.put(session.getAddress().toString(), new ClientSessionInfo(session));
}
}

完整加入会话流程是：

1.在auth完成前,先加入PreAuthenticatedSessions中

2.xmpp来回几步,到auth成功前一步，客户端发来xx请求的时候，发来的请求被 IQHandler - process（packet）接收

3.然后被 IQAuthHandler - handlerIQ（IQ packet）接管，它先从sessionManager取到之前的preAuthenticatedSessions里面的客户端session，

4. 然后判断 if (JiveGlobals.getBooleanProperty("xmpp.auth.iqauth",true)) { 说明是auth请求

5.然后在preAuthenticatedSessions里面的判断session的状态 if (session.getStatus() == Session.STATUS_AUTHENTICATED)
发现还没有完成auth ，则执行login

else {
// it is an auth attempt
response = login(username, query, packet, password, session, digest);
resourceBound = session.getStatus() == Session.STATUS_AUTHENTICATED;
}

6.然后进入login方法，一系列判断后,走到最后面的 session.setAuthToken(token, resource);

7.这个方法先把preAuthenticatedSessions的状态设置为setStatus(Session.STATUS_AUTHENTICATED); 然后就执行上述最重要的一步 sessionManager.addSession(this);

8.先把新的session加入route里面， Map<String, RoutableChannelHandler> routes = new ConcurrentHashMap<String, RoutableChannelHandler>();

9.然后从preSession里面移除旧的 localSessionManager.getPreAuthenticatedSessions().remove(session.getStreamID().toString());

10.然后看集群是否启动来决定是否加入sessionInfoCache

if (ClusterManager.isClusteringStarted()) {
// Track information about the session and share it with other cluster nodes
sessionInfoCache.put(session.getAddress().toString(), new ClientSessionInfo(session));
}

11.然后创建一个返回的IQ

12.完毕

———————————————————————罪恶的分割线—————————————————————

2016-01-18:

研究成果:

今天考虑了下如何把openfire的cache和session对象保存入redis,发现redis只支持存储序列化的byte或者一些基本类型,

之前想到了序列化,但是session对象是没法序列化的,hazelcast集群插件实现同步(网络传输)都是自己手写的序列化方法实现的序列化.我就不用这种方法了.思来想去想到了一个好方法,JSON来实现序列化!

然后验证了下JSON序列化对象的可行性,发现完全可以!