美文网首页
Tigase--packet流转机制

Tigase--packet流转机制

作者: zelo | 来源:发表于2017-11-09 09:17 被阅读81次

    初看Tigase的packet内部流转机制。Tigase通过tigase.io包当中的代码读取网络中的字节数组,然后通过tigase.net包当中的类把字节数组转换为字符,最后通过tigase.xml包当中的XML解析器把这些字符转换成XML DOM对象。

    图片流程说明

    看tigase源码你会发现所有的tigase处理都是基于多线程,每个component都有自己的in和out处理线程,线程间的数据传输通过queue
    总的流程大致就是:


    Tigase的packet内部流转机制

    文字流程说明

    下面是请求和响应的步骤说明(只列出了关键的步骤)。

    A. 从client到server的过程(请求-Request)

    ClientConnectionManager和MessageRouter都间接或直接继承了AbstractMessageReceiver。

    tigase.server.AbstractMessageReceiver

    – 它已经实现了四个接口:ServerComponent,MessageReceiver,Configurable和StatisticsContainer。

    它通过自己的多个线程来管理内部数据队列,且能避免死锁。

    它使用事件驱动的方式来处理数据,当packet被发送到AbstractMessageReceiver实例的abstract void processPacket(Packet packet)方法时,就立即启动了packet的处理工作。当然你还是需要实现抽象类当中的抽象方法,如果你还希望输出packet数据(例如当它收到请求时还需要发送响应),可以调用boolean addOutPacket(Packet packet)方法。

    a. XMPPIOService负责接收客户端报文并转换为对应的packet放入队列receivedPackets(相当于in_queues)中,而SocketThread负责创建ReadThread和WriteThread线程,

    ResultsListener则是SocketThread的一个内部类。ResultsListener负责调度socketReadThread()和
    socketWriteThread(), Client2Server读取和写入数据包的IO操作主要由XMPPIOService来完XMPPIOService实例则在read和write线程中被使用。

    b. ClientConnectionManager则负责从XMPPIOService获取receivedPackets,并对收到的报文进行处理,并将处理后的报文放入MessageRouter的out_queues队列中。

    c.MessageRouter重写了AbstractMessageReceiver的processPacket方法,在该方法中MessageRouter通过packet.getTo()得到组件的名称并转发packet到该组件的in_queues队列中,该目标服务组件(如ses-man,即SessionManager)可能在本机服务器上,也可能在该域(domain)集群的其他服务器上运行MessageRouter的查找顺序是先查本机服务器,找不到的话再去集群中查找;如果找到则将packet转发给目标组件处理,如果没找到则返回没有找到目标组件的错误消息。

    d. 我们的组件一般都有自己的in_queues和out_queues. 当前组件的in_queues的数据来自于上一组件的out_queues。

    int  queueIdx = Math.abs(hashCodeForPacket(packet) % in_queues_size*);
    boolean  result = in_queues.get(queueIdx).offer(packet, packet.getPriority().ordinal());
    

    //这里的packet数据来自于packetFrom="xxx",即传给当前组件的上一组件。

    B.从Service到Client的过程(响应-Response)

    服务端响应的数据也是放到out_queues中的,各组件的对应的线程会对out_queues中的packet的to属性做解析,并将消息转发到指定目标。其实消息的流转传递机制实现的核心就是packet的 from包含packetFrom和stanzaFrom) 和 to(packetTo和stanzaTo)属性,路由的路径会默认先取packetFrom或packetTo,其次再去取stanzaFrom和stanzaTo。

    客户端发送一条ping命令

    <iq type='get' id='purplee4ad721'>
    <ping xmlns='urn:xmpp:ping'/>
    </iq>
    

    IOService.java

    public abstract class IOService<RefObject>
                implements Callable<IOService<?>>, TLSEventHandler,
                IOListener {
    
    /**接收报文后call->处理客户端报文数据信息*/
    @Override
    public IOService<?> call() throws IOException {
       writeData(null);
    
       boolean readLock = true;
    
       if (stopping) {
          stop();
       } else {
          readLock = readInProgress.tryLock();
          if (readLock) {
             try {
                processSocketData();
                if ((receivedPackets() > 0) && (serviceListener != null)) {
                   serviceListener.packetsReady(this);
                }    // end of if (receivedPackets.size() > 0)
             } finally {
                readInProgress.unlock();
                if (!isConnected()) {
                   // added to sooner detect disconnection of peer - ie. client
                   if (log.isLoggable(Level.FINEST)) {
                      log.log(Level.FINEST, "{0}, stopping connection due to the fact that it was disconnected, forceStop()", toString());
                   }
                   forceStop();
                }
             }
          }
       }
    
       return readLock
             ? this
             : null;
    }
    
    /**
     * Describe
     * <code>writeData</code> method here.
     * 这里是最后响应消息给客户端的方法,写入TCP连接中的Socket
     * @param data a
     * <code>String</code> value
     */
    protected void writeData(final String data) {
    

    返回消息前还调用了下面两个类的对应write方法

    SocketIO.java
    @Override
    public int write(final ByteBuffer buff) throws IOException {
    }
    
    
    TLSIO.java
    public class TLSIO implements IOInterface {
             private int writeBuff(ByteBuffer buff) throws IOException {
              }
    }
    

    IOService打包好了消息之后形成响应的packet包,->然后再执行到ConnectionManager的writePacketToSocket()方法->再执行到ClientConnectionManager的processPacket方法。

    至此本次消息的请求和响应结束!server会继续通过AbstractMessageReceiver获取下一条消息进行处理,如此循环。。。


    1. tigase.server.ServerComponent – 这是一个非常基本的component接口。所有的component都必须实现接口中定义的方法。

    2. tigase.server.MessageReceiver – 这个接口extends ServerComponent,所有希望接收数据packets的Component都需要实现接口中定义的方法,比如session manager和c2s connection manager。
      tigase.conf.Configurable – 如果希望components可以被配置,则需要实现这个接口,所有默认定义基本都在这。在运行时,配置信息会被推送到这种类型的对象。components必须能够在运行时对变更的配置项进行处理,这一点在实现时要留神。

    3. tigase.disco.XMPPService – 实现了这个对象的类可以对“ServiceDiscovery”请求做出响应。

    4. tigase.stats.StatisticsContainer – 实现了这个对象的类可以返回运行时的统计信息。任何一个对象都可以实现这个接口用来收集统计信息

    5. tigase.server.AbstractMessageReceiver – 它已经实现了四个接口:ServerComponent,MessageReceiver,Configurable和StatisticsContainer。它通过自己的多个线程来管理内部数据队列,且能避免死锁。
      它使用事件驱动的方式来处理数据,当packet被发送到AbstractMessageReceiver实例的abstract void processPacket(Packet packet)方法时,就立即启动了packet的处理工作。当然你还是需要实现抽象类当中的抽象方法,如果你还希望输出packet数据(例如当它收到请求时还需要发送响应),可以调用boolean addOutPacket(Packet packet)方法。

    6. tigase.server.ConnectionManager – 这是一个extend AbstractMessageReceiver的抽象类。正如其名,这个类专注于对连接进行管理工作。如果你的组件需要通过网络直接发送或接受数据(比如c2s connection,s2s connection 或者 连接到外部第三方jabber服务),你可以把它作为基类进行扩展。它会帮你把所有和网络有关的工作都打理好(例如io,重连,socket监听和连接握手等工作)。
      如果你extend这个类,你需要知道数据来源于哪里:如果来源于MessageRouter,那么abstract void processPacket(Packet packet)方法会被调用; 如果来源于网络连接,那么abstract Queue processSocketData(XMPPIOService serv)方法会被调用。

    网络
    
    connectionManager同时协调ConnectionOpenThread与SocketThread。
    ConnectionOpenThread脱离上述组件,属于网络层实现,操作selector。它负责Selector.open。
    IOService提供线程安全的call方法,XMPPIOService继承它,保存了连接信息,每个连接一个IOService。
    SocketThread在实例化时,会启动多个线程,同时盯住selector。负责将每个确定的IOService进行数据处理。
    实现ConnectionOpenListener接口accept方法接收SocketChannel,组装IOService,交由SocketThread处理。
    ConnectionManager用ConcurrentHashMap记录了所有的连接。
    零碎
    
    AbstractMessageReceiver.addPacket 往自己的in_queue里加数据,是阻塞的,如果满了会出事。
    AbstractMessageReceiver.addPacketNB 往自己的in_queue里加数据,非阻塞的,和上一个的区别在于,一个是put一个是offer到queue。
    AbstractMessageReceiver.addPackets 来一堆数据。
    所有in_queue里的数据,会被processPacket方法所处理。
    对应有addOutPacket。
    所有out_queue里的数据,都默认扔给parent的in_queue,没有parent就扔到自己的in_queue。
    所有in_queue的数据,都由processPacket具体的实现来处理。
    

    企业级独立部署应用:知行办公http://zx.naton.cn
    【总监】十二春秋之,3483099@qq.com
    【Master】zelo,616701261@qq.com
    【运营】运维艄公,897221533@qq.com
    【产品设计】流浪猫,364994559@qq.com
    【体验设计】兜兜,2435632247@qq.com
    【iOS】淘码小工,492395860@qq.com;iMcG33K,imcg33k@gmail.com
    【Android】人猿居士,1059604515@qq.com;思路的顿悟,1217022114@qq.com
    【java】首席工程师MR_W,feixue300@qq.com
    【测试】土镜问道,847071279@qq.com
    【数据】fox009521,42151960@qq.com
    【安全】保密,你懂的。

    相关文章

      网友评论

          本文标题:Tigase--packet流转机制

          本文链接:https://www.haomeiwen.com/subject/fgtpmxtx.html