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

苦恼中寻找方法

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

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

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

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

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

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

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

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

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

柳暗花明

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

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

这个看起来很美好的东西,发现不大好用啊。于是看了自己的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,所以手机端应该也是可以支持。

参考与引用

http://developerworks.github.io/2014/10/03/xmpp-xep-0198-stream-management/ http://blog.csdn.net/chszs/article/details/48576553

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏源码之家

DEDE 5.7 伪静态 完美版

2058
来自专栏互联网杂技

HTTP,HTTP2.0,SPDY,HTTPS你应该知道的一些事

作为一个经常和web打交道的程序员,了解这些协议是必须的,本文就向大家介绍一下这些协议的区别和基本概念,文中可能不局限于前端知识,还包括一些运维,协议方面的知识...

51511
来自专栏漏斗社区

工具| 诸神之眼之邮件发送nmap的扫描报告

VPS上的Nmap还在扫描工作着,而你已经把它忘了忙着其他事情。突然,一封邮件发来,向你汇报本次扫描完毕,和具体的扫描报告。是不是觉得很方便?借助Nmap的库文...

2824
来自专栏张戈的专栏

再次扩散:Linux系统bash漏洞CVE-2014-6271仍未被彻底修复,红帽再发补丁

昨天发文扩散这次 Linux 严重漏洞的检测和修复方法。 今天上午,RedHat 官网再发新补丁: ? 详见:https://rhn.redhat.com/er...

37010
来自专栏农夫安全

【原创】中间件测试方法总结---(3)

0x01 应用测试 对于类似4A、BOP这样的应用,需要进行记录,并按照常规的渗透办法进行安全测试。【必要时候,需要进行登录测试】 其他应用测试 发现办法 协议...

4597
来自专栏公有云大数据平台弹性 MapReduce

Presto内存调优及原理(基础篇)

Presto是一个开源的分布式SQL查询引擎,适用于交互式分析查询,数据量支持GB到PB字节。

1.6K8
来自专栏IT技术精选文摘

Presto内存调优及原理(基础篇)

Presto是一个开源的分布式SQL查询引擎,适用于交互式分析查询,数据量支持GB到PB字节。Presto支持在线数据查询,包括Hive, Cassandra,...

1515
来自专栏Hadoop数据仓库

HAWQ技术解析(五) —— 连接管理

        服务器启动后,还要经过一系列配置,才能被客户端程序所连接。本篇说明如何配置客户端身份认证,HAWQ的权限管理机制,HAWQ最常用的命令行客户端工...

2539
来自专栏前端侠2.0

ng6中,在HTTP拦截器里,异步请求数据,之后再返回拦截器继续执行用户请求的方法研究

      我现在项目就是利用拦截器,在请求头里增加:'Authorization': this.storage.token 的请求头。

1862
来自专栏大魏分享(微信公众号:david-share)

NGINX的定制化 | API Management学习第四篇

3 SCALE中API gateway,是基于NGINX(OpenResty Web Platform = Nginx + Lua )。

1672

扫码关注云+社区

领取腾讯云代金券