最近突然對安卓消息推送的原理感興趣,找了不少資料,實現(xiàn)了一個包括服務(wù)端和客戶端的簡單Demo。

在具體實現(xiàn)的時候踩了不少坑,這里做一下筆記,防止以后忘記。

安卓消息推送的實現(xiàn)方案有下面幾種:

• MQTT協(xié)議實現(xiàn)
• XMPP協(xié)議實現(xiàn)
• C2DM云端推送功能(google官方提供,系統(tǒng)內(nèi)置,但是國內(nèi)用不了......)
• 中國統(tǒng)一推送(工信部牽頭成立,但是目前只是開了幾次會議,并沒有什么實際的接口出來,不過以后應(yīng)該會是中國境內(nèi)的首選方案)

我這里選擇了MQTT協(xié)議去實現(xiàn)。

MQTT協(xié)議

MQTT是一個客戶端服務(wù)端架構(gòu)的發(fā)布/訂閱模式的消息傳輸協(xié)議吕粗。它的設(shè)計思想是輕巧、開放、簡單祖秒、規(guī)范，因此易于實現(xiàn)舟奠。這些特點使得它對很多場景來說都是很好的選擇竭缝，包括受限的環(huán)境如機器與機器的通信（M2M）以及物聯(lián)網(wǎng)環(huán)境（IoT），這些場景要求很小的代碼封裝或者網(wǎng)絡(luò)帶寬非常昂貴沼瘫。

本協(xié)議運行在TCP/IP抬纸，或其它提供了有序、可靠耿戚、雙向連接的網(wǎng)絡(luò)連接上湿故。它有以下特點：

使用發(fā)布/訂閱消息模式，提供了一對多的消息分發(fā)和應(yīng)用之間的解耦膜蛔。

消息傳輸不需要知道負(fù)載內(nèi)容坛猪。

提供三種等級的服務(wù)質(zhì)量：.

“最多一次”，盡操作環(huán)境所能提供的最大努力分發(fā)消息皂股。消息可能會丟失砚哆。例如，這個等級可用于環(huán)境傳感器數(shù)據(jù)屑墨，單次的數(shù)據(jù)丟失沒關(guān)系躁锁，因為不久之后會再次發(fā)送。
“至少一次”卵史，保證消息可以到達(dá)战转，但是可能會重復(fù)。
“僅一次”以躯，保證消息只到達(dá)一次槐秧。例如啄踊，這個等級可用在一個計費系統(tǒng)中，這里如果消息重復(fù)或丟失會導(dǎo)致不正確的收費刁标。
很小的傳輸消耗和協(xié)議數(shù)據(jù)交換颠通，最大限度減少網(wǎng)絡(luò)流量

異常連接斷開發(fā)生時，能通知到相關(guān)各方膀懈。

上面這一段話是從網(wǎng)友翻譯的MQTT中文文檔直接復(fù)制的顿锰。有興趣的同學(xué)可以直接訪問MQTT協(xié)議中文版查看具體的協(xié)議細(xì)節(jié)。

MQTT原理

MQTT協(xié)議原理的原理簡單說來就是客戶端與服務(wù)端通過心跳包來保持連接启搂∨鹂兀客戶接收端向服務(wù)端訂閱消息,客戶發(fā)布端向服務(wù)端發(fā)布消息。服務(wù)端再將消息分發(fā)給訂閱了該消息的客戶接收端胳赌。

原理圖如下:

1.png

實現(xiàn)庫的選擇

因為MQTT協(xié)議中文版上面已經(jīng)有了整個QMTT的協(xié)議細(xì)節(jié),所以理論上如果你夠厲害的話,完全可以自己從零開始實現(xiàn)服務(wù)端和客戶端牢撼。

但是從實際項目中,我還是傾向選擇官方提供或者第三方開源的項目直接使用。

其實官方已經(jīng)給我們提供了一些推薦實現(xiàn):

https://github.com/mqtt/mqtt.github.io/wiki/software?id=software

MQTT 服務(wù)器搭建

我這邊選擇使用apache-apollo這個開源的MQTT服務(wù)器疑苫。

網(wǎng)上有不少的博客都有講它的配置方法的,但是我按著做之后都出現(xiàn)了一些問題熏版。

1、安裝jdk

首先需要去安裝jdk:

sudo apt-get install default-jdk

2捍掺、下載apache-apollo

然后到官網(wǎng)下載最新的軟件撼短。我這邊使用的是騰訊云的ubuntu服務(wù)器,所以就下載了linux的版本。

下載完之后解壓到/opt目錄下(其實任意目錄均可,只不過我用linux習(xí)慣放這里):

/opt/apache-apollo-1.7.1

3乡小、創(chuàng)建項目

然后進(jìn)入任意目錄使用下面命令創(chuàng)建一個項目(官方管它叫broker):

/opt/apache-apollo-1.7.1/bin/apollo create mybroker

它會在當(dāng)前目錄創(chuàng)建一個mybroker目錄,里面就是你的項目。

4饵史、編輯admin ip配置

可以編輯mybroker/etc/apollo.xml進(jìn)行一些配置满钟。

admin后臺會默認(rèn)被綁定到127.0.0.1,這樣你是不能通過你電腦的瀏覽器去訪問服務(wù)器的admin后臺的:

<web_admin bind="http://127.0.0.1:61680"/>
<web_admin bind="https://127.0.0.1:61681"/>

我們將它改成0.0.0.0:

<web_admin bind="http://0.0.0.0:61680"/>
<web_admin bind="https://0.0.0.0:61681"/>

注意這里的61680和61681端口,之后需要訪問該端口去登陸admin后臺

5、啟動MQTT服務(wù)

你可以進(jìn)到mybroker/bin/目錄中使用下面兩種方式中的一種去啟動服務(wù):

• 當(dāng)前進(jìn)程阻塞啟動:

./apollo-broker run

• 啟動后臺服務(wù):

./apollo-broker-service start

然后你就可以在你的電腦打開瀏覽器輸入網(wǎng)址訪問MQTT后臺了:

• 如果你的服務(wù)是跑在阿里云胳喷、騰訊云這樣的服務(wù)器上:

http://服務(wù)器ip:61680

• 如果你的服務(wù)就是跑在你自己的電腦上:

http://0.0.0.0:61680

它的需要輸入賬號密碼才能登陸湃番。默認(rèn)賬號是admin、密碼是password

2.png

3.png

Python paho-mqtt

我們可以用python寫一個客戶端去驗證搭建的mqtt服務(wù)器是否可用吭露。

首先需要下載paho-mqtt:

pip install paho-mqtt

Python paho-mqtt 簡單收發(fā)端Demo

接收端代碼:

import paho.mqtt.client as mqtt                                                  

def on_connect(client, userdata, flags, rc):                                     
    print("Connected with result code "+str(rc))                                 
    client.subscribe("topic/test")                                               

def on_message(client, userdata, msg):                                           
    print("on_message "+msg.topic+" "+str(msg.payload))                          

client = mqtt.Client(client_id="", clean_session=True, userdata=None, protocol=mqtt.MQTTv31, transport="tcp")
client.username_pw_set("admin", "password")                                      
client.on_connect = on_connect                                                   
client.on_message = on_message                                                   

client.connect("www.islinjw.cn", 61613, 60)                                      
client.loop_forever()

發(fā)送端代碼:

import paho.mqtt.client as mqtt                                                  

def on_connect(client, userdata, flags, rc):                                     
    print("Connected with result code "+str(rc))                                 

client = mqtt.Client(client_id="", clean_session=True, userdata=None, protocol=mqtt.MQTTv31, transport="tcp")
client.username_pw_set("admin", "password")                                      
client.on_connect = on_connect                                                         

client.connect("www.islinjw.cn", 61613, 60)                                                                     
client.publish("topic/test", "hello world")

我們先運行接收端,再運行發(fā)送端,就可以在接收端看到"hello world"的打印

常見錯誤

這里需要注意的是一定要配置協(xié)議版本為MQTTv31,網(wǎng)上的demo代碼都沒有配置,沒有的話python這就會報錯:

[Errno 104] Connection reset by peer

服務(wù)器則會報空指針,我們可以在mybroker/log/stacktrace.log中看到:

java.lang.NullPointerException
        at org.apache.activemq.apollo.mqtt.MqttProtocolHandler.on_mqtt_connect(MqttProtocolHandler.java:443)
        at org.apache.activemq.apollo.mqtt.MqttProtocolHandler$9.call(MqttProtocolHandler.java:410)
        at org.apache.activemq.apollo.util.UnitFn1.apply(Scala2JavaHelper.scala:41)
        at org.apache.activemq.apollo.mqtt.MqttProtocolHandler.on_transport_command(MqttProtocolHandler.java:377)
        at org.apache.activemq.apollo.broker.BrokerConnection.on_transport_command(Connection.scala:144)
        at org.apache.activemq.apollo.broker.Connection$$anon$1.onTransportCommand(Connection.scala:71)
        at org.fusesource.hawtdispatch.transport.TcpTransport.drainInbound(TcpTransport.java:709)
        at org.fusesource.hawtdispatch.transport.TcpTransport$9.run(TcpTransport.java:770)
        at org.fusesource.hawtdispatch.internal.SerialDispatchQueue.run(SerialDispatchQueue.java:100)
        at org.fusesource.hawtdispatch.internal.pool.SimpleThread.run(SimpleThread.java:77)

而如果沒有設(shè)置賬號密碼的話就收到result code 4:

Connected with result code 4

我們可以從官方文檔看到result code 4代表用戶名或者密碼錯誤:

0: Connection successful
1: Connection refused - incorrect protocol version
2: Connection refused - invalid client identifier
3: Connection refused - server unavailable
4: Connection refused - bad username or password
5: Connection refused - not authorised 6-255: Currently unused.

權(quán)限配置

我們設(shè)想一下,如果沒有賬戶系統(tǒng),那么只要知道服務(wù)器的ip和端口,就能隨便發(fā)送消息了,這樣誰都能給你的應(yīng)用推送消息,十分危險吠撮。

所以mqtt是需要用賬戶密碼去建權(quán)的,有些賬戶只能發(fā)送,有些賬戶只能接收,而有些賬戶全部都能做。

創(chuàng)建用戶

我們可以編輯mybroker/etc/users.properties添加user1和user2:

admin=password
user1=123456
user2=654321

等號的左邊是賬戶,右邊是密碼讲竿。所以我們也能在這里改admin的密碼

創(chuàng)建用戶組

創(chuàng)建完用用戶,我們還需要編輯mybroker/etc/groups.properties給用戶指定用戶組:

admins=admin
groupsend=user1
grouprecv=user2

# 還可以用下面的方式將多個用戶指定到一個用戶組
# groupdemo = user1|user2

設(shè)置用戶組權(quán)限

最后我們就能在mybroker/etc/apollo.xml設(shè)置用戶組權(quán)限了:

<access_rule allow="admins" action="*"/>
<access_rule allow="*" action="connect" kind="connector"/>
<access_rule allow="groupsend" action="connect create send"/>
<access_rule allow="grouprecv" action="connect receive"/>

可以給一個用戶組設(shè)置多個權(quán)限,多個權(quán)限之間用空格分割泥兰。從官方文檔可以看到權(quán)限有下面的類別:

• admin : use of the administrative web interface
• monitor : read only use of the administrative web interface
• config : use of the administrative web interface to access and change the broker configuration.
• connect : allows connections to the connector or virtual host
• create : allows creation
• destroy : allows destruction
• send : allows the user to send to the destination
• receive : allows the user to send to do non-destructive reads from the destination
• consume : allows the user to do destructive reads against a destination
• * : All actions

配置好之后我們的user1就只能發(fā)送消息,user2就只能接收消息了。

安卓端實現(xiàn)

官方推薦的qatja-android我看了一下题禀，它的實現(xiàn)太挫了,所以在github上搜索了下,發(fā)現(xiàn)了個不錯的庫mqtt-client鞋诗。

添加依賴：

compile 'org.fusesource.mqtt-client:mqtt-client:1.14'

因為代碼比較簡單,所以我就直接貼上來了:

public class MainActivity extends AppCompatActivity {
    public static final String TAG = "MainActivity";
    public static final String TOPIC = "topic/test";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        try {
            testMqtt();
        } catch (URISyntaxException e) {
            Log.e(TAG, "testMqtt failed", e);
        }
    }

    private void testMqtt() throws URISyntaxException {
        MQTT mqtt = new MQTT();
        mqtt.setHost("www.islinjw.cn", 61613);
        mqtt.setVersion("3.1");
        mqtt.setUserName("admin");
        mqtt.setPassword("password");

        final CallbackConnection connection = mqtt.callbackConnection();

        //設(shè)置監(jiān)聽
        connection.listener(new ExtendedListener() {
            @Override
            public void onPublish(UTF8Buffer topic, Buffer body, Callback<Callback<Void>> ack) {
                Log.d(TAG, "onPublish " + topic.toString() + " " + body.toString());

                NotificationManager notifyManager
                        = (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);
                NotificationCompat.Builder builder = new NotificationCompat
                        .Builder(MainActivity.this)
                        .setSmallIcon(R.mipmap.ic_launcher)
                        .setContentTitle(topic.toString())
                        .setContentText(body.ascii().toString());
                notifyManager.notify(1, builder.build());
            }

            @Override
            public void onConnected() {
                Log.d(TAG, "onConnected");
            }

            @Override
            public void onDisconnected() {
                Log.d(TAG, "onDisconnected");
            }

            @Override
            public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
                Log.d(TAG, "onPublish " + topic.toString() + " " + body);
                ack.run();
            }

            @Override
            public void onFailure(Throwable value) {
                Log.d(TAG, "onFailure");
            }
        });

        //連接服務(wù)器
        connection.connect(new Callback<Void>() {
            public void onFailure(Throwable value) {
                Log.d(TAG, "connect failure");
            }

            public void onSuccess(Void v) {
                //訂閱消息
                Topic[] topics = {new Topic(TOPIC, QoS.AT_LEAST_ONCE)};
                connection.subscribe(topics, new Callback<byte[]>() {
                    public void onSuccess(byte[] qoses) {
                        Log.d(TAG, "subscribe success");
                    }

                    public void onFailure(Throwable value) {
                        Log.e(TAG, "subscribe failure", value);
                        connection.disconnect(null); //斷開連接
                    }
                });

                //發(fā)布一個消息
                byte[] payload = "hello world".getBytes();
                connection.publish(TOPIC, payload, QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
                    public void onSuccess(Void v) {
                        Log.d(TAG, "publish success");
                    }

                    public void onFailure(Throwable value) {
                        Log.e(TAG, "publish failure", value);
                        connection.disconnect(null); //斷開連接
                    }
                });

//                //斷開連接
//                connection.disconnect(new Callback<Void>() {
//                    public void onSuccess(Void v) {
//                        Log.d(TAG, "disconnect success");
//                    }
//
//                    public void onFailure(Throwable value) {
//                        Log.d(TAG, "disconnect failure");
//                        // disconnects是不會失敗的,也就是這里永遠(yuǎn)不會被調(diào)到
//                    }
//                });
            }
        });
    }
}

然后我們就可以用發(fā)送端的py腳本將消息推送給安卓客戶端了。

4.png

主題名和主題通配符

發(fā)布的消息都有一個主題名,例如我們之前作為例子的"topic/test"迈嘹∠鞅颍客戶端向服務(wù)端訂閱感興趣的主題全庸。當(dāng)有某主題的消息被發(fā)布的時候,服務(wù)端就會將消息分發(fā)給訂閱了該主題的客戶端。

主題名可以用分割符"/"如果存在的話就會將主題分割為多個主題層級融痛。

有了主題層級的概念之后我們就可以用主題通配符去過濾不同的主題壶笼。

我這里只做簡單介紹,詳細(xì)的信息可以參考文檔

多層通配符

數(shù)字標(biāo)志（"#"）是用于匹配主題中任意層級的通配符。多層通配符表示它的父級和任意數(shù)量的子層級雁刷。多層通配符必須位于它自己的層級或者跟在主題層級分隔符后面覆劈。不管哪種情況，它都必須是主題過濾器的最后一個字符安券。

例如墩崩，如果客戶端訂閱主題 "sport/tennis/player1/#"，它會收到使用下列主題名發(fā)布的消息：

• "sport/tennis/player1"
• "sport/tennis/player1/ranking"
• "sport/tennis/player1/score/wimbledon"
• "sport/#"也匹配單獨的 "sport" 侯勉，因為 # 包括它的父級鹦筹。
• "#"也是有效的，會收到所有的應(yīng)用消息址貌。

多層通配符用法舉例:

• "sport/tennis/#"也是有效的铐拐。
• "sport/tennis#"是無效的。
• "sport/tennis/#/ranking"是無效的练对。

單層通配符

加號 ("+") 是只能用于單個主題層級匹配的通配符遍蟋。在主題過濾器的任意層級都可以使用單層通配符，包括第一個和最后一個層級螟凭。然而它必須占據(jù)過濾器的整個層級虚青。可以在主題過濾器中的多個層級中使用它螺男，也可以和多層通配符一起使用棒厘。

例如， "sport/tennis/+" 匹配 "sport/tennis/player1" 和 "sport/tennis/player2",但是不匹配 "sport/tennis/player1/ranking" 下隧。同時奢人，由于單層通配符只能匹配一個層級， "sport/+" 不匹配 "sport" 但是卻匹配 "sport/"淆院。

單層通配符的一些使用情況:

• "+" 是有效的何乎。
• "+/tennis/#" 是有效的。
• "sport+" 是無效的土辩。
• "sport/+/player1" 也是有效的支救。
• "/finance" 匹配 "+/+" 和 "/+" ，但是不匹配 "+"拷淘。

開始通配符

美元符號("$") 用于匹配起始主題,如"$SYS/" 被廣泛用作包含服務(wù)器特定信息或控制接口的主題的前綴搂妻。

開始通配符的一些用法:

• 訂閱 "#" 的客戶端不會收到任何發(fā)布到以 "$" 開頭主題的消息。
• 訂閱 "+/monitor/Clients" 的客戶端不會收到任何發(fā)布到 "$SYS/monitor/Clients" 的消息辕棚。
• 訂閱 "$SYS/#" 的客戶端會收到發(fā)布到以 “$SYS/” 開頭主題的消息欲主。
• 訂閱 "$SYS/monitor/+" 的客戶端會收到發(fā)布到 "$SYS/monitor/Clients" 主題的消息邓厕。
• 如果客戶端想同時接受以 "$SYS/" 開頭主題的消息和不以 "$" 開頭主題的消息，它需要同時訂閱 "#" 和 "$SYS/#"扁瓢。