OpenFire源码学习之十：连接管理(上)

关于连接管理分为上下两部分

连接管理

在大并发环境下，连接资源 需要随着用户并发访问量的增加而增加，所以可伸缩的连接资源就是支持大访问量的关键技术。openfire系统通过增加独立部署的连接管理器程序提高并发的能力，连接管理的前端是一台负载均衡设备，它负责把用户访问分配到指定连接管理器，多台连接管理器在访问服务器。使用连接管理器后，服务器的连接池是提供给连接管理器连 接，而一台连接管理器也提供指定数量的连接给客户端访问，这样通过添加连接管理器，以适应用户访问量。

l 它是啥？

你部署的每个openfire连接管理单元，依靠处理一部分客户端连接，来提高openfire服  务器的容量。它适用于大规模的openfire安装（几千的在线用户）。

l 每个连接管理能够处理多少用户呢？

每个连接管理应该只是能够处理5000的在线用户。对于无绑定的实验性支持正在被提  高，用以更多的提高每个连接管理单元所能处理的连接数量。

l 连接管理可以被用在其他服务器上吗？

当然（理论上）。连接管理协议正在通过开放的XEP处理来提高。我们希望和其他服务 器供应商一起协作，以便全部的XMPPP/Jabber协会支持这个协 议。我们同样期望会       有其他的兼容openfire的连接管理实现。例如，一个用本地代码编写的连接管理可以在       一个特殊的平台上达到一个非常高的容量。

l openfire连接管理用的许可和openfire本身一样吗？

当然，这个模块是是在开源GPL下的dual-licensed，或者是Jive Software同意授权的一 个商业软件许可。

连接配置

Of配置Connection Manager——So easy！

第一步：服务器设置

        在服务器设置/连接管理中，将连接管理设置为启动。并设置端口号和密码。如图

第二步：配置manager.xml文件

    配置<domain>xxxx</domain>节点，这里面的内容为of服务器域名

    配置<password></password>节点，这里密码填写在控制台的密码。

第三步：启动连接管理器

       找到connection_managerin目录下的cmanager.bat(windows环境)文件点击启动。          如图：

        启动完后，在of控制台将出现如图信息：

第四步：测试连接

       使用spark登陆连接管理器，如果配置成功，控制台“客户端回会话”值将变为“1”

Connection

Connection是一个接口代表在服务器上的一个连接。该接口的主要方法：

方法名	说明
validate	验证连接,通常这样做是通过发送一个数据包之间的空格字符。
init	初始化会话连接，允许连接类与会话相关信息配置
deliver	发送数据包
deliverRawText	发送文本字节

类图：

Connection接口下面有三个实现类：NIOConnection、SocketConnection、VirtualConnection。其中VirtualConnection是个抽象接口，而它又派生两个子类：

ClientSessionConnection、HttpVirtualConnection。

NIOConnection

该类中最为常用的方法是deliver()。

代码清代：

public void deliver(Packet packet) throws UnauthorizedException {
        if (isClosed()) {
            backupDeliverer.deliver(packet);
        }
        else {
            ByteBuffer buffer = ByteBuffer.allocate(4096);
            buffer.setAutoExpand(true);

            boolean errorDelivering = false;
            try {
                XMLWriter xmlSerializer =
                        new XMLWriter(new ByteBufferWriter(buffer, encoder.get()), new  
                                                    OutputFormat());
                xmlSerializer.write(packet.getElement());
                xmlSerializer.flush();
                if (flashClient) {
                    buffer.put((byte) '');
                }
                buffer.flip();
                ioSession.write(buffer);
            }
            catch (Exception e) {
                Log.debug("NIOConnection: Error delivering packet" + "
" + this.toString(), e);
                errorDelivering = true;
            }
           。。。。。。。
        }
    }

该方法中用了Mina框架的字节缓冲类ByteBuffer。ByteBuffer是最常用的缓冲区,它提供了读写其他数据类型的方法,且信道的读写方法只接收ByteBuffer。

1.创建ByteBuffer

  ByteBufferbuffer = ByteBuffer.allocate(4096);

  以上方法将创建一个容量为4096字节的ByteBuffer，如果发现创建的缓冲区容量太小，

  唯一的选择就是重新创建一个大小合适的缓冲区。

2.设置自动扩张，of在这里设置了true

  buffer.setAutoExpand(true);

3.回绕缓冲区
   buffer.flip();
   这个方法用来将缓冲区准备为数据传出状态，执行以上方法后,输出通道会从数据的开头  

   而不是末尾开始。回绕保持缓冲区中的数据不变，只是准备写入而不是读取。

ConnectionManager

Openfire Connection Manager是Openfire服务器的扩展，它可以透明的处理大规模并发XMPP客户端对Openfire服务器的联接。根据Connection Manager主页介绍，每个Connection Manager可以处理至少五千个并发用户。Openfire服务器可以连接一个或多个Connection Manager。每个Connection Manager可以处理一个或多个支持XMPP的客户端。为Openfire配置Connection Manager后，支持XMPP的客户端仅需要连接Connection Manager，由Connection Manager负责管理对来自XMPP客户端对Openfire服务器的连接。

连接管理的端口说明：

5222	默认端口为客户XMPP。这个端口可以用于担保和无担保的连接。客户将首先使用一个未加密的连接,连接可能安全使用StartTLS。
5223	Jabbe默认SSL通行遗留的端口。这个老方法,很快被弃用,使用加密的连接就被创建。
5275	默认的XMPP端口外部组件。
5269	默认端口为服务端对服务端的 XMPP通信。
5262	默认XMPP连接多路传输端口

管理流程

1. 打开 Openfire Admin Console

2. 点击 Server 选项卡下面的 Server Settings 页面

3. 点击左侧的 Connection Managers 页面

4. 选择 Enable 单选按钮以打开 Connection Manager 监听服务

5. 设置 password 。例如， 123 。

通过以上配置，完成了开启 Openfire 监听Connection Manager 服务。下面是对连接管理的源码分析。

通过控制台页面connection-managers-settings.jsp，会调用如下两个方法来启动连接管理：

步骤一：connectionManager.enableConnectionManagerListener(true); 

         这是一个启动服务端对服务端通信的端口侦听器：

         1.设置xmpp socket有效值true.

           JiveGlobals.setProperty("xmpp.multiplex.socket.active",                                          "true");

         2.创建连接管理的监听程序createConnectionManagerListener()

private void createConnectionManagerListener() {
        // Start多路复用器soket
        if (isConnectionManagerListenerEnabled()) {
            // Create SocketAcceptor with correct number of processors
            multiplexerSocketAcceptor = buildSocketAcceptor();
            // Customize Executor that will be used by processors to process incoming stanzas
            ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("connectionManager");
            int eventThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.threads", 16);
            ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor) threadModel.getExecutor();//固定大小线程池
            eventExecutor.setCorePoolSize(eventThreads + 1);
            eventExecutor.setMaximumPoolSize(eventThreads + 1);
            eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);
            multiplexerSocketAcceptor.getDefaultConfig().setThreadModel(threadModel);
            // Add the XMPP codec filter
            multiplexerSocketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));

        }

    3.启动连接管理的监听程序    

步骤二：connectionManager.setConnectionManagerListenerPort(port);

        设置启动监听端口

 

ConnectionManagerImpl

ConnectionManager声明了五种连接：client、clientssl、external component、other server、connectionmanager，它们有各自的端口并且可禁用。ConnectionManagerImpl创建多个SocketAcceptor监听各个端口，并绑定不 同的ConnectionHandlerOpenfire连接管理配置服务端配置步骤。

创建监听的方法：

private synchronized void createListeners() {
        if (isSocketStarted || sessionManager == null || deliverer == null || router == null || serverName == null)
        {
            return;
        }
        //创建s2s通信端口监听
        createServerListener(localIPAddress);
        // 创建连接管理监听
        createConnectionManagerListener();
        // 创建外部组件监听
        createComponentListener();
        // 创建客户端监听
        createClientListeners();
        // 创建SSL监听
        createClientSSLListeners();
    }

启动监听的方法：

private synchronized void startListeners() {
        if (isSocketStarted || sessionManager == null || deliverer == null || router == null || serverName == null) {
            return;
        }

        // 检查插件是否被加载
        PluginManager pluginManager = XMPPServer.getInstance().getPluginManager();
        if (!pluginManager.isExecuted()) {
            pluginManager.addPluginManagerListener(new PluginManagerListener() {
                public void pluginsMonitored() {
                    // 停止插件监听事件
                    XMPPServer.getInstance().getPluginManager().removePluginManagerListener(this);
                    // 启动监听
                    startListeners();
                }
            });
            return;
        }

        isSocketStarted = true;

        // Setup port info
        try {
            localIPAddress = InetAddress.getLocalHost().getHostAddress();
        }
        catch (UnknownHostException e) {
            if (localIPAddress == null) {
                localIPAddress = "Unknown";
            }
        }
        // 启动s2s端口监听
        startServerListener();
        // 启动连接管理端口监听
        startConnectionManagerListener(localIPAddress);
        // 启动外部组件监听
        startComponentListener();
        // 启动客户端监听
        startClientListeners(localIPAddress);
        // 启动SSL监听
        startClientSSLListeners(localIPAddress);
        // 启动http监听
        startHTTPBindListeners();
}