MQTT实现消息推送

智能服装通过蓝牙连接手机，实现远程控制和信息推送。 #生活知识# #生活感悟# #科技生活变迁# #服装科技#

MQTT实现消息推送

MQTT实现消息接收(接收消息需实现MqttSimpleCallback接口并实现它的publishArrived方法)必须注册接收消息方法

mqttClient.registerSimpleHandler(simpleCallbackHandler);// 注册接收消息方法

和订阅接主题

mqttClient.subscribe(TOPICS, QOS_VALUES);// 订阅接主题

服务端：

package com.gmcc.kuchuan.business;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.ibm.mqtt.MqttClient;

import com.ibm.mqtt.MqttException;

import com.ibm.mqtt.MqttSimpleCallback;

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 = true;

private final static short KEEP_ALIVE = 30;

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 mqttClient;

public static MqttBroker getInstance() {

return instance;

}

private void connect() throws MqttException {

logger.info("connect to mqtt broker.");

mqttClient = new MqttClient(CONNECTION_STRING);

logger.info("***********register Simple Handler***********");

SimpleCallbackHandler simpleCallbackHandler = new SimpleCallbackHandler();

mqttClient.registerSimpleHandler(simpleCallbackHandler);

mqttClient.connect(CLIENT_ID, CLEAN_START, KEEP_ALIVE);

logger.info("***********subscribe receiver topics***********");

mqttClient.subscribe(TOPICS, QOS_VALUES);

logger.info("***********CLIENT_ID:" + CLIENT_ID);

mqttClient.publish("keepalive", "keepalive".getBytes(), QOS_VALUES[0],

true);

}

public void sendMessage(String clientId, String message) {

try {

if (mqttClient == null || !mqttClient.isConnected()) {

connect();

}

logger.info("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();

}

}

class SimpleCallbackHandler implements MqttSimpleCallback {

@Override

public void connectionLost() throws Exception {

System.out.println("客户机和broker已经断开");

}

@Override

public void publishArrived(String topicName, byte[] payload, int Qos,

boolean retained) throws Exception {

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.ByteArrayInputStream;

import java.io.File;

import java.io.IOException;

import java.io.InputStream;

import java.io.RandomAccessFile;

import java.net.HttpURLConnection;

import java.net.URL;

import java.util.ArrayList;

import java.util.Timer;

import java.util.TimerTask;

import android.app.AlarmManager;

import android.app.Notification;

import android.app.NotificationManager;

import android.app.PendingIntent;

import android.app.Service;

import android.content.BroadcastReceiver;

import android.content.Context;

import android.content.Intent;

import android.content.IntentFilter;

import android.content.SharedPreferences;

import android.database.Cursor;

import android.net.ConnectivityManager;

import android.net.NetworkInfo;

import android.os.Binder;

import android.os.Bundle;

import android.os.IBinder;

import android.provider.ContactsContract;

import android.util.Log;

import com.ibm.mqtt.IMqttClient;

import com.ibm.mqtt.MqttClient;

import com.ibm.mqtt.MqttException;

import com.ibm.mqtt.MqttPersistence;

import com.ibm.mqtt.MqttPersistenceException;

import com.ibm.mqtt.MqttSimpleCallback;

public class PushService extends Service {

private MyBinder mBinder = new MyBinder();

public static final String TAG = "PushService";

private static final String MQTT_HOST = "120.197.230.53";

private static int MQTT_BROKER_PORT_NUM = 9901;

private static MqttPersistence MQTT_PERSISTENCE = null;

private static boolean MQTT_CLEAN_START = true;

private static short MQTT_KEEP_ALIVE = 60 * 15;

private static int[] MQTT_QUALITIES_OF_SERVICE = { 0 };

private static int MQTT_QUALITY_OF_SERVICE = 0;

private static boolean MQTT_RETAINED_PUBLISH = false;

public static String MQTT_CLIENT_ID = "client";

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";

private ConnectionLog mLog;

private ConnectivityManager mConnMan;

private NotificationManager mNotifMan;

private boolean mStarted;

private static final long KEEP_ALIVE_INTERVAL = 1000 * 60 * 28;

private static final long INITIAL_RETRY_INTERVAL = 1000 * 10;

private static final long MAXIMUM_RETRY_INTERVAL = 1000 * 60 * 30;

private SharedPreferences mPrefs;

public static final String PREF_STARTED = "isStarted";

public static final String PREF_DEVICE_ID = "deviceID";

public static final String PREF_RETRY = "retryInterval";

public static String NOTIF_TITLE = "client";

private static final int NOTIF_CONNECTED = 0;

private MQTTConnection mConnection;

private long mStartTime;

boolean mShowFlag = true;

public static Context ctx;

private boolean mRunFlag = true;

Timer mTimer = new Timer();

public static void actionStart(Context ctx) {

Intent i = new Intent(ctx, PushService.class);

i.setAction(ACTION_START);

ctx.startService(i);

PushService.ctx = ctx;

}

public static void actionStop(Context ctx) {

Intent i = new Intent(ctx, PushService.class);

i.setAction(ACTION_STOP);

ctx.startService(i);

}

public static void actionPing(Context ctx) {

Intent i = new Intent(ctx, PushService.class);

i.setAction(ACTION_KEEPALIVE);

ctx.startService(i);

}

@Override

public void onCreate() {

super.onCreate();

log("Creating service");

mStartTime = System.currentTimeMillis();

try {

mLog = new ConnectionLog();

Log.i(TAG, "Opened log at " + mLog.getPath());

} catch (IOException e) {

Log.e(TAG, "Failed to open log", e);

}

mPrefs = getSharedPreferences(TAG, MODE_PRIVATE);

mConnMan = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);

mNotifMan = (NotificationManager) getSystemService(NOTIFICATION_SERVICE);

handleCrashedService();

}

private void handleCrashedService() {

if (wasStarted() == true) {

log("Handling crashed service...");

stopKeepAlives();

start();

}

}

@Override

public void onDestroy() {

log("Service destroyed (started=" + mStarted + ")");

if (mStarted == true) {

stop();

}

try {

if (mLog != null)

mLog.close();

} catch (IOException e) {

}

}

@Override

public void onStart(Intent intent, int startId) {

super.onStart(intent, startId);

log("Service started with intent=" + intent);

if (intent == null) {

return;

}

if (intent.getAction().equals(ACTION_STOP) == true) {

stop();

stopSelf();

} else if (intent.getAction().equals(ACTION_START) == true) {

start();

} 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.this;

}

}

@Override

public IBinder onBind(Intent intent) {

return mBinder;

}

private void log(String message) {

log(message, null);

}

private void log(String message, Throwable e) {

if (e != null) {

Log.e(TAG, message, e);

} else {

Log.i(TAG, message);

}

if (mLog != null) {

try {

mLog.println(message);

} catch (IOException ex) {

}

}

}

private boolean wasStarted() {

return mPrefs.getBoolean(PREF_STARTED, false);

}

private void setStarted(boolean started) {

mPrefs.edit().putBoolean(PREF_STARTED, started).commit();

mStarted = started;

}

private synchronized void start() {

log("Starting service...");

if (mStarted == true) {

Log.w(TAG, "Attempt to start connection that is already active");

return;

}

connect();

mRunFlag = true;

mTimer.schedule(new TimerTask() {

@Override

public void run() {

if (!mRunFlag) {

return;

}

System.out.println("run");

try {

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();

}

}

}, 0, 60 * 1000);

registerReceiver(mConnectivityChanged, new IntentFilter(

ConnectivityManager.CONNECTIVITY_ACTION));

}

private synchronized void stop() {

if (mStarted == false) {

Log.w(TAG, "Attempt to stop connection not active.");

return;

}

setStarted(false);

unregisterReceiver(mConnectivityChanged);

cancelReconnect();

if (mConnection != null) {

mConnection.disconnect();

mConnection = null;

}

}

private synchronized void connect() {

log("Connecting...");

String deviceID = "GMCC/client/"

+ mPrefs.getString(PREF_DEVICE_ID, null);

try {

mConnection = new MQTTConnection(MQTT_HOST, deviceID);

} catch (MqttException e) {

log("MqttException: "

+ (e.getMessage() != null ? e.getMessage() : "NULL"));

if (isNetworkAvailable()) {

scheduleReconnect(mStartTime);

}

}

setStarted(true);

mRunFlag = true;

}

private synchronized void keepAlive() {

try {

if (mStarted == true && mConnection != null) {

mConnection.sendKeepAlive();

}

} catch (MqttException e) {

log("MqttException: "

+ (e.getMessage() != null ? e.getMessage() : "NULL"), e);

mConnection.disconnect();

mConnection = null;

cancelReconnect();

}

}

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);

}

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);

}

public void scheduleReconnect(long startTime) {

long interval = mPrefs.getLong(PREF_RETRY, INITIAL_RETRY_INTERVAL);

long now = System.currentTimeMillis();

long elapsed = now - startTime;

if (elapsed < interval) {

interval = Math.min(interval * 4, MAXIMUM_RETRY_INTERVAL);

} else {

interval = INITIAL_RETRY_INTERVAL;

}

log("Rescheduling connection in " + interval + "ms.");

mPrefs.edit().putLong(PREF_RETRY, interval).commit();

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);

}

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();

}

}

private BroadcastReceiver mConnectivityChanged = new BroadcastReceiver() {

@Override

public void onReceive(Context context, final Intent intent) {

NetworkInfo info = (NetworkInfo) intent

.getParcelableExtra(ConnectivityManager.EXTRA_NETWORK_INFO);

boolean hasConnectivity = (info != null && info.isConnected()) ? true

: false;

log("Connectivity changed: connected=" + hasConnectivity);

if (hasConnectivity) {

reconnectIfNecessary();

} else if (mConnection != null) {

log("cancelReconnect");

mConnection.disconnect();

mConnection = null;

log("cancelReconnect" + mConnection);

cancelReconnect();

}

}

};

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.ico;

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);

PendingIntent pi = PendingIntent.getActivity(this, 0, intent, 0);

n.setLatestEventInfo(this, NOTIF_TITLE, text, pi);

mNotifMan.notify(NOTIF_CONNECTED, n);

}

private boolean isNetworkAvailable() {

NetworkInfo info = mConnMan.getActiveNetworkInfo();

if (info == null) {

return false;

}

return info.isConnected();

}

private class MQTTConnection implements MqttSimpleCallback {

IMqttClient mqttClient = null;

public MQTTConnection(String brokerHostName, String initTopic)

throws MqttException {

String mqttConnSpec = "tcp://" + brokerHostName + "@"

+ MQTT_BROKER_PORT_NUM;

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);

mqttClient.registerSimpleHandler(this);

subscribeToTopic(initTopic);

log("Connection established to " + brokerHostName + " on topic "

+ initTopic);

mStartTime = System.currentTimeMillis();

startKeepAlives();

}

public void disconnect() {

stopKeepAlives();

log("stopKeepAlives");

Thread t = new Thread() {

public void run() {

try {

mqttClient.disconnect();

log("mqttClient.disconnect();");

} catch (MqttPersistenceException e) {

log("MqttException"

+ (e.getMessage() != null ? e.getMessage()

: " NULL"), e);

}

};

};

t.start();

}

private void subscribeToTopic(String topicName) throws MqttException {

if ((mqttClient == null) || (mqttClient.isConnected() == false)) {

log("Connection error" + "No connection");

} else {

String[] topics = { topicName };

mqttClient.subscribe(topics, MQTT_QUALITIES_OF_SERVICE);

}

}

private void publishToTopic(String topicName, String message)

throws MqttException {

if ((mqttClient == null) || (mqttClient.isConnected() == false)) {

log("No connection to public to");

} else {

mqttClient.publish(topicName, message.getBytes(),

MQTT_QUALITY_OF_SERVICE, MQTT_RETAINED_PUBLISH);

}

}

public void connectionLost() throws Exception {

log("Loss of connection" + "connection downed");

stopKeepAlives();

mRunFlag = false;

mConnection = null;

if (isNetworkAvailable() == true) {

reconnectIfNecessary();

}

}

public void publishArrived(String topicName, byte[] payload, int qos,

boolean retained) throws MqttException {

String s = new String(payload);

Request request = null;

try {

request = XmlPaserTool.getMessage(new ByteArrayInputStream(s

.getBytes()));

} catch (Exception e) {

e.printStackTrace();

}

final Request mRequest = request;

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");

publishToTopic(MQTT_CLIENT_ID + "/keepalive",

mPrefs.getString(PREF_DEVICE_ID, ""));

}

}

class AdvancedCallbackHandler {

IMqttClient mqttClient = null;

public final int[] QOS_VALUES = { 0, 0, 2, 0 };

private void connect() throws MqttException {

String mqttConnSpec = "tcp://" + MQTT_HOST + "@"

+ MQTT_BROKER_PORT_NUM;

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);

}

public void sendMessage(String clientId, String message) {

try {

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);

} catch (MqttException e) {

Log.d(TAG, e.getCause() + "");

e.printStackTrace();

}

}

}

public String getPeople(String phone_number) {

String name = "";

String[] projection = { ContactsContract.PhoneLookup.DISPLAY_NAME,

ContactsContract.CommonDataKinds.Phone.NUMBER };

Log.d(TAG, "getPeople ---------");

Cursor cursor = this.getContentResolver().query(

ContactsContract.CommonDataKinds.Phone.CONTENT_URI,

projection,

ContactsContract.CommonDataKinds.Phone.NUMBER + " = '"

+ phone_number + "'",

null,

null);

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 name;

}

return phone_number;

}

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 = true;

Request mRequest;

public float progress;

public MessageBean bean = null;

DownloadInfo download = null;

MessageDetailDao dao = new MessageDetailDao(

PushService.this.getApplicationContext());

public synchronized void stopthread() {

runflag = false;

}

public synchronized boolean getrunflag() {

return runflag;

}

DownloadInfo(Request mRequest) {

this.mRequest = mRequest;

}

public void setDownLoad(DownloadInfo download) {

this.download = download;

}

@Override

public void run() {

try {

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();

hurlconn.setConnectTimeout(5000);

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,

"rwd");

while ((len = instream.read(buffer)) != -1) {

if (getrunflag() && progress < 100) {

rasf.seek(number);

number += len;

rasf.write(buffer, 0, len);

progress = (((float) number) / fileSize) * 100;

sendUpdateBroast();

} else {

this.interrupt();

if (number != fileSize) {

File file = new File(filePath);

if (file.exists())

file.delete();

}

PushService.downloadInfos.remove(download);

sendUpdateBroast();

return;

}

}

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.downloadInfos;

}

public void setDownloadInfos(ArrayList<DownloadInfo> downloadInfos) {

PushService.downloadInfos = downloadInfos;

}

}

 ps：

接收者必须订阅发送者的TOPICS才能收到消息

网址：MQTT实现消息推送 https://www.yuejiaxmz.com/news/view/429726

相关内容

android 消息推送 push
消息推送架构介绍
推送消息&推送机制（推送机制如何规划设置）
PUSH消息推送的实现方法
【APP消息推送 】为安卓APP实现消息推送能力【基础功能】
SpringBoot整合钉钉实现消息推送
移动推送服务：实现消息通知和个性化推送
服务号推送：轻松实现无限制消息推送，模板消息不再受限
缩短推送时间—这个Push消息推送平台帮你实现！
微信小程序实现订阅消息推送的实现步骤

随便看看