Openfire的启动过程与session管理

说明

本文源码基于Openfire4.0.2。

Openfire的启动

    Openfire的启动过程非常的简单,通过一个入口初始化lib目录下的openfire.jar包,并启动一个XMPPServer实例。

下面就是ServerStarter.start方法的代码片断:

Class containerClass = loader.loadClass("org.jivesoftware.openfire.XMPPServer");
containerClass.newInstance();

这样一个openfire实例就已经启动了。

XMPPServer类

这个XmppServer类是单实例的对象,这样在服务器调用时可以获取一个实例。既然是个对象就会有构造的过程,XMPPServer在构造过程中会对服务进行初始化,这个过程包括:

  • 初始化配置参数
  • 检查是否需要安装
  • 初始化Module
  • 启动统计模块
  • 启动plugin

基本就是这么简单,还是非常简洁明了。这里也可以大概知道在openfire里主要是module和plugin两类模块,一般情况下内部的模块都用module,对于一些功能的扩展或者第三方的开发扩展使用Plugin。官方其实也会自己写一个插件来扩展功能,说明插件还是比较灵活的。

提一提Module的加载过程

下面代码是module的加载过程

if (!setupMode) {
    verifyDataSource();
    // First load all the modules so that modules may access other modules while
    // being initialized
    loadModules();
    // Initize all the modules
    initModules();
    // Start all the modules
    startModules();
}

可以看到,分了三个步骤:

加载模块:是对模块类的实例化过程,就是创建对象

初始化模块:就是调用module.initialize(this);,其实就是调用模块的初始化方法

启动模块:module.start();,同理就是调用启动模块

这是因为openfire规范了module的接口抽象Module,所有的模块都要按照这个规范实现,看代码:

public interface Module {
 
    /**
     * Returns the name of the module for display in administration interfaces.
     *
     * @return The name of the module.
     */
    String getName();
 
    /**
     * Initialize the module with the container.
     * Modules may be initialized and never started, so modules
     * should be prepared for a call to destroy() to follow initialize().
     *
     * @param server the server hosting this module.
     */
    void initialize(XMPPServer server);
 
    /**
     * Start the module (must return quickly). Any long running
     * operations should spawn a thread and allow the method to return
     * immediately.
     */
    void start();
 
    /**
     * Stop the module. The module should attempt to free up threads
     * and prepare for either another call to initialize (reconfigure the module)
     * or for destruction.
     */
    void stop();
 
    /**
     * Module should free all resources and prepare for deallocation.
     */
    void destroy();
}

这也标示了Module的生命周期,Openfire会管理这些Module的生命周期,以此来保证各个模块的启动与释放。

Connection管理模块

整个启动过程有点奇怪,并没有看到Openfire是如何监听端口的,如果不监听如何获利客户端连接呢?因为Openfire只通过Module来管理的,那么对应的网络管理应该就在Module中。于是在XMPPServer.loadModules方法中看到下面的代码:

// Load this module always last since we don't want to start listening for clients
// before the rest of the modules have been started
loadModule(ConnectionManagerImpl.class.getName());

ConnectionManagerImpl就是连接的管理模块,这里有个注释,就是在其他模块启动后之后再启动监听模块。

在ConnectionManagerImpl中管理了主要的连接,都是以ConnectionListener的来管理,这个类用于包装连接。我的理解就是一个连接抽象吧,这样对于代码来说写起来比较统一。看下面代码中Manager管理着哪些:

  private final ConnectionListener clientListener;
    private final ConnectionListener clientSslListener;
    private final ConnectionListener boshListener;
    private final ConnectionListener boshSslListener;
    private final ConnectionListener serverListener;
    private final ConnectionListener componentListener;
    private final ConnectionListener componentSslListener;
    private final ConnectionListener connectionManagerListener; // Also known as 'multiplexer'
    private final ConnectionListener connectionManagerSslListener; // Also known as 'multiplexer'
    private final ConnectionListener webAdminListener;
    private final ConnectionListener webAdminSslListener;

这里面除了server只有一个外,其他的都是两个,其中一个是SSL的。它们主要是什么链接?

  • client:表示客户端连接
  • bosh:就是HTTP绑定的连接
  • server:服务器到服务器的socket连接
  • component:组件到服务器的连接
  • connectionManager:是指通过connectionManager连接器过来的连接
  • webAdmin:是指web控制台的连接

这里面bosh和webAdmin使用的是http协议,所以连接并不是长连接,其他的都是socket。

openfire里使用了Mina来实现socket网络处理。只不过看代码中对于S2S类型的连接使用的不是mina,如下代码:

if ( getType() == ConnectionType.SOCKET_S2S )
{
    connectionAcceptor = new LegacyConnectionAcceptor( generateConnectionConfiguration() );
}
else
{
    connectionAcceptor = new MINAConnectionAcceptor( generateConnectionConfiguration() );
}

LegacyConnectionAcceptor是个废弃的类,但不知道为什么s2s还要用这个呢?看了看实现,LegacyConnectionAcceptor就是起了一个线程,在线程里建了一个ServerSocket。可能以后还是会迁移这部分代码吧。

在connectionAcceptor中会根据类型创建一个ConnectionHandler用于实现具体的业务功能,而ConnectionHandler都是基于org.apache.mina.core.service.IoHandlerAdapter派生的类,而IoHandlerAdapter又是IoHandler的适配接口,所以实质上就是IoHandler。下面是类继承关系:

在这些Handler里完成的主要是每个连接打开、关闭和数据收发等操作的处理。而其中比较关键的一个步骤就是在sessionOpened中设置了StanzeHandler,而每种ConnectionHandler都有自己的StanzeHandler实现。以ClientConnectionHandler为例子,其中ClientConnectionHandler复写了父类的createStanzaHandler方法,这里面

@Override
    StanzaHandler createStanzaHandler(NIOConnection connection) {
        return new ClientStanzaHandler(XMPPServer.getInstance().getPacketRouter(), connection);
    }

这里使用的是clientStanzaHandler,表示是客户端的数据节处理者。而最终的createStanzaHandler调用是在父类ConnectionHandler的sessionOpened完成的,

@Override
public void sessionOpened(IoSession session) throws Exception {
    // Create a new XML parser for the new connection. The parser will be used by the XMPPDecoder filter.
    final XMLLightweightParser parser = new XMLLightweightParser(StandardCharsets.UTF_8);
    session.setAttribute(XML_PARSER, parser);
    // Create a new NIOConnection for the new session
    final NIOConnection connection = createNIOConnection(session);
    session.setAttribute(CONNECTION, connection);
    session.setAttribute(HANDLER, createStanzaHandler(connection));
    // Set the max time a connection can be idle before closing it. This amount of seconds
    // is divided in two, as Openfire will ping idle clients first (at 50% of the max idle time)
    // before disconnecting them (at 100% of the max idle time). This prevents Openfire from
    // removing connections without warning.
    final int idleTime = getMaxIdleTime() / 2;
    if (idleTime > 0) {
        session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime);
    }
}

这样每一个session在打开时都会设置handler,而具体的handler由各个派生类创建返回。这里的StanzHandler就是Openfire里的数据包处理单元。和connection类型一样,包处理也是对应的几个类:

注:

关于openfire与mina的关系可以看看下面的文章,但是版本相对比较老些,代码有些不同,只不过思路差不多:

http://blog.csdn.net/huwenfeng_2011/article/details/43413009

Session模块

对于Openfire来说一个比较重要的功能就是管理session,因为要与客户端实时的进行数据通讯,所以必须保持着连接。在Openfire中对于Session的管理都集中在SessionManager模块。但在前面说到连接管理时已经知道了IoSession的创建过程,但并没有看到openfire是如何管理它的。接着ConnectionHandler和StanzaHandler就能知道其中有奥秘。

前面知道了ConnectionHandler是连接的处理者,这里会有连接的创建、关闭、数据收发的处理,回到ConnectionHandler这个抽象类中。对于创建时(sessionOpend)主要是创建了StanzaHandler,这样就把数据包的处理委托给了StzanzHandler(派生类)。但是这个时候并没有将session放入到openfire的session管理模块中,而是在客户端发送数据过来后才开始的。

先看看ConnectionHandler的messageReceived方法:

@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    // Get the stanza handler for this session
    StanzaHandler handler = (StanzaHandler) session.getAttribute(HANDLER);
    // Get the parser to use to process stanza. For optimization there is going
    // to be a parser for each running thread. Each Filter will be executed
    // by the Executor placed as the first Filter. So we can have a parser associated
    // to each Thread
    final XMPPPacketReader parser = PARSER_CACHE.get();
    // Update counter of read btyes
    updateReadBytesCounter(session);
    //System.out.println("RCVD: " + message);
    // Let the stanza handler process the received stanza
    try {
        handler.process((String) message, parser);
    } catch (Exception e) {
        Log.error("Closing connection due to error while processing message: " + message, e);
        final Connection connection = (Connection) session.getAttribute(CONNECTION);
        if ( connection != null ) {
            connection.close();
        }
 
    }
}

在接收到数据包后获取到StanzaHandler,然后调用了它的process方法,也就是让实际的包处理者去处理数据。这样就回到了StanzeHanler,以ClientStanzaHandler为例子。只不过这个派生类中没有重写process方法,也就是说要看父类的实现:

public void process(String stanza, XMPPPacketReader reader) throws Exception {
 
    boolean initialStream = stanza.startsWith("<stream:stream") || stanza.startsWith("<flash:stream");
    if (!sessionCreated || initialStream) {
        if (!initialStream) {
..........
        // Found an stream:stream tag...
        if (!sessionCreated) {
            sessionCreated = true;
            MXParser parser = reader.getXPPParser();
            parser.setInput(new StringReader(stanza));
            createSession(parser);
        }
..........
        return;
    }
..........
}

由于代码较多,我省略了一些代码。看到这应该明白了吧,对于当前的连接没有创建Openfire的session对象时,会进行创建过程createSession,对于不同的StanzeHandler会有些不一样,这里ClientStanzaHandler的实现就是把创建好的session放到本地的LocalClientSession中:

@Override
boolean createSession(String namespace, String serverName, XmlPullParser xpp, Connection connection)
        throws XmlPullParserException {
    if ("jabber:client".equals(namespace)) {
        // The connected client is a regular client so create a ClientSession
        session = LocalClientSession.createSession(serverName, xpp, connection);
        return true;
    }
    return false;
}

到这一个session算是建立完成了。

集群下的session

之前一篇关于《Openfire集群源码分析》提到了session的一些内容。其中也提到了session是不会向每一台服务器进行同步复制的,这就有一个问题,如果A用户先是连接了服务器1,但是接下来的操作又到服务器2,这不就会造成session无法找到吗?同样的问题,如果想要获取到当前所有的client session怎么办?

1、如何在集群中发消息

对于消息最终还是通过session来发送的,前后代码太多,就直接看一下sessionManager中的getSession方法吧。

public ClientSession getSession(JID from) {
    // Return null if the JID is null or belongs to a foreign server. If the server is
    // shutting down then serverName will be null so answer null too in this case.
    if (from == null || serverName == null || !serverName.equals(from.getDomain())) {
        return null;
    }
 
    // Initially Check preAuthenticated Sessions
    if (from.getResource() != null) {
        ClientSession session = localSessionManager.getPreAuthenticatedSessions().get(from.getResource());
        if (session != null) {
            return session;
        }
    }
 
    if (from.getResource() == null || from.getNode() == null) {
        return null;
    }
 
    return routingTable.getClientRoute(from);
}

先是获取本地的session,如果能找到直接返回,找不到则跳到routingTable里获取客户端的路由信息。

@Override
public ClientSession getClientRoute(JID jid) {
    // Check if this session is hosted by this cluster node
    ClientSession session = (ClientSession) localRoutingTable.getRoute(jid.toString());
    if (session == null) {
        // The session is not in this JVM so assume remote
        RemoteSessionLocator locator = server.getRemoteSessionLocator();
        if (locator != null) {
            // Check if the session is hosted by other cluster node
            ClientRoute route = usersCache.get(jid.toString());
            if (route == null) {
                route = anonymousUsersCache.get(jid.toString());
            }
            if (route != null) {
                session = locator.getClientSession(route.getNodeID().toByteArray(), jid);
            }
        }
    }
    return session;
}

这里更直接的可以看到,查找本地路由不null则会通过RemoteSessionLocator来完成。当然这里最大的奥秘其实是usersCache和anonymousUsersCache这两个cache。之前写的集群源码分析中提过,最终openfire集群后会对缓存进行同步,这样每台服务器上都会有缓存的副本。所以usersCache是拥有所有用户信息的,有了user的信息就有了jid的信息,这样不管是哪台服务器都可以对数据包处理并发送给客户端。

这里的RemoteSessionLocator是由于适配不同的集群组件所抽象的接口,使得加入不同集群组件提供了透明处理。

2、如何获利所有的在线用户

对于获取所有在线用户这个功能思路也挺简单,一样是找本地所有的缓存。看getSessions的代码:

public Collection<ClientSession> getSessions() {
        return routingTable.getClientsRoutes(false);
    }

其实就是访问路由表,因为路由表里有所有的cache,和获取单个的session不一样,需要对所有的路由都遍历返回。

@Override
public Collection<ClientSession> getClientsRoutes(boolean onlyLocal) {
    // Add sessions hosted by this cluster node
    Collection<ClientSession> sessions = new ArrayList<ClientSession>(localRoutingTable.getClientRoutes());
    if (!onlyLocal) {
        // Add sessions not hosted by this JVM
        RemoteSessionLocator locator = server.getRemoteSessionLocator();
        if (locator != null) {
            // Add sessions of non-anonymous users hosted by other cluster nodes
            for (Map.Entry<String, ClientRoute> entry : usersCache.entrySet()) {
                ClientRoute route = entry.getValue();
                if (!server.getNodeID().equals(route.getNodeID())) {
                    sessions.add(locator.getClientSession(route.getNodeID().toByteArray(), new JID(entry.getKey())));
                }
            }
            // Add sessions of anonymous users hosted by other cluster nodes
            for (Map.Entry<String, ClientRoute> entry : anonymousUsersCache.entrySet()) {
                ClientRoute route = entry.getValue();
                if (!server.getNodeID().equals(route.getNodeID())) {
                    sessions.add(locator.getClientSession(route.getNodeID().toByteArray(), new JID(entry.getKey())));
                }
            }
        }
    }
    return sessions;
}

总结

对于查看Openfire的源代码学习了一些东西,特别是对于服务化系统的开发思路。而且在集群化上也有了一些认识,知道了多机部署后系统应该要解决哪些问题。

继续学习吧。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Java技术分享

Redis特性和应用场景

Redis特性 速度快 Redis使用标准C编写实现,而且将所有数据加载到内存中,所以速度非常快。官方提供的数据表明,在一个普通的Linux机器上,Redis读...

1K7
来自专栏PPV课数据科学社区

python多线程编程(1): python对多线程的支持

前面介绍过多线程的基本概念,理解了这些基本概念,掌握python多线程编程就比较容易了。 在开始之前,首先要了解一下python对多线程的支持。 虚拟机层面 P...

37215
来自专栏后端技术探索

利用 PHP 名称空间编写可读且可维护的代码

“Conan 是我榜样。” 如果我在餐桌上说这句话,我儿子会以为我说的是游戏 “野蛮人柯南”,而我妻子会以为我说的是脱口秀主持人 Conan O'Brien。这...

1322
来自专栏非著名程序员

你真的会用Android中Strings资源吗

Android为了帮助开发者把应用更方便发布给全球不同语言的人们使用,建议开发者在进行开发时不要把UI呈现相关的文本内容硬编码,而是把内容写入到strings....

2509
来自专栏用户2442861的专栏

Makefile经典教程(掌握这些足够)

makefile很重要       什么是makefile?或许很多Winodws的程序员都不知道这个东西,因为那些Windows的IDE都为你做了这个...

1361
来自专栏栗霖积跬步之旅

java多线程编程核心技术——第一章总结 1.5sleep()方法

1.1进程、多线程的概念及线程的优点 进程的概念: 百度百科讲解: 进程(Process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和...

20510
来自专栏大内老A

依赖注入[3]: 依赖注入模式

IoC主要体现了这样一种设计思想:通过将一组通用流程的控制权从应用转移到框架中以实现对流程的复用,并按照“好莱坞法则”实现应用程序的代码与框架之间的交互。我们可...

1453
来自专栏腾讯IVWEB团队的专栏

nodejs 中错误捕获的一些最佳实践

本文为翻译文章,原文比较长,感觉也有点啰嗦,所以根据个人理解猜测梳理出本文。

6270
来自专栏逸鹏说道

Python3 与 C# 并发编程之~ 进程实战篇

之前说过 Queue:在 Process之间使用没问题,用到 Pool,就使用 Manager().xxx, Value和 Array,就不太一样了:

1684
来自专栏java 成神之路

并发编程基础知识点

3536

扫码关注云+社区

领取腾讯云代金券