XMPP协议之消息回执解决方案

WindySky 2017-10-26 发布于广东

展开全文

苦恼中寻找方法

在开始做即时通信时就知道了消息回执这个概念，目的是解决通讯消息因为各种原因未送达对方而提供的一种保障机制。产生这个问题的原因主要是网络不稳定、服务器或者客户端一些异常导致没有接收到消息。

因为产品中使用的是openfire和spark的组合，所以一直就想在这个范围内找一个现成的方案，只不过通过阅读一些开发者的总结提到说openfire没有消息回执的方案。于是也看到了别人的方案：

发送者发送消息给服务端
服务端接收到消息后发送回执给发送者
发送者确认收到则结束，如果未收到就重发
服务端将消息记录一下，并推送给接收者，等待接收者的回执
接收者接收消息并发回执给服务端
服务端接收回执删除掉消息回执记录，表示已经发送完毕
如果一定时间内没收到重新推送消息给客户端
接收者如果收到消息进行去重处理，如果不重复的执行第5-6步

这个流程基本就是完成了消息回执的功能，核心点就是在于发送者-服务端-接收者三者之间建立一个消息确认机制。这个方案如果要自己实现的话需要定制一套消息协议了，这个实现方法比较多，对于XMPP来说发message、iq都可以。当然也可以看到这套方案会带来问题，就是每条消息都要执行一套确认，所以会增大流量和计算量。

流量对于移动网络来说还是很重要的，而且移动网络因为移动的原因很容易出现不稳定，所以自然这部分的流量可能会更大些。但是也正因为移动网络的不稳定就更需要消息回执来确认消息状态了，解决丢包的问题。

于是这就变成了一个双向的问题，只要能是尽量减少消息的体积以此来减少流量吧。

只不过对于我来说方法有了，怎么做是个问题，毕竟要实现一套这样的功能，还要保证稳定，否则这个消息回执功能本身不稳定还不如不要呢。基本的设计思路也有了：

客户端维护两个列表（发送回执队列和接收回执队列），用于保存发送/接收消息回执情况
服务端也维护一个列表，用于记录消息回执的接收与发送情况，服务端对列表进行超时检查，如果回执未发送的重发消息，如果收到重复的消息则去重处理
客户端定期检查两个列表里的回执状态，如果未收到回执的要做重发处理，如果收到的是重复的回执则进行去重处理

方案差不多有了，只不过在检阅网上资料时有了新的发现。

柳暗花明

在看别人的总结时发现XMPP有扩展协议是支持消息回执功能的，就是XEP-0184.了解下来这个协议确实是一套消息回执的实现方法，但是呢。。

它必须在openfire3.9以上版本才支持，这个可以在openfire的版本日志里可以看到
它只是一种端到端的消息回执，而且只有接收端收到消息后才会返回回执，这样对于发送者来说很麻烦，如果接收者不在线无法得知消息是否发出了，因为服务端不会告知发送者已经拿到消息了。只有等到接收者上线获取了消息后，由接收者发送一条确认的回执给接收者

这个看起来很美好的东西，发现不大好用啊。于是看了自己的openfire是4以上版本的，所以确实支持。然后检查了客户端使用的smack包里确实有XEP-0184的实现。

//这个类是一个统一调用的类
org.jivesoftware.smackx.receipts.DeliveryReceiptManager

//这个是发送者发送一个回执请求，告知客户端我要消息回执
org.jivesoftware.smackx.receipts.DeliveryReceiptRequest

//这个是接收者收到消息后返回的回执确认
org.jivesoftware.smackx.receipts.DeliveryReceipt

//这个是用于发送者监听接收者发来回执确认的事件
public interface ReceiptReceivedListener {
    /**
     * Callback invoked when a new receipt got received.
     * <p>
     * {@code receiptId} correspondents to the message ID, which can be obtained with
     * {@link org.jivesoftware.smack.packet.Stanza#getStanzaId()}.
     * </p>
     * 
     * @param fromJid the jid that send this receipt
     * @param toJid the jid which received this receipt
     * @param receiptId the message ID of the stanza(/packet) which has been received and this receipt is for
     * @param receipt the receipt
     */
    void onReceiptReceived(String fromJid, String toJid, String receiptId, Stanza receipt);
}

有了这三个家伙确实是可以做一套消息确认的机制，但是要在客户端发送消息时发送一个DeliveryReceiptRequest，然后等待接收者发送回来的消息确认DeliveryReceipt。

public class ChatDemo {

    public static void main(String[] args) {
        AbstractXMPPConnection connection = SesseionHelper.newConn("192.168.11.111", 5222, "abc", "user1", "pwd1");
        
        //在发消息之前通过DeliveryReceiptManager订阅回执
        DeliveryReceiptManager drm = DeliveryReceiptManager.getInstanceFor(connection);
        drm.addReceiptReceivedListener(new ReceiptReceivedListener() {
            
            @Override
            public void onReceiptReceived(String fromJid, String toJid,
                    String receiptId, Stanza receipt) {
                System.err.println((new Date()).toString()+ " - drm:" + receipt.toXML());
                
            }
        });
        
        Message msg = new Message("100069@bkos");
        msg.setBody("回复我的消息1.");
        msg.setType(Type.chat);
        //将消息放到DeliveryReceiptRequest中，这样就可以在发送Message后发送回执请求
        DeliveryReceiptRequest.addTo(msg);
        
        try {
            connection.sendStanza(msg);
        } catch (NotConnectedException e) {
            e.printStackTrace();
        }
        
        connection.addAsyncStanzaListener(new StanzaListener() {
            
            @Override
            public void processPacket(Stanza packet) throws NotConnectedException {
                System.out.println((new Date()).toString()+ "- processPacket:" + packet.toXML());
            }
        }, new StanzaFilter() {
            @Override
            public boolean accept(Stanza stanza) {
                return stanza instanceof Message;
            }
        });
        
        while (true) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
        }
    }

}

上面代码是发送者要完成的代码，这里并没有看到接收者返回回执的过程，这个实现在DeliveryReceiptManager里完成的。

private DeliveryReceiptManager(XMPPConnection connection) {
    super(connection);
    ServiceDiscoveryManager sdm = ServiceDiscoveryManager.getInstanceFor(connection);
    sdm.addFeature(DeliveryReceipt.NAMESPACE);

    // Add the packet listener to handling incoming delivery receipts
    connection.addAsyncStanzaListener(new StanzaListener() {
        @Override
        public void processPacket(Stanza packet) throws NotConnectedException {
            DeliveryReceipt dr = DeliveryReceipt.from((Message) packet);
            // notify listeners of incoming receipt
            for (ReceiptReceivedListener l : receiptReceivedListeners) {
                l.onReceiptReceived(packet.getFrom(), packet.getTo(), dr.getId(), packet);
            }
        }
    }, MESSAGES_WITH_DELIVERY_RECEIPT);

    // Add the packet listener to handle incoming delivery receipt requests
    connection.addAsyncStanzaListener(new StanzaListener() {
        @Override
        public void processPacket(Stanza packet) throws NotConnectedException {
            final String from = packet.getFrom();
            final XMPPConnection connection = connection();
            switch (autoReceiptMode) {
            case disabled:
                return;
            case ifIsSubscribed:
                if (!Roster.getInstanceFor(connection).isSubscribedToMyPresence(from)) {
                    return;
                }
                break;
            case always:
                break;
            }

            final Message messageWithReceiptRequest = (Message) packet;
            Message ack = receiptMessageFor(messageWithReceiptRequest);
            if (ack == null) {
                LOGGER.warning("Received message stanza with receipt request from '" + from
                                + "' without a stanza ID set. Message: " + messageWithReceiptRequest);
                return;
            }
            connection.sendStanza(ack);
        }
    }, MESSAGES_WITH_DEVLIERY_RECEIPT_REQUEST);
}

DeliveryReceiptManager里会订阅消息事件，当收到消息是需要回执时发送ack包，这里的ack就是带了DeliveryReceipt的一个消息包。

好了，这个XEP-0184差不多看明白了，但并不是想要的那种消息回执。它更像是手机消息或者邮件的那种接收确认回执。是端到端的一种确认机制。但是如果在服务端对这个消息做一些截取处理，做一个中间状态也是可以达到我们要的消息回执的状态的。

做法就是在服务端截取XEP-0184的消息，如果是请求消息DeliveryReceiptRequest则在服务端保存记录，同时服务端发送DeliveryReceipt（ack）给发送方。然后客户端照样接收消息返回ack后服务端截获更新服务端记录即可。

这种做法就是借用xep-0184协议来完成消息回执的功能。

真正的又一村

也不知道是否意外，在看一篇博文时发现了一个更有意思东西，就是XEP-0198.

它是干啥的呢？

流管理背后的基本概念是,初始化的实体(一个服务端或者客户端)和接收的实体(一个服务端)可以为更灵活的管理stream交换命令.下面两条流管理的特性被广泛的关注,因为它们可以提高网络的可靠性和终端用户的体验:

Stanza确认(Stanza Acknowledgements) – 能够确认一段或者一系列Stanza是否已被某一方接收.
流恢复(Stream Resumption) – 能够迅速的恢复(resume)一个已经被终止的流.

这就突然发现又一村原来在这啊，XMPP毕竟最开始是基于TCP协议的，可以在流的基础上完成消息到达回执。它的特征也表明了这点，一是可以做消息确认，保证消息是否被另一方接收。另外一点就是在消息未确认接收时可以做恢复（也就是重试）。这不就完全满足我们消息回执的要求了吗？

它的工作过程是：一端发起请求,另一端必须以应答。

只不过在smack要4.1.x以上版本，而且默认是不开启流管理功能的，所以要手动的开启一下，剩下的事情由smack和openfire来完成。在建立TCPConnection前执行正面这句：

XMPPTCPConnection.setUseStreamManagementResumptionDefault(true);

这个代码就是说开启流恢复，当然流恢复开启了Stanza确认也是要开启的，可以看setUseStreamManagementResumptionDefault的实现，里面调用setUseStreamManagementDefault：

public static void setUseStreamManagementResumptionDefault(boolean useSmResumptionDefault) {
    if (useSmResumptionDefault) {
        // Also enable SM is resumption is enabled
        setUseStreamManagementDefault(useSmResumptionDefault);
    }
    XMPPTCPConnection.useSmResumptionDefault = useSmResumptionDefault;
}

openfire服务端默认是开启这个功能的，在openfire.xml里有设置：

  <!-- XEP-0198 properties -->  
  <stream> 
    <management> 
      <!-- Whether stream management is offered to clients by server. -->  
      <active>true</active>  
      <!-- Number of stanzas sent to client before a stream management
                 acknowledgement request is made. -->  
      <requestFrequency>5</requestFrequency> 
    </management> 
  </stream>

好了，这样就完成了消息回执的功能了。没想到XMPP协议已经支持了整个流程，省去了很多事情，同时openfire中websocket也是支持xep-198，所以手机端应该也是可以支持。