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序列化对象的可行性,发现完全可以!

可以看到,redis已经成功存入了.

并且被序列化的Man不仅仅是个JavaBean,我还给予了它一些方法,以及私有参数引入其他的对象.

都完全没问题.所以初步判定cache和session通过JSON方式存储读取是完全可行的.

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

2016-01-19

今早上思考了下,存储整个DefaultCache对象代价太高了,那样效率估计比原来还低.因为每次取都需要把整个对象取出然后得到它的map,进而得到相应的键值.

于是决定只存储Defaultcache里面的Map,这个才是主要影响用户量及性能的.并且能和redis的map无缝对接.

如图:

HMSET CacheName JID SessionInfoOrCacheInfo 是完全OK的.

现在的问题在于DefaultCache里面的map

/**
     * The map the keys and values are stored in.
     */
    protected Map<K, DefaultCache.CacheObject<V>> map;

第一个参数是泛型,第二个是cacheObject.

第二个比较好解决,就是个JavaBean.

现在难点在于第一个,接下来攻克第一个泛型JSON序列化的问题！Cache就算打通了.之后再搞session.