Tigase开发笔记6：packet流转机制 -> 一条消息（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)

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方法。

ConnectionManager

ClientConnectionManager

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

下面的资料希望对你也有所帮助

网络
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具体的实现来处理。

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

tigase.server.MessageReceiver – 这个接口extends ServerComponent，所有希望接收数据packets的Component都需要实现接口中定义的方法，比如session manager和c2s connection manager。

tigase.conf.Configurable – 如果希望components可以被配置，则需要实现这个接口,所有默认定义基本都在这。在运行时，配置信息会被推送到这种类型的对象。components必须能够在运行时对变更的配置项进行处理，这一点在实现时要留神。

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

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

版权声明：本文为博主原创文章，未经博主允许不得转载。