Androidpn 实现 PUSH 消息推送

项目中的推送模块的原型为名为androidpn的开源项目.所以与androidpn相同,其内部使用asmack来实现xmpp协议的解析和拓展,使用MINA框架来进行多线程的socket管理。

1、当客户端安装应用后，会根据xmpp协议(这里是注册信息)通过长连接在服务器端进行注册绑定.
2、当服务器端与客户端完成注册后，会建立起相应的session（会话）,这个session是维护长连接的很重要的介质.服务器端通过此管理客户端状态.
3、当服务器需要对活动用户发送消息时，会对存活的session通过socket连接发送消息。客户端会根据xmpp协议对消息进行解析，而后最终显示给用户.


其实这个androidpn主要由 MINA+XMPP实现,下载androidpn-server-0.5.0的代码部分你会发现:
NioSocketAcceptor 是消息服务器的主类(查看使用请猛击这里),当然和使用普通socket一样，设置下通信端口5222,以及Filter和Handler.这两个概念是mina的核心实现,同样这也是消息服务器的重点实现。

首先说下Filter,在mina中filter同WEB开发中的Servlet filter，你可以把它想象成串的形式，一个filter挨着一个filter.他在mima中已经实现了很多事情,
比如：
使用LoggingFilter记录发送的数据内容。
使用ProtocolCodecFilter进行二进制内容或者POJO的对象传输。
使用CompressionFilter对数据内容进行压缩。
用SSLFilter对传输数据进行加密。

这些都是filter可以做的事情，在消息服务器中，他只是简单的指定了两个filter，ExecutorFilter用于对多线程进行管理，ProtocolCodecFilter 用于对数据指定传送格式(XML)以及编码(utf-8)。

如图，Handler处在mina架构的最下方，当过滤器执行完毕后，会交由Handler去处理相应。


Handler类主要包括如下几个方法
SESSIONCREATED
SESSIONOPENED
SESSIONCLOSED
SESSIONIDLE
EXCEPTIONCAUGHT
MESSAGERECEIVED
MESSAGESENT

这里的SESSION你可以把它看做connection，即一个与服务器端的连接。这里的Handler主要处理连接的各种状态，以及发送接收消息的处理事件。

服务器的主类,由于其使用了mina的架构设计，所以在服务端启动前需要加载一些处理的Handler,在common-end.xml配置了解码类 XmppDecoder 和 XmppEncoder.以及最重要的处理类 XmppIoHandler.

XmppIoHandler负责处理了与客户端的交互部分。
比如连接建立时以及出错时只打印日志，

    /**
     * Invoked from an I/O processor thread when a new connection has been created.
     */
    public void sessionCreated(IoSession session) throws Exception {
        log.debug("sessionCreated()...");
    }

    /**
     * Invoked when any exception is thrown.
     */
    public void exceptionCaught(IoSession session, Throwable cause)
            throws Exception {
        log.debug("exceptionCaught()...");
        log.error(cause);
    }

当连接闲置或者关闭时除了打印外还要关闭其与客户端得链接，

    /**
     * Invoked when a connection is closed.
     */
    public void sessionClosed(IoSession session) throws Exception {
        log.debug("sessionClosed()...");
        Connection connection = (Connection) session.getAttribute(CONNECTION);
        connection.close();
    }

    /**
     * Invoked with the related IdleStatus when a connection becomes idle.
     */
    public void sessionIdle(IoSession session, IdleStatus status)
            throws Exception {
        log.debug("sessionIdle()...");
        Connection connection = (Connection) session.getAttribute(CONNECTION);
        if (log.isDebugEnabled()) {
            log.debug("Closing connection that has been idle: " + connection);
        }
        connection.close();
    }

当连接打开时加载指定编码集的解析器，建立链接，并加载：

    /**
     * Invoked when a connection has been opened.
     */
    public void sessionOpened(IoSession session) throws Exception {
        log.debug("sessionOpened()...");
        log.debug("remoteAddress=" + session.getRemoteAddress());
        // Create a new XML parser
        XMLLightweightParser parser = new XMLLightweightParser("UTF-8");
        session.setAttribute(XML_PARSER, parser);
        // Create a new connection
        Connection connection = new Connection(session);
        session.setAttribute(CONNECTION, connection);
        session.setAttribute(STANZA_HANDLER, new StanzaHandler(serverName, connection));
    }

当通过连接传送或者收发数据时的处理：

    /**
     * Invoked when a message is received.
     */
    public void messageReceived(IoSession session, Object message) throws Exception {
        log.debug("messageReceived()...");
        log.debug("RCVD: " + message);

        // Get the stanza handler
        StanzaHandler handler = (StanzaHandler) session.getAttribute(STANZA_HANDLER);

        // Get the XMPP packet parser
        int hashCode = Thread.currentThread().hashCode();
        XMPPPacketReader parser = parsers.get(hashCode);
        if (parser == null) {
            parser = new XMPPPacketReader();
            parser.setXPPFactory(factory);
            parsers.put(hashCode, parser);
        }

        // The stanza handler processes the message
        try {
            handler.process((String) message, parser);
        } catch (Exception e) {
            log.error("Closing connection due to error while processing message: " + message, e);
            Connection connection = (Connection) session.getAttribute(CONNECTION);
            connection.close();
        }
    }

    /**
     * Invoked when a message written by IoSession.write(Object) is sent out.
     */
    public void messageSent(IoSession session, Object message) throws Exception {
        log.debug("messageSent()...");
    }

MINA的基本基本架构与实现可以先到此暂停一下，后面需要在偏向XMPP协议方面说下消息服务器的通讯内容与实现，以方便了解消息服务器的整体流程。当然，最好的方式还是去通读一下XMPP的官方协议，笔者参考的是RFC6120.英文,看着头大的话可以参考中文WIKI站点.以便可以了解不同的信息头的含义以及交互方式。

首先，先说下xmpp的基本格式
RESPONSE STREAM(Server端)
INITIAL STREAM(Client端)


<stream><stream>
<presence><show/></presence>
<message to='foo'> <body/></message>
<iq to='bar' type='get'><query/> </iq>
<iq from='bar'  type='result'><query/></iq>
[ ... ]
</stream></stream>

当一端发起一个长连接的会话<stream></stream>称之为XML流，而在<stream>之间的完整信息片段称之为XML节。简单来说消息服务器就是在一个XML流中，通过各种不同XML节完成信息的交互。不明白的话，赶紧去看XMPP协议，现在还来得及。

下面的交互是客户端与服务器端进行通讯，以完成服务器端的注册监听.
Server(消息服务器)
Client(手机客户端)

<stream:stream to="192.168.0.68" xmlns="jabber:client" xmlns:stream="http://etherx.jabber.org/streams" version="1.1">

<?xml version='1.0' encoding='UTF-8'?><stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" from="127.0.0.1" id="deb12279" xml:version="1.0">

<stream:features><starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"></starttls><auth xmlns="http://jabber.org/features/iq-auth"/><register xmlns="http://jabber.org/features/iq-register"/></stream:features>

<starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"/>

<iq id="xfP9F-0" type="set"><query xmlns="jabber:iq:register"><password>354316034600157</password><username>354316034600157</username></query></iq>

<iq type="result" id="xfP9F-0" to="127.0.0.1/deb12279"/>

<iq id="xfP9F-1" type="get"><query xmlns="jabber:iq:auth"><username>354316034600157</username></query></iq>

<iq type="result" id="xfP9F-1"><query xmlns="jabber:iq:auth"><username>354316034600157</username><password/><digest/><resource/></query></iq>

<iq id="xfP9F-2" type="set"><query xmlns="jabber:iq:auth"><username>354316034600157</username><digest>a77d56f44a572aef5414446ba473cbfd7ba5fb41</digest><resource>AndroidpnClient</resource></query></iq>

<iq type="result" id="xfP9F-2" to="354316034600157@127.0.0.1/AndroidpnClient"/>

<iq id="xfP9F-3" type="get"><query xmlns="jabber:iq:roster" ></query></iq>

<presence id="xfP9F-4"></presence>

A.当手机端安装应用后，手机会向指定好的服务器发送一个XML节
  <stream:stream to="192.168.0.68" xmlns="jabber:client" xmlns:stream="http://etherx.jabber.org/streams" version="1.1">

由于是内网测试，消息服务器为192.168.0.68.
To表示该消息的目标地址即消息服务器的地址，

Xmlns表示命名空间，当然androidpn这里不大规范,客户端，服务器端均使用的是jabber:client值。

B．当服务器收到后，也会发起一个XML流,标明一系列的规范。
  <?xml version='1.0' encoding='UTF-8'?><stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" from="127.0.0.1" id="deb12279" xml: version="1.0">

当然为了安全起见，服务器还不会立刻进行资源交互。会需要进行所谓的流协商
  <stream:features><starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"></starttls><auth xmlns="http://jabber.org/features/iq-auth"/><register xmlns="http://jabber.org/features/iq-register"/></stream:features>

stream:features表明在进行资源交互前，还需要进行协商，协商内容包括

TLS方式的安全认证,用户认证，以及注册

C．客户端收到后，会发送
  <starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"/>
表明客户端也是用tls方式的认证(默认虽然使用的是tls,但更安全的方案是使用sasl方式)
<iq id="xfP9F-0" type="set">
<query xmlns="jabber:iq:register">
<password>354316034600157</password>
<username>354316034600157</username>
</query></iq>

Iq是(Info/Query)的缩写, 是三种通讯原语之一，其他两个为message和presence.
该原语应用于”请求-应答”机制。
服务端收到注册请求后，会发送
  <iq type="result" id="xfP9F-0" to="127.0.0.1/deb12279"/>
进行确认。注意这里的id与客户端请求的id是相同的,类型为result,当然注册完成后，客户端会继续进行认证操作。
客户端会查询服务器端的认证方式。
  <iq id="xfP9F-1" type="get"><query xmlns="jabber:iq:auth"><username>354316034600157</username></query></iq>
服务器返回如下信息，username、password、digest、resource标明需要填充。
<iq type="result" id="xfP9F-1"><query xmlns="jabber:iq:auth"><username>354316034600157</username><password/>
<digest/><resource/></query></iq>
客户端返回已填充内容的XML节信息
<iq id="xfP9F-2" type="set"><query xmlns="jabber:iq:auth"><username>354316034600157</username><digest>
a77d56f44a572aef5414446ba473cbfd7ba5fb41</digest><resource>AndroidpnClient</resource></query></iq>

服务器接收到消息后，确认无误后返回，注意to的内容即所谓xmpp中的JID,消息服务器是通过jid 来选择客户端的。当然这里的客户端JID由android设备的“IMEI号“+@+IP+/AndroidpnClient构成的。
<iq type="result" id="xfP9F-2" to="354316034600157@127.0.0.1/AndroidpnClient"/>

D.至此，流协商中，服务器端要求的信息，客户端都已经发送完毕，但现在还不算完，客户端还需要发送下面的消息，表示订阅消息服务器的所有通知
<iq id="xfP9F-3" type="get"><query xmlns="jabber:iq:roster" ></query></iq>
<presence id="xfP9F-4"></presence>
至此，流协商的工作已全部完成，后面就可以进行资源信息交互了，也就是所谓的发送通知。

E．PUSH通知
服务器推送消息，to内容表示目标地址,apikey与google的那种公共服务类似的调用密钥。Title表示通知标题,message表示正文内容，pushid表示此次通知的唯一标识
<iq type="set" id="975-0" to="354316034600157@127.0.0.1/AndroidpnClient">
<notification xmlns="androidpn:iq:notification"><id>fd632e6e</id><apiKey>
1234567890</apiKey><title>DokdoIsland</title><message>Dokdo is a Korean
</message><uri></uri><pushId>402880c43588f7f6013588f966920005</pushId></notification></iq>

客户端返回信息，表示收到。
<iq id="xfP9F-5" type="get"><query xmlns="jabber:iq:register"><pushId>402880c43588f7f6013588f966920005
</pushId></query></iq>

<iq type="result" id="xfP9F-5" to="354316034600157@127.0.0.1/AndroidpnClient"/>

至此，一个完整的推送业务已经基本实现了。但目前服务器和客户端均未关闭此远程连接，该XML流仍然开启，只不过状态改为闲置。如果向服务器端发送</stream:stream>的话则服务器端会回应一个</stream:stream>而后将XML流关闭。

评论

4 楼 小菜鸟蜕变之路 2014-06-10  

LZ,你好：
   非常感谢你分享的这篇文章。我这几天遇到消息推送功能，我只想用Androidpn的服务端，自己写一段代码来联接服务端，实现消息推送。我是这么做的
客户端
String string = "<stream:stream to='127.0.0.1' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams' version='1.1'><starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>";

session.write(string);
可是不能注册。你能给我点支持吗？非常感谢！

3 楼 evenfun 2013-10-11  

MINA自身带有一些常用的过滤器，例如LoggingFilter（日志记录）、BlackListFilter（黑名单过滤）、CompressionFilter（压缩）、SSLFilter（SSL加密）等。

具体用法，请参考下面例子:

public class MinaTimeServer {

    private static final int PORT = 9123;

    public static void main(String[] args) throws IOException {
 
        //Mina包装了J2SE中的ByteBuffer类,用法也一样
        ByteBuffer.setUseDirectBuffers(false);

    //设置缓存策略
        ByteBuffer.setAllocator(new SimpleByteBufferAllocator());

       //创建一个SocketAcceptor,用于在服务器端监听客户端的连接.
        IoAcceptor acceptor = new SocketAcceptor();

        //创建一个与SocketAcceptor相关联的配置对象.
        SocketAcceptorConfig cfg = new SocketAcceptorConfig();
        cfg.getSessionConfig().setReuseAddress( true );
        cfg.getFilterChain().addLast( "logger", new LoggingFilter() );
        cfg.getFilterChain().addLast(”compress”, new org.apache.mina.filter.CompressionFilter()); 

        //绑定端口,处理对象和配置对象.
        acceptor.bind( new InetSocketAddress(PORT), new TimeServerHandler(), cfg);
        System.out.println("MINA Time server started.");
    }
}

2 楼 evenfun 2013-10-11  

我想了一下，思路如下：

因为网络传输的数据量比较大，为了节省带宽，我们可以在服务器端把传输数据先进行压缩处理，然后再写入流进行传输；客户端接收到压缩流后，先解压，再解析成原文。

在AndroidPN需要处理的地方，服务器写入流的地方，和客户端接收到流的地方。

网络传输数据压缩请参考：
http://sunyf.iteye.com/blog/1955248

1 楼 majian_1987 2013-08-26  

请问，关于AndroidPN中的消息压缩具体该怎么实现，具体是什么样的流程，客户端和服务器应该做哪些修改，因项目中遇到了要用AndroidPN的消息压缩，请指教下。谢谢