Android Mqtt 消息推送使用 _android

知识的价值不在于占有，而在于使用。这篇文章主要讲述Android Mqtt 消息推送使用相关的知识，希望能为你提供帮助。
初始化SDK：

/** * 初始化SDK * * @param context context */ public void initSDK(Context context) { String clientId = String.valueOf(System.currentTimeMillis()+userId); mqttandroidClient = new MqttAndroidClient(mContext, serverUri, clientId); subscriptionTopics = new ArrayList< > (); mqttAndroidClient.setCallback(new MqttCallbackExtended() { @Override public void connectComplete(boolean reconnect, String serverURI) {if (reconnect) {Log.d(TAG, "Reconnected to : " + serverURI); // Because Clean Session is true, we need to re-subscribe //subscribeToTopic(); //publishMessage(); } else { Log.d(TAG, "Connected to: " + serverURI); } connectSuccess = true; subscribeToTopic(); }@Override public void connectionLost(Throwable cause) { connectSuccess = false; Log.e(TAG, "The Connection was lost." + cause.getLocalizedMessage()); }// THIS DOES NOT WORK! @Override public void messageArrived(String topic, MqttMessage message) throws Exception { Log.d(TAG, "Incoming message: " +topic+ new String(message.getPayload())); }@Override public void deliveryComplete(IMqttDeliveryToken token) {} }); }

连接远程服务：

/** * 连接远程服务 */ public void connectServer() { MqttConnectOptions mqttConnectOptions = new MqttConnectOptions(); mqttConnectOptions.setAutomaticReconnect(true); mqttConnectOptions.setCleanSession(false); try { //addToHistory("Connecting to " + serverUri); mqttAndroidClient.connect(mqttConnectOptions, null, new IMqttActionListener() { @Override public void onSuccess(IMqttToken asyncActionToken) { connectSuccess = true; DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions(); disconnectedBufferOptions.setBufferEnabled(true); disconnectedBufferOptions.setBufferSize(100); disconnectedBufferOptions.setPersistBuffer(false); disconnectedBufferOptions.setDeleteOldestMessages(false); mqttAndroidClient.setBufferOpts(disconnectedBufferOptions); }@Override public void onFailure(IMqttToken asyncActionToken, Throwable exception) { Log.e(TAG, "Failed to connect to: " + serverUri); exception.printStackTrace(); Log.d(TAG, "onFailure: " + exception.getCause()); connectSuccess = false; } }); } catch (MqttException ex) { ex.printStackTrace(); } }

获取订阅信息：

/** *获取订阅信息
*/

public void connectGateway(String gatewayId, String userId) {

//获取订阅信息 if (!subscriptionTopics.contains(gatewayId)) { subscriptionTopics.add(gatewayId); } Log.d(TAG, "pre sub topic: connect status=" + connectSuccess); Log.d(TAG, "subtopic " + subscriptionTopics); subscribeToTopic(); }

订阅mqtt消息：

/** * 订阅mqtt消息 */ private void subscribeToTopic() { try { if(subscriptionTopics.size()==0) return; String[] topics = new String[subscriptionTopics.size()]; subscriptionTopics.toArray(topics); int[] qoc = new int[topics.length]; IMqttMessageListener[] mqttMessageListeners = new IMqttMessageListener[topics.length]; for (int i = 0; i < topics.length; i++) { IMqttMessageListener mqttMessageListener = new IMqttMessageListener() { @Override public void messageArrived(String topic, MqttMessage message) throws Exception { // message Arrived!消息送达后做出的处理 Log.d(TAG, topic + " : " + new String(message.getPayload())); handleReceivedMessage(new String(message.getPayload()), topic); } }; mqttMessageListeners[i] = mqttMessageListener; Log.d(TAG, "subscribeToTopic: qoc= " + qoc[i]); } mqttAndroidClient.subscribe(topics, qoc, null, new IMqttActionListener() { @Override public void onSuccess(IMqttToken iMqttToken) { Log.d(TAG, "Subscribed!"); }@Override public void onFailure(IMqttToken iMqttToken, Throwable throwable) { Log.d(TAG, "Failed to subscribe"); } }, mqttMessageListeners); } catch (MqttException ex) { System.err.println("Exception whilst subscribing"); ex.printStackTrace(); }}

处理收到的消息：

private void handleReceivedMessage(String message, String gatewayId) { //可以发送一条广播通知程序 }

发送mqtt消息：

/** * 发送 mqtt 消息 * * @param publishMessage 要发送的信息的字符串 */ private void publishMessage(String publishMessage, String publishTopic) { try { publishTopic = userId + "/" + publishTopic; MqttMessage message = new MqttMessage(); message.setPayload(publishMessage.getBytes()); mqttAndroidClient.publish(publishTopic, message); Log.d(TAG, "publishMessage:Message Published \n" + publishTopic + ":" + message); if (!mqttAndroidClient.isConnected()) { Log.d(TAG, mqttAndroidClient.getBufferedMessageCount() + " messages in buffer."); } } catch (MqttException e) { System.err.println("Error Publishing: " + e.getMessage()); e.printStackTrace(); } }

没有封装的类：

public class SubscribeClient { private final static String CONNECTION_STRING = "tcp://mqtt地址:mqtt端口"; private final static boolean CLEAN_START = true; private final static short KEEP_ALIVE = 30; //低耗网络，但是又需要及时获取数据，心跳30s private final static String CLIENT_ID = "client1"; private final static String[] TOPICS = { //订阅信息 }; private final static int[] QOS_VALUES = {0, 0, 2, 0}; private MqttClient mqttClient = null; public SubscribeClient(String i) { try { mqttClient = new MqttClient(CONNECTION_STRING); SimpleCallbackHandler simpleCallbackHandler = new SimpleCallbackHandler(); mqttClient.registerSimpleHandler(simpleCallbackHandler); //注册接收消息方法 mqttClient.connect(CLIENT_ID + i, CLEAN_START, KEEP_ALIVE); mqttClient.subscribe(TOPICS, QOS_VALUES); //订阅接主题/** * 完成订阅后，可以增加心跳，保持网络通畅，也可以发布自己的消息 */mqttClient.publish(PUBLISH_TOPICS, "keepalive".getBytes(), QOS_VALUES[0], true); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } }/** * 简单回调函数，处理client接收到的主题消息 * * @author pig */ class SimpleCallbackHandler implements MqttSimpleCallback {/** * 当客户机和broker意外断开时触发 * 可以再此处理重新订阅 */ @Override public void connectionLost() throws Exception { // TODO Auto-generated method stub System.out.println("客户机和broker已经断开"); }/** * 客户端订阅消息后，该方法负责回调接收处理消息 */ @Override 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); }}/** * 高级回调 * * @author pig */ class AdvancedCallbackHandler implements MqttSimpleCallback {@Override public void connectionLost() throws Exception { // TODO Auto-generated method stub}@Override public void publishArrived(String arg0, byte[] arg1, int arg2, boolean arg3) throws Exception { // TODO Auto-generated method stub}}/** * @param args */ public static void main(String[] args) { // TODO Auto-generated method stub new SubscribeClient("" + i); }}

【Android Mqtt 消息推送使用】