杂七杂八（42）

推送功能在手机应用开发中越来越重要，已经成为手机开发的必须。在Android应用开发中，由于众所周知的原因，Android消息推送我们不得不大费周折。本文就是用来和大家共同探讨一种Android消息推送的完美解决方案。
　　1.消息推送基础
& && && &消息推送，就是在互联网上通过定期传送用户需要的信息来减少信息过载的一项新技术。推送技术通过自动传送信息给用户，来减少用于网络上搜索的时间。它根据用户的兴趣来搜索、过滤信息，并将其定期推给用户，帮助用户高效率地发掘有价值的信息
当我们开发需要和服务器交互的移动应用时，基本上都需要和服务器进行交互，包括上传数据到服务器，同时从服务器上获取数据。
一般情况下，客户端与服务器之间通讯客户端是主动的，但这就存在一个问题就是一旦服务器数据有更新或者服务器要下发通知给客户端只能等客户端连接的时候才能实现。这种方式使消息失去了实时性。
如何使客户端能够实时的收到服务器的消息和通知，总体来说有两种方式，第一种是客户端使用Pull（拉）的方式，就是隔一段时间就去服务器上获取一下信息，看是否有更新的信息出现。第二种就是&服务器使用Push（推送）的方式，当服务器端有新信息了，则把最新的信息Push到客户端上。这样，客户端就能自动的接收到消息。
　　虽然Pull和Push两种方式都能实现获取服务器端更新信息的功能，但是明显来说Push方式比Pull方式更优越。因为Pull方式更费客户端的网络流量，更主要的是费电量，还需要我们的程序不停地去监测服务端的变化。
　　2.&几种常见的解决方案实现原理
　　1）轮询(Pull)方式：客户端定时向服务器发送询问消息，一旦服务器有变化则立即同步消息。
　　2）SMS(Push)方式：通过拦截SMS消息并且解析消息内容来了解服务器的命令，但这种方式一般用户在经济上很难承受。
　&3）持久连接(Push)方式：客户端和服务器之间建立长久连接，这样就可以实现消息的及时行和实时性。
　　3、消息推送解决方案概述
　　A、C2DM云端推送方案
在Android手机平台上，Google提供了C2DM（Cloudto Device Messaging）服务。Android Cloud to Device Messaging (C2DM)是一个用来帮助开发者从服务器向Android应用程序发送数据的服务。该服务提供了一个简单的、轻量级的机制，允许服务器可以通知移动应用程序直接与服务器进行通信，以便于从服务器获取应用程序更新和用户数据。
该方案存在的主要问题是C2DM需要依赖于Google官方提供的C2DM服务器，由于国内的网络环境，这个服务经常不可用。
　　B、MQTT协议实现Android推送
　　采用MQTT协议实现Android推送功能也是一种解决方案。MQTT是一个轻量级的消息发布/订阅协议，它是实现基于手机客户端的消息推送服务器的理想解决方案。
　　wmqtt.jar&是IBM提供的MQTT协议的实现。我们可以从这里（/tokudu/AndroidPushNotificationsDemo）下载该项目的实例代码，并且可以找到一个采用PHP书写的服务器端实现（/tokudu/PhpMQTTClient）。
　　C、RSMB实现推送功能
　　Really Small Message Broker (RSMB)&，是一个简单的MQTT代理，同样由IBM提供，其查看地址是：http://www./tech/rsmb。缺省打开1883端口，应用程序当中，它负责接收来自服务器的消息并将其转发给指定的移动设备。SAM是一个针对MQTT写的PHP库。我们可以从这个http://pecl.php.net/package/sam/download/0.2.0地址下载它.
　&&D、XMPP协议实现Android推送
　　Google官方的C2DM服务器底层也是采用XMPP协议进行的封装。XMPP(可扩展通讯和表示协议)是基于可扩展标记语言（XML）的协议，它用于即时消息（IM）以及在线探测。这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息。
　　androidpn是一个基于XMPP协议的java开源Android push notification实现。它包含了完整的客户端和服务器端。但也存在一些不足之处：
　　1）&比如时间过长时，就再也收不到推送的信息了。
　　2）性能上也不够稳定。
3）如果将消息从服务器上推送出去，就不再管理了，不管消息是否成功到达客户端手机上。
如果我们要使用androidpn，则还需要做大量的工作，需要理解XMPP协议、理解Androidpn的实现机制，需要调试内部存在的BUG。
　　E、使用第三方平台
　　目前国内、国外有一些推送平台可供使用，但是涉及到收费问题、保密问题、服务质量问题、扩展问题等等，又不得不是我们望而却步。
& && && &4、消息推送完美方案
　& &&&综合以上论述，在建立Android消息推送方面可谓方案多多，但每一款方案都有其优缺点。但无论如何，还是自己搭建一个推送平台是上策。因为你有、他有不如自己有。
& && && &在搭建自有推送平台上建议使用《九日升Android消息推送组件》（http://www.bjjrs.net/product/37.html）。该组不仅可以拿来即用，并且还可以提供源码以便扩展，实现自己的特殊需求。
& && &&A、推送原理
& &&九日升Android消息推送组件基于XMPP协议实现Android推送。XMPP（可扩展通讯和表示协议）是基于可扩展标记语言（XML）的协议，它用于即时消息（IM）以及在线探测。这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息。
& &&九日升Android消息推送组件实现原理见下图：
& && && &&&&
& && && && && && && && && && && && && && && && &&&图1-消息推送原理图
& &&九日升Android消息推送组件由服务器部分和客户端部分组成。每一部分都由XMPP协议组件和外部接口组件构成。XMPP协议组件负责服务器和Android客户端间的连接管理、消息通讯，外部接口组件负责接收应用系统、客户端应用的命令，向应用系统发送接收到的通知消息。
& &&九日升Android消息组件提供基于Tomcat的服务器应用和Android开发jar包。其中基于Tomcat的服务器应用直接在Tomcat上部署即可，Android开发jar包引入Android项目即可。
& &B&集成方式
& & 1）、服务器部署
& &&九日升Android消息组件Tomcat的服务器应用直接部署在Tomcat中，端口号任意设定。
& & 2）、客户端jar包引用
& &&在Android项目中建立libs目录，然后将提供的Android开发jar包复制到该目录即可。见下图：
& && && && && && && && && && && && && && && && &&
& && && && && && && && && && && && && && && && && && && && && && && && && &&&
图2-jar包引入图
& & 3）、Android项目AndroidManifest.xml文件修改
&&在该文件中增加以下权限：
& &&&&uses-permission android:name=&android.permission.READ_PHONE_STATE& /&
& &&&&uses-permission android:name=&android.permission.ACCESS_NETWORK_STATE& /&
& &&&&uses-permission android:name=&android.permission.INTERNET& /&
& &&&&uses-permission android:name=&android.permission.ACCESS_WIFI_STATE& /&
& &&&&uses-permission android:name=&android.permission.CHANGE_WIFI_STATE& /&
& &&&&uses-permission android:name=&android.permission.VIBRATE& /&
& &在该文件中注册服务：
& &&&&service android:enabled=&true&
& &&&android:name=&com.bjjrs.server.NotificationService&
& &&&android:label=&NotificationService&&
& && && & &intent-filter&
& && && && & &action android:name=&com.bjjrs.server.NotificationService& /&
& && && &&/intent-filter&
& & &/service&
& &至此，九日升Android消息组件集成工作完成。
&&&&C、接口方式
& & 1）、服务器端接口采用基于http协议的访问方式，采用http协议从服务器中获取各种信息，实现通知消息的推送。
如使用以下方式和参数就可以实现各种用户消息的查询：
&&http://localhost:8080/user.do?action=getAllUser&isOnline=&userID=&userType=&deptID=&deptName=&realName=
& &&使用如下方式就可以实现各种消息的推送：
& & http://localhost:8080/notification.do?action=pushNoti&userNames=&title=&content=
& & 2）、Android客户端接口采用广播机制。
& &&消息接收：当XMPP协议组件接收到推送消息时，将按照一定格式广播该消息，通知客户端其他应用接收并处理该消息。
& &&消息发送：客户端应用需要向服务器或者其他客户端发送即时消息时，只需按一定格式广播该消息，XMPP组件就会自动接收该消息并发送到指定的其他客户端。
& &&D、优势特点
& & 1）、系统集成简单，无需复杂的设置。
& & 2）、Android客户端应用和九日升Android消息推送组件完全分离，通过接口相互调用，实现模块应用最优化。
& & 3）、客户端通讯机制采用广播方式，给客户端应用带来极大的灵活性和可扩展性，可以自由处理接收到的推送消息。
& & 4）、九日升Android消息推送组件在服务器端具备消息存储、消息重发、消息路由等功能，在客户端部分具备断线重连、、收到确认、阅读确认、消息发送、命令执行等功能，确保消息能够推送到客户端，同时也保证客户端能够收到、阅读消息。
& &E、&应用范围
&&九日升Android消息推送组件可在以下场景中使用：
& & 1）、用于消息推送。如：通知下达、应急指挥等。
& & 2）、用户及时消息交互。如在线聊天、工作情况交互等。
& & 3）、用于远程控制。如控制远程客户端的状态、数据上报等。
　最后，希望转载的朋友能够尊重作者的劳动成果，加上转载地址。
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：20595次
排名：千里之外
原创：30篇
转载：23篇
(3)(1)(2)(1)(1)(1)(4)(1)(2)(4)(1)(3)(3)(3)(2)(6)(5)(2)(1)(3)(1)(3)匿名用户不能发表回复！|
每天回帖即可获得10分可用分！小技巧：
你还可以输入10000个字符
(Ctrl+Enter)
请遵守CSDN，不得违反国家法律法规。
转载文章请注明出自“CSDN（www.csdn.net）”。如是商业用途请联系原作者。54564人阅读
Android（12）
MQTT实现消息推送
MQTT实现消息接收(接收消息需实现MqttSimpleCallback接口并实现它的publishArrived方法)必须注册接收消息方法
mqttClient.registerSimpleHandler(simpleCallbackHandler);// 注册接收消息方法
和订阅接主题
mqttClient.subscribe(TOPICS, QOS_VALUES);// 订阅接主题
package com.gmcc.kuchuan.
import mons.logging.L
import mons.logging.LogF
import com.ibm.mqtt.MqttC
import com.ibm.mqtt.MqttE
import com.ibm.mqtt.MqttSimpleC
* MQTT消息发送与接收
* @author Join
public class MqttBroker {
private final static Log logger = LogFactory.getLog(MqttBroker.class);// 日志对象
// 连接参数
private final static String CONNECTION_STRING = &tcp://localhost:9901&;
private final static boolean CLEAN_START =
private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
private final static String CLIENT_ID = &master&;// 客户端标识
private final static int[] QOS_VALUES = { 0, 0, 2, 0 };// 对应主题的消息级别
private final static String[] TOPICS = { &Test/TestTopics/Topic1&,
&Test/TestTopics/Topic2&, &Test/TestTopics/Topic3&,
&client/keepalive& };
private static MqttBroker instance = new MqttBroker();
private MqttClient mqttC
* 返回实例对象
public static MqttBroker getInstance() {
* 重新连接服务
private void connect() throws MqttException {
(&connect to mqtt broker.&);
mqttClient = new MqttClient(CONNECTION_STRING);
(&***********register Simple Handler***********&);
SimpleCallbackHandler simpleCallbackHandler = new SimpleCallbackHandler();
mqttClient.registerSimpleHandler(simpleCallbackHandler);// 注册接收消息方法
mqttClient.connect(CLIENT_ID, CLEAN_START, KEEP_ALIVE);
(&***********subscribe receiver topics***********&);
mqttClient.subscribe(TOPICS, QOS_VALUES);// 订阅接主题
(&***********CLIENT_ID:& + CLIENT_ID);
* 完成订阅后，可以增加心跳，保持网络通畅，也可以发布自己的消息
mqttClient.publish(&keepalive&, &keepalive&.getBytes(), QOS_VALUES[0],
true);// 增加心跳，保持网络通畅
* 发送消息
* @param clientId
* @param messageId
public void sendMessage(String clientId, String message) {
if (mqttClient == null || !mqttClient.isConnected()) {
connect();
(&send message to & + clientId + &, message is &
+ message);
// 发布自己的消息
mqttClient.publish(&GMCC/client/& + clientId, message.getBytes(),
0, false);
} catch (MqttException e) {
logger.error(e.getCause());
e.printStackTrace();
* 简单回调函数，处理server接收到的主题消息
* @author Join
class SimpleCallbackHandler implements MqttSimpleCallback {
* 当客户机和broker意外断开时触发 可以再此处理重新订阅
public void connectionLost() throws Exception {
// TODO Auto-generated method stub
System.out.println(&客户机和broker已经断开&);
* 客户端订阅消息后，该方法负责回调接收处理消息
public void publishArrived(String topicName, byte[] payload, int Qos,
boolean retained) throws Exception {
// TODO Auto-generated method stub
System.out.println(&订阅主题: & + topicName);
System.out.println(&消息数据: & + new String(payload));
System.out.println(&消息级别(0,1,2): & + Qos);
System.out.println(&是否是实时发送的消息(false=实时，true=服务器上保留的最后消息): &
+ retained);
public static void main(String[] args) {
new MqttBroker().sendMessage(&client&, &message&);
Android客户端：
核心代码：MQTTConnection内部类
import java.io.ByteArrayInputS
import java.io.F
import java.io.IOE
import java.io.InputS
import java.io.RandomAccessF
import java.net.HttpURLC
import java.net.URL;
import java.util.ArrayL
import java.util.T
import java.util.TimerT
import android.app.AlarmM
import android.app.N
import android.app.NotificationM
import android.app.PendingI
import android.app.S
import android.content.BroadcastR
import android.content.C
import android.content.I
import android.content.IntentF
import android.content.SharedP
import android.database.C
import android.net.ConnectivityM
import android.net.NetworkI
import android.os.B
import android.os.B
import android.os.IB
import android.provider.ContactsC
import android.util.L
//此部分项目导包已被删除import com.ibm.mqtt.IMqttC
import com.ibm.mqtt.MqttC
import com.ibm.mqtt.MqttE
import com.ibm.mqtt.MqttP
import com.ibm.mqtt.MqttPersistenceE
import com.ibm.mqtt.MqttSimpleC
* PushService that does all of the work.
* Most of the logic is borrowed from KeepAliveService.
* /p/android-random/source/browse/trunk/TestKeepAlive/src/org/devtcg/demo/keepalive/KeepAliveService.java?r=219
public class PushService extends Service {
private MyBinder mBinder = new MyBinder();
// this is the log tag
public static final String TAG = &PushService&;
// the IP address, where your MQTT broker is running.
private static final String MQTT_HOST = &120.197.230.53&; // &209.124.50.174&;//
// the port at which the broker is running.
private static int MQTT_BROKER_PORT_NUM = 9901;
// Let's not use the MQTT persistence.
private static MqttPersistence MQTT_PERSISTENCE =
// We don't need to remember any state between the connections, so we use a
// clean start.
private static boolean MQTT_CLEAN_START =
// Let's set the internal keep alive for MQTT to 15 mins. I haven't tested
// this value much. It could probably be increased.
private static short MQTT_KEEP_ALIVE = 60 * 15;
// Set quality of services to 0 (at most once delivery), since we don't want
// push notifications
// arrive more than once. However, this means that some messages might get
// lost (delivery is not guaranteed)
private static int[] MQTT_QUALITIES_OF_SERVICE = { 0 };
private static int MQTT_QUALITY_OF_SERVICE = 0;
// The broker should not retain any messages.
private static boolean MQTT_RETAINED_PUBLISH =
// MQTT client ID, which is given the broker. In this example, I also use
// this for the topic header.
// You can use this to run push notifications for multiple apps with one
// MQTT broker.
public static String MQTT_CLIENT_ID = &client&;
// These are the actions for the service (name are descriptive enough)
public static final String ACTION_START = MQTT_CLIENT_ID + &.START&;
private static final String ACTION_STOP = MQTT_CLIENT_ID + &.STOP&;
private static final String ACTION_KEEPALIVE = MQTT_CLIENT_ID
+ &.KEEP_ALIVE&;
private static final String ACTION_RECONNECT = MQTT_CLIENT_ID
+ &.RECONNECT&;
// Connection log for the push service. Good for debugging.
private ConnectionLog mL
// Connectivity manager to determining, when the phone loses connection
private ConnectivityManager mConnM
// Notification manager to displaying arrived push notifications
private NotificationManager mNotifM
// Whether or not the service has been started.
private boolean mS
// This the application level keep-alive interval, that is used by the
// AlarmManager
// to keep the connection active, even when the device goes to sleep.
private static final long KEEP_ALIVE_INTERVAL = 1000 * 60 * 28;
// Retry intervals, when the connection is lost.
private static final long INITIAL_RETRY_INTERVAL = 1000 * 10;
private static final long MAXIMUM_RETRY_INTERVAL = 1000 * 60 * 30;
// Preferences instance
private SharedPreferences mP
// We store in the preferences, whether or not the service has been started
public static final String PREF_STARTED = &isStarted&;
// We also store the deviceID (target)
public static final String PREF_DEVICE_ID = &deviceID&;
// We store the last retry interval
public static final String PREF_RETRY = &retryInterval&;
// Notification title
public static String NOTIF_TITLE = &client&;
// Notification id
private static final int NOTIF_CONNECTED = 0;
// This is the instance of an MQTT connection.
private MQTTConnection mC
private long mStartT
boolean mShowFlag =// 是否显示通知
public static C
private boolean mRunFlag =// 是否向服务器发送心跳
Timer mTimer = new Timer();
// Static method to start the service
public static void actionStart(Context ctx) {
Intent i = new Intent(ctx, PushService.class);
i.setAction(ACTION_START);
ctx.startService(i);
PushService.ctx =
// Static method to stop the service
public static void actionStop(Context ctx) {
Intent i = new Intent(ctx, PushService.class);
i.setAction(ACTION_STOP);
ctx.startService(i);
// Static method to send a keep alive message
public static void actionPing(Context ctx) {
Intent i = new Intent(ctx, PushService.class);
i.setAction(ACTION_KEEPALIVE);
ctx.startService(i);
public void onCreate() {
super.onCreate();
log(&Creating service&);
mStartTime = System.currentTimeMillis();
mLog = new ConnectionLog();
Log.i(TAG, &Opened log at & + mLog.getPath());
} catch (IOException e) {
Log.e(TAG, &Failed to open log&, e);
// Get instances of preferences, connectivity manager and notification
// manager
mPrefs = getSharedPreferences(TAG, MODE_PRIVATE);
mConnMan = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);
mNotifMan = (NotificationManager) getSystemService(NOTIFICATION_SERVICE);
* If our process was reaped by the system for any reason we need to
* restore our state with merely a call to onCreate. We record the last
* &started& value and restore it here if necessary.
handleCrashedService();
// This method does any necessary clean-up need in case the server has been
// destroyed by the system
// and then restarted
private void handleCrashedService() {
if (wasStarted() == true) {
log(&Handling crashed service...&);
// stop the keep alives
stopKeepAlives();
// Do a clean start
public void onDestroy() {
log(&Service destroyed (started=& + mStarted + &)&);
// Stop the services, if it has been started
if (mStarted == true) {
if (mLog != null)
mLog.close();
} catch (IOException e) {
public void onStart(Intent intent, int startId) {
super.onStart(intent, startId);
log(&Service started with intent=& + intent);
if (intent == null) {
// Do an appropriate action based on the intent.
if (intent.getAction().equals(ACTION_STOP) == true) {
stopSelf();
} else if (intent.getAction().equals(ACTION_START) == true) {
} else if (intent.getAction().equals(ACTION_KEEPALIVE) == true) {
keepAlive();
} else if (intent.getAction().equals(ACTION_RECONNECT) == true) {
if (isNetworkAvailable()) {
reconnectIfNecessary();
public class MyBinder extends Binder {
public PushService getService() {
return PushService.
public IBinder onBind(Intent intent) {
// log helper function
private void log(String message) {
log(message, null);
private void log(String message, Throwable e) {
if (e != null) {
Log.e(TAG, message, e);
Log.i(TAG, message);
if (mLog != null) {
mLog.println(message);
} catch (IOException ex) {
// Reads whether or not the service has been started from the preferences
private boolean wasStarted() {
return mPrefs.getBoolean(PREF_STARTED, false);
// Sets whether or not the services has been started in the preferences.
private void setStarted(boolean started) {
mPrefs.edit().putBoolean(PREF_STARTED, started).commit();
mStarted =
private synchronized void start() {
log(&Starting service...&);
// Do nothing, if the service is already running.
if (mStarted == true) {
Log.w(TAG, &Attempt to start connection that is already active&);
// Establish an MQTT connection
connect();
// 向服务器定时发送心跳，一分钟一次
mRunFlag =
mTimer.schedule(new TimerTask() {
public void run() {
if (!mRunFlag) {
// this.cancel();
// PushService.this.stopSelf();
System.out.println(&run&);
if (isNetworkAvailable()) {
SharedPreferences pref = getSharedPreferences(
&client&, 0);
String MOBILE_NUM = pref.getString(&MOBILE_NUM&, &&);
HttpUtil.post(Constants.KEEPALIVE + &&mobile=&
+ MOBILE_NUM + &&online_flag=1&);
} catch (Exception e) {
e.printStackTrace();
// TODO: handle exception
}, 0, 60 * 1000);
// Register a connectivity listener
registerReceiver(mConnectivityChanged, new IntentFilter(
ConnectivityManager.CONNECTIVITY_ACTION));
private synchronized void stop() {
// Do nothing, if the service is not running.
if (mStarted == false) {
Log.w(TAG, &Attempt to stop connection not active.&);
// Save stopped state in the preferences
setStarted(false);
// Remove the connectivity receiver
unregisterReceiver(mConnectivityChanged);
// Any existing reconnect timers should be removed, since we explicitly
// stopping the service.
cancelReconnect();
// Destroy the MQTT connection if there is one
if (mConnection != null) {
mConnection.disconnect();
mConnection =
private synchronized void connect() {
log(&Connecting...&);
// Thread t = new Thread() {
// @Override
// public void run() {
// fetch the device ID from the preferences.
String deviceID = &GMCC/client/&
+ mPrefs.getString(PREF_DEVICE_ID, null);
// Create a new connection only if the device id is not NULL
mConnection = new MQTTConnection(MQTT_HOST, deviceID);
} catch (MqttException e) {
// Schedule a reconnect, if we failed to connect
log(&MqttException: &
+ (e.getMessage() != null ? e.getMessage() : &NULL&));
if (isNetworkAvailable()) {
scheduleReconnect(mStartTime);
setStarted(true);
// t.start();
// 向服务器定时发送心跳，一分钟一次
mRunFlag =
private synchronized void keepAlive() {
// Send a keep alive, if there is a connection.
if (mStarted == true && mConnection != null) {
mConnection.sendKeepAlive();
} catch (MqttException e) {
log(&MqttException: &
+ (e.getMessage() != null ? e.getMessage() : &NULL&), e);
mConnection.disconnect();
mConnection =
cancelReconnect();
// Schedule application level keep-alives using the AlarmManager
private void startKeepAlives() {
Intent i = new Intent();
i.setClass(this, PushService.class);
i.setAction(ACTION_KEEPALIVE);
PendingIntent pi = PendingIntent.getService(this, 0, i, 0);
AlarmManager alarmMgr = (AlarmManager) getSystemService(ALARM_SERVICE);
alarmMgr.setRepeating(AlarmManager.RTC_WAKEUP,
System.currentTimeMillis() + KEEP_ALIVE_INTERVAL,
KEEP_ALIVE_INTERVAL, pi);
// Remove all scheduled keep alives
private void stopKeepAlives() {
Intent i = new Intent();
i.setClass(this, PushService.class);
i.setAction(ACTION_KEEPALIVE);
PendingIntent pi = PendingIntent.getService(this, 0, i, 0);
AlarmManager alarmMgr = (AlarmManager) getSystemService(ALARM_SERVICE);
alarmMgr.cancel(pi);
// We schedule a reconnect based on the starttime of the service
public void scheduleReconnect(long startTime) {
// the last keep-alive interval
long interval = mPrefs.getLong(PREF_RETRY, INITIAL_RETRY_INTERVAL);
// Calculate the elapsed time since the start
long now = System.currentTimeMillis();
long elapsed = now - startT
// Set an appropriate interval based on the elapsed time since start
if (elapsed & interval) {
interval = Math.min(interval * 4, MAXIMUM_RETRY_INTERVAL);
interval = INITIAL_RETRY_INTERVAL;
log(&Rescheduling connection in & + interval + &ms.&);
// Save the new internval
mPrefs.edit().putLong(PREF_RETRY, interval).commit();
// Schedule a reconnect using the alarm manager.
Intent i = new Intent();
i.setClass(this, PushService.class);
i.setAction(ACTION_RECONNECT);
PendingIntent pi = PendingIntent.getService(this, 0, i, 0);
AlarmManager alarmMgr = (AlarmManager) getSystemService(ALARM_SERVICE);
alarmMgr.set(AlarmManager.RTC_WAKEUP, now + interval, pi);
// Remove the scheduled reconnect
public void cancelReconnect() {
Intent i = new Intent();
i.setClass(PushService.this, PushService.class);
i.setAction(ACTION_RECONNECT);
PendingIntent pi = PendingIntent.getService(PushService.this, 0, i, 0);
AlarmManager alarmMgr = (AlarmManager) getSystemService(ALARM_SERVICE);
alarmMgr.cancel(pi);
private synchronized void reconnectIfNecessary() {
log(&mStarted& + mStarted);
log(&mConnection& + mConnection);
if (mStarted == true && mConnection == null) {
log(&Reconnecting...&);
connect();
// This receiver listeners for network changes and updates the MQTT
// connection
// accordingly
private BroadcastReceiver mConnectivityChanged = new BroadcastReceiver() {
public void onReceive(Context context, final Intent intent) {
// Get network info
// Thread mReconnect = new Thread(){
// public void run() {
NetworkInfo info = (NetworkInfo) intent
.getParcelableExtra(ConnectivityManager.EXTRA_NETWORK_INFO);
// Is there connectivity?
boolean hasConnectivity = (info != null && info.isConnected()) ? true
log(&Connectivity changed: connected=& + hasConnectivity);
if (hasConnectivity) {
reconnectIfNecessary();
} else if (mConnection != null) {
// Thread cancelConn = new Thread(){
// public void run() {
// // if there no connectivity, make sure MQTT connection is
// destroyed
log(&cancelReconnect&);
mConnection.disconnect();
mConnection =
log(&cancelReconnect& + mConnection);
cancelReconnect();
// cancelConn.start();
// mReconnect.start();
// Display the topbar notification
private void showNotification(String text, Request request) {
Notification n = new Notification();
n.flags |= Notification.FLAG_SHOW_LIGHTS;
n.flags |= Notification.FLAG_AUTO_CANCEL;
n.defaults = Notification.DEFAULT_ALL;
n.icon = R.drawable.
n.when = System.currentTimeMillis();
Intent intent = new Intent();
Bundle bundle = new Bundle();
bundle.putSerializable(&request&, request);
bundle.putString(&currentTab&, &1&);
intent.putExtras(bundle);
intent.setClass(this, MainActivity.class);
intent.setAction(Intent.ACTION_MAIN);
intent.addCategory(Intent.CATEGORY_LAUNCHER);
intent.setFlags(Intent.FLAG_ACTIVITY_NEW_TASK
| Intent.FLAG_ACTIVITY_RESET_TASK_IF_NEEDED);
// Simply open the parent activity
PendingIntent pi = PendingIntent.getActivity(this, 0, intent, 0);
// Change the name of the notification here
n.setLatestEventInfo(this, NOTIF_TITLE, text, pi);
mNotifMan.notify(NOTIF_CONNECTED, n);
// Check if we are online
private boolean isNetworkAvailable() {
NetworkInfo info = mConnMan.getActiveNetworkInfo();
if (info == null) {
return info.isConnected();
// This inner class is a wrapper on top of MQTT client.
private class MQTTConnection implements MqttSimpleCallback {
IMqttClient mqttClient =
// Creates a new connection given the broker address and initial topic
public MQTTConnection(String brokerHostName, String initTopic)
throws MqttException {
// Create connection spec
String mqttConnSpec = &tcp://& + brokerHostName + &@&
+ MQTT_BROKER_PORT_NUM;
// Create the client and connect
mqttClient = MqttClient.createMqttClient(mqttConnSpec,
MQTT_PERSISTENCE);
String clientID = MQTT_CLIENT_ID + &/&
+ mPrefs.getString(PREF_DEVICE_ID, &&);
Log.d(TAG, &mqttConnSpec:& + mqttConnSpec + &
clientID:&
+ clientID);
mqttClient.connect(clientID, MQTT_CLEAN_START, MQTT_KEEP_ALIVE);
// register this client app has being able to receive messages
mqttClient.registerSimpleHandler(this);
// Subscribe to an initial topic, which is combination of client ID
// and device ID.
// initTopic = MQTT_CLIENT_ID + &/& + initT
subscribeToTopic(initTopic);
log(&Connection established to & + brokerHostName + & on topic &
+ initTopic);
// Save start time
mStartTime = System.currentTimeMillis();
// Star the keep-alives
startKeepAlives();
// Disconnect
public void disconnect() {
stopKeepAlives();
log(&stopKeepAlives&);
Thread t = new Thread() {
public void run() {
mqttClient.disconnect();
log(&mqttClient.disconnect();&);
} catch (MqttPersistenceException e) {
log(&MqttException&
+ (e.getMessage() != null ? e.getMessage()
: & NULL&), e);
t.start();
// } catch (MqttPersistenceException e) {
// log(&MqttException&
// + (e.getMessage() != null ? e.getMessage() : & NULL&),
* Send a request to the message broker to be sent messages published
* with the specified topic name. Wildcards are allowed.
private void subscribeToTopic(String topicName) throws MqttException {
if ((mqttClient == null) || (mqttClient.isConnected() == false)) {
// quick sanity check - don't try and subscribe if we don't have
// a connection
log(&Connection error& + &No connection&);
String[] topics = { topicName };
mqttClient.subscribe(topics, MQTT_QUALITIES_OF_SERVICE);
* Sends a message to the message broker, requesting that it be
* published to the specified topic.
private void publishToTopic(String topicName, String message)
throws MqttException {
if ((mqttClient == null) || (mqttClient.isConnected() == false)) {
// quick sanity check - don't try and publish if we don't have
// a connection
log(&No connection to public to&);
mqttClient.publish(topicName, message.getBytes(),
MQTT_QUALITY_OF_SERVICE, MQTT_RETAINED_PUBLISH);
* Called if the application loses it's connection to the message
public void connectionLost() throws Exception {
log(&Loss of connection& + &connection downed&);
stopKeepAlives();
// 取消定时发送心跳
mRunFlag =
// 向服务器发送请求，更改在线状态
// SharedPreferences pref = getSharedPreferences(&client&,0);
// String MOBILE_NUM=pref.getString(&MOBILE_NUM&, &&);
// HttpUtil.post(Constants.KEEPALIVE + &&mobile=&
// + MOBILE_NUM+&&online_flag=0&);
// null itself
mConnection =
if (isNetworkAvailable() == true) {
reconnectIfNecessary();
* Called when we receive a message from the message broker.
public void publishArrived(String topicName, byte[] payload, int qos,
boolean retained) throws MqttException {
// Show a notification
// synchronized (lock) {
String s = new String(payload);
Request request =
try {// 解析服务端推送过来的消息
request = XmlPaserTool.getMessage(new ByteArrayInputStream(s
.getBytes()));
// request=Constants.
} catch (Exception e) {
e.printStackTrace();
final Request mRequest =
DownloadInfo down = new DownloadInfo(mRequest);
down.setDownLoad(down);
downloadInfos.add(down);
sendUpdateBroast();
down.start();
showNotification(&您有一条新的消息!&, mRequest);
Log.d(PushService.TAG, s);
Log.d(PushService.TAG, mRequest.getMessageId());
// 再向服务端推送消息
new AdvancedCallbackHandler().sendMessage(MQTT_CLIENT_ID
+ &/keepalive&, &***********send message**********&);
public void sendKeepAlive() throws MqttException {
log(&Sending keep alive&);
// publish to a keep-alive topic
publishToTopic(MQTT_CLIENT_ID + &/keepalive&,
mPrefs.getString(PREF_DEVICE_ID, &&));
class AdvancedCallbackHandler {
IMqttClient mqttClient =
public final int[] QOS_VALUES = { 0, 0, 2, 0 };// 对应主题的消息级别
* 重新连接服务
private void connect() throws MqttException {
String mqttConnSpec = &tcp://& + MQTT_HOST + &@&
+ MQTT_BROKER_PORT_NUM;
// Create the client and connect
mqttClient = MqttClient.createMqttClient(mqttConnSpec,
MQTT_PERSISTENCE);
String clientID = MQTT_CLIENT_ID + &/&
+ mPrefs.getString(PREF_DEVICE_ID, &&);
mqttClient.connect(clientID, MQTT_CLEAN_START, MQTT_KEEP_ALIVE);
Log.d(TAG, &连接服务器，推送消息&);
Log.d(TAG, &**mqttConnSpec:& + mqttConnSpec + &
clientID:&
+ clientID);
Log.d(TAG, MQTT_CLIENT_ID + &/keepalive&);
// 增加心跳，保持网络通畅
mqttClient.publish(MQTT_CLIENT_ID + &/keepalive&,
&keepalive&.getBytes(), QOS_VALUES[0], true);
* 发送消息
* @param clientId
* @param messageId
public void sendMessage(String clientId, String message) {
if (mqttClient == null || !mqttClient.isConnected()) {
connect();
Log.d(TAG, &send message to & + clientId + &, message is &
+ message);
// 发布自己的消息
// mqttClient.publish(MQTT_CLIENT_ID + &/keepalive&,
// message.getBytes(), 0, false);
mqttClient.publish(MQTT_CLIENT_ID + &/keepalive&,
message.getBytes(), 0, false);
} catch (MqttException e) {
Log.d(TAG, e.getCause() + &&);
e.printStackTrace();
public String getPeople(String phone_number) {
String name = &&;
String[] projection = { ContactsContract.PhoneLookup.DISPLAY_NAME,
monDataKinds.Phone.NUMBER };
Log.d(TAG, &getPeople ---------&);
// 将自己添加到 msPeers 中
Cursor cursor = this.getContentResolver().query(
monDataKinds.Phone.CONTENT_URI,
projection, // Which columns to return.
monDataKinds.Phone.NUMBER + & = '&
+ phone_number + &'&, // WHERE clause.
null, // WHERE clause value substitution
null); // Sort order.
if (cursor == null) {
Log.d(TAG, &getPeople null&);
return &&;
Log.d(TAG, &getPeople cursor.getCount() = & + cursor.getCount());
if (cursor.getCount() & 0) {
cursor.moveToPosition(0);
// 取得联系人名字
int nameFieldColumnIndex = cursor
.getColumnIndex(ContactsContract.PhoneLookup.DISPLAY_NAME);
name = cursor.getString(nameFieldColumnIndex);
Log.i(&Contacts&, && + name + & .... & + nameFieldColumnIndex); // 这里提示
System.out.println(&联系人姓名：& + name);
return phone_
public void sendUpdateBroast() {
Intent intent = new Intent();
intent.setAction(&update&);
sendBroadcast(intent);
public void sendUpdateFinishBroast() {
Intent intent = new Intent();
intent.setAction(&updateFinishList&);
sendBroadcast(intent);
public class DownloadInfo extends Thread {
boolean runflag =
Request mR
public MessageBean bean =
DownloadInfo download =
MessageDetailDao dao = new MessageDetailDao(
PushService.this.getApplicationContext());
public synchronized void stopthread() {
public synchronized boolean getrunflag() {
DownloadInfo(Request mRequest) {
this.mRequest = mR
public void setDownLoad(DownloadInfo download) {
this.download =
public void run() {
File dir = new File(Constants.DOWNLOAD_PATH);
if (!dir.exists()) {
dir.mkdirs();
String filePath = Constants.DOWNLOAD_PATH
+ mRequest.getMessageId() + &.& + mRequest.getExt();
bean = new MessageBean();
bean.setPath(filePath);
bean.setStatus(0);
bean.setDate(mRequest.getTimestamp());
bean.setLayoutID(R.layout.list_say_he_item);
bean.setPhotoID(R.drawable.receive_ico);
bean.setMessage_key(mRequest.getMessageId());
bean.setPhone_number(mRequest.getReceiver());
bean.setAction(1);
String name = getPeople(mRequest.getSender());
bean.setName(name);
bean.setFileType(Integer.parseInt(mRequest.getCommand()));
if (mRequest.getCommand().equals(Request.TYPE_MUSIC)) {
bean.setMsgIco(R.drawable.music_temp);
bean.setText(name + &给你发送了音乐&);
mRequest.setBody(Base64.encodeToString(bean.getText()
.getBytes(), Base64.DEFAULT));
} else if (mRequest.getCommand().equals(Request.TYPE_CARD)) {
bean.setMsgIco(R.drawable.card_temp);
bean.setText(new String(Base64.decode(mRequest.getBody(),
Base64.DEFAULT)));
mRequest.setBody(Base64.encodeToString(bean.getText()
.getBytes(), Base64.DEFAULT));
} else if (mRequest.getCommand().equals(Request.TYPE_LBS)) {
bean.setMsgIco(R.drawable.address_temp);
bean.setText(new String(Base64.decode(mRequest.getBody(),
Base64.DEFAULT)));
mRequest.setBody(Base64.encodeToString(bean.getText()
.getBytes(), Base64.DEFAULT));
} else if (mRequest.getCommand().equals(Request.TYPE_PHOTO)) {
bean.setText(name + &向你发送了照片&);
bean.setMsgIco(-1);
} else if (mRequest.getCommand().equals(Request.TYPE_PIC)) {
bean.setText(name + &向你发送了图片&);
bean.setMsgIco(-1);
} else if (mRequest.getCommand().equals(Request.TYPE_SMS)) {
bean.setFileType(0);
if (!mRequest.getCommand().equals(Request.TYPE_CARD)
&& !mRequest.getCommand().equals(Request.TYPE_SMS)) {
String path = Constants.FILE_DOWNLOAD_URL
+ mRequest.getMessageId();
URL url = new URL(path);
HttpURLConnection hurlconn = (HttpURLConnection) url
.openConnection();// 基于HTTP协议的连接对象
hurlconn.setConnectTimeout(5000);// 请求超时时间 5s
hurlconn.setRequestMethod(&GET&);// 请求方式
hurlconn.connect();
long fileSize = hurlconn.getContentLength();
InputStream instream = hurlconn.getInputStream();
byte[] buffer = new byte[1024];
int len = 0;
int number = 0;
RandomAccessFile rasf = new RandomAccessFile(filePath,
while ((len = instream.read(buffer)) != -1) {// 开始下载文件
if (getrunflag() && progress & 100) {
rasf.seek(number);
rasf.write(buffer, 0, len);
progress = (((float) number) / fileSize) * 100;
// 发送广播，修改进度条进度
sendUpdateBroast();
this.interrupt();
if (number != fileSize) {// 取消下载，将已经缓存的未下载完成的文件删除
File file = new File(filePath);
if (file.exists())
file.delete();
PushService.downloadInfos.remove(download);
sendUpdateBroast();
instream.close();
PushService.downloadInfos.remove(download);
sendUpdateBroast();
} else {// 收到消息，将信息保存到数据库
PushService.downloadInfos.remove(download);
sendUpdateBroast();
// 将文件信息保存到数据库
dao.create(bean);
sendUpdateFinishBroast();
} catch (Exception e) {
PushService.downloadInfos.remove(download);
sendUpdateBroast();
e.printStackTrace();
public static ArrayList&DownloadInfo& downloadInfos = new ArrayList&DownloadInfo&();
public ArrayList&DownloadInfo& getDownloadInfos() {
return PushService.downloadI
public void setDownloadInfos(ArrayList&DownloadInfo& downloadInfos) {
PushService.downloadInfos = downloadI
接收者必须订阅发送者的TOPICS才能收到消息
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：126746次
积分：1064
积分：1064
排名：千里之外
原创：14篇
转载：16篇
评论：122条
(1)(1)(2)(2)(1)(13)(10)