一个S2S通信中的同步、异步问题的解决

最近在搞Openfire S2S时遇到了一个数据通信时的同步与异步问题。具体描述如下：

OF中的基本通信结构如下：

客户端与服务器的通信是同步的，服务器与服务器之间的通信却是异步的：客户端发送一个请求后就会阻塞一段时间，等待服务器的响应；服务器之间则采用消息通知的方式异步交流数据。

所以会出现下面这种情况：

客户端发送消息给服务器A，消息经入口路由到相应的功能模块进行处理，而此时需要从B服务器中获取一些额外的数据，比如验证发送上来的账号是否在另外的服务器有注册。一般情况下数据在本地的话就直接查找处理再给用户返回就行，但在其它服务器时直接返回就会出现结果错误或不完整，不返回则造成客户端超时。

处理这种情况的方法就是：接收到客户端的请求后，服务器A发送消息给B查询数据，同时将客户端的请求阻塞一个小的时间段（保证客户端不超时），等B返回数据后再继续执行下面的逻辑。（看了OF代码才明白了）

以查询是否为注册用户为例，OF是如下实现的：

UserManager.java
public boolean isRegisteredUser(JID user) {
        XMPPServer server = XMPPServer.getInstance();
        if (server.isLocal(user)) {
            //要查找的域为本地......
        }
        else {
            //先按全JID的形式查找本地cache
            Boolean isRegistered = remoteUsersCache.get(user.toString());
 if (isRegistered == null) {
 // Check if the bare JID of the user is cached
 isRegistered = remoteUsersCache.get(user.toBareJID());
 if (isRegistered == null) {
 // 本地cache没有的情况下发送IQ消息去查询，接收方为对应的用户，这个接收方改为其它服务器就成了S2S了。
 IQ iq = new IQ(IQ.Type.get);
 iq.setFrom(server.getServerInfo().getXMPPDomain());
 iq.setTo(user.toBareJID());
 iq.setChildElement("query", "http://jabber.org/protocol/disco#info");
 // Send the disco#info request to the remote server. The reply will be
 // processed by the IQResultListener (interface that this class implements)
                    //注册一个接受处理结果消息的Listener
 server.getIQRouter().addIQResultListener(iq.getID(), this);
                    //这里进行了阻塞等待一段时间，等待对端的回应将结果放到cache中
 synchronized (user.toBareJID().intern()) {
 server.getIQRouter().route(iq);
 // Wait for the reply to be processed. Time out in 1 minute.
 try {
 user.toBareJID().intern().wait(60000);
 }
 catch (InterruptedException e) {
 // Do nothing
 }
 }
 // 再去cache中查找结果
 isRegistered = remoteUsersCache.get(user.toBareJID());
 if (isRegistered == null) {
 // Disco failed for some reason (i.e. we timed out before getting a result)
 // so assume that user is not anonymous and cache result
 isRegistered = Boolean.FALSE;
 remoteUsersCache.put(user.toString(), isRegistered);
 }
 }
 }
 return isRegistered;
          }
//处理接收查询结果消息，将结果放入cache
public void receivedAnswer(IQ packet) {
 JID from = packet.getFrom();
 // Assume that the user is not a registered user
 Boolean isRegistered = Boolean.FALSE;
 // Analyze the disco result packet
 if (IQ.Type.result == packet.getType()) {
 Element child = packet.getChildElement();
 if (child != null) {
 for (Iterator it=child.elementIterator("identity"); it.hasNext();) {
 Element identity = (Element) it.next();
 String accountType = identity.attributeValue("type");
 if ("registered".equals(accountType) || "admin".equals(accountType)) {
 isRegistered = Boolean.TRUE;
 break;
 }
 }
 }
 }
 // Update cache of remote registered users
 remoteUsersCache.put(from.toBareJID(), isRegistered);
 // Wake up waiting thread
 synchronized (from.toBareJID().intern()) {
 from.toBareJID().intern().notifyAll();
 }
 }

 

 

再来看看外层路由是如何捕获查询结果消息，并交给相应的业务逻辑处理的：

IQRouter.java
private void handle(IQ packet) {
  ......
 if (packet.getID() != null && (IQ.Type.result == packet.getType() || IQ.Type.error == packet.getType())) {
            // The server got an answer to an IQ packet that was sent from the server
            IQResultListener iqResultListener = resultListeners.remove(packet.getID());
            if (iqResultListener != null) {
                resultTimeout.remove(packet.getID());
                if (iqResultListener != null) {
                    try {
                        iqResultListener.receivedAnswer(packet);
                    }
                    catch (Exception e) {
                        Log.error("Error processing answer of remote entity", e);
                    }
                    return;
                }
            }
        }
  }
  ......

结合上面的那段代码，就可以看出OF中是如何实现这个功能的了。

 

但这种做法的缺点也是很明显的：

1.需要原进程阻塞一个固定时间

2.需要每个涉及到同步、异步问题的地方都有一个接收结果的机制

所以可以针对以上两点做些改进。