android中mqtt怎么做

MQTT连接建立的代码(SSL方式)

[java] view plain copy

public static void connect(Driver driver) {

ServerConfig serverConfig = UserModule.Instance.getServerConfig()

MqttConnectOptions conOpt = new MqttConnectOptions()

try {

SSLContext sslContext

KeyStore ts = KeyStore.getInstance("BKS")

ts.load(context.getResources().openRawResource(R.raw.test_cert),

"123456".toCharArray())

TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509")

tmf.init(ts)

TrustManager[] tm = tmf.getTrustManagers()

sslContext = SSLContext.getInstance("TLS")

sslContext.init(null, tm, null)

SocketFactory factory = sslContext.getSocketFactory()

conOpt.setSocketFactory(factory)

} catch (Exception e) {

e.printStackTrace()

}

<span style="white-space:pre"> </span>

[java] view plain copy

<span style="white-space:pre"> </span>//paho库得

Iterator<Map.Entry<String, Connection>>it = Connections

.getInstance(context).getConnections().entrySet().iterator()

while (it.hasNext()) {

MqttClientAndroidService detectClient = it.next().getValue()

.getClient()

try {

detectClient.disconnect()

} catch (MqttException e) {

e.printStackTrace()

}

it.remove()

}

// The basic client information

MqttClientAndroidService client

client = Connections.getInstance(context).createClient(context,

serverConfig.getUri(), serverConfig.clientId)

Integer qos = Integer.parseInt(context.getResources().getString(

R.string.qos))

Boolean retained = Boolean.parseBoolean(context.getResources()

.getString(R.string.retained))

// connection options

int timeout = Integer.parseInt(context.getResources().getString(

(R.string.timeout)))

int keepalive = Integer.parseInt(context.getResources().getString(

R.string.keepalive))

Connection connection = new Connection(serverConfig.getClientHandle(),

serverConfig.clientId, serverConfig.server,

Integer.parseInt(serverConfig.port), context, client,

serverConfig.ssl)

// connection.registerChangeListener(changeListener)

// connect client

String[] actionArgs = new String[1]

actionArgs[0] = serverConfig.clientId

connection.changeConnectionStatus(ConnectionStatus.CONNECTING)

boolean cleanSession = false

conOpt.setCleanSession(cleanSession)

conOpt.setConnectionTimeout(timeout)

conOpt.setKeepAliveInterval(keepalive)

if (!TextUtils.isEmpty(serverConfig.user)) {

conOpt.setUserName(serverConfig.user)

}

if (!TextUtils.isEmpty(serverConfig.pwd)) {

conOpt.setPassword(serverConfig.pwd.toCharArray())

}

// conOpt.setPassword("1111".toCharArray())

final ActionListener callback = new ActionListener(context,

ActionListener.Action.CONNECT, driver.getMqttUtilsCallback(),

serverConfig.getClientHandle(), actionArgs)

boolean doConnect = true

String message = ActivityConstants.message

String topic = ActivityConstants.topic

if ((!TextUtils.isEmpty(message) || !TextUtils.isEmpty(topic))) {

// need to make a message since last will is set

try {

conOpt.setWill(topic, message.getBytes(), qos.intValue(),

retained.booleanValue())

} catch (Exception e) {

e.printStackTrace()

doConnect = false

callback.onFailure(null, e)

}

}

client.setCallback(new MqttCallbackHandler(context, serverConfig

.getClientHandle(), driver))

connection.addConnectionOptions(conOpt)

Connections.getInstance(context).addConnection(connection)

if (doConnect) {

try {

client.connect(conOpt, null, callback)

} catch (MqttException e) {

Log.e(TAG, "MqttException Occured", e)

}

}

}

发布(publish)代码

[java] view plain copy

public static void publish(String clientHandle, String topic,

JSONObject jsonObj, int qos) {

MqttClientAndroidService client = Connections.getInstance(context)

.getConnection(clientHandle).getClient()

if (!isConnected(context)) {

try {

client.connect()

} catch (MqttException e) {

e.printStackTrace()

}

Toast.makeText(context, "please try again", Toast.LENGTH_SHORT)

.show()

return

}

if (topic == null) {

Toast.makeText(context, "can not get other's identity for now",

Toast.LENGTH_SHORT).show()

return

}

String[] args = new String[2]

if (jsonObj.optLong("time") != 0) {

args[0] = String.valueOf(jsonObj.optLong("time"))

args[1] = topic

}

Boolean retained = Boolean.parseBoolean(context.getResources()

.getString(R.string.retained))

try {

client.publish(topic, jsonObj.toString().getBytes(), qos, retained,

null, new ActionListener(context, Action.PUBLISH, null,

clientHandle, args))

} catch (MqttSecurityException e) {

Log.e(TAG,

"Failed to publish a messged from the client with the handle "

+ clientHandle, e)

} catch (MqttException e) {

Log.e(TAG,

"Failed to publish a messged from the client with the handle "

+ clientHandle, e)

}

}

订阅(subscribe)代码

[java] view plain copy

public static void subscribe(MqttUtilsCallback mqttUtilsCallback,

String clientHandle, String topic) {

MqttClientAndroidService client = Connections.getInstance(context)

.getConnection(clientHandle).getClient()

if (client == null || (client != null &&!client.isConnected())) {

Toast.makeText(context, "please connect to server first",

Toast.LENGTH_SHORT).show()

return

}

if (TextUtils.isEmpty(topic)) {

topic = "hello"

}

String[] topics = new String[1]

topics[0] = topic

try {

client.subscribe(topic, 1, null, new ActionListener(context,

ActionListener.Action.SUBSCRIBE, mqttUtilsCallback,

clientHandle, topics))

} catch (MqttSecurityException e) {

Log.e(TAG, "Failed to subscribe to" + topic

+ " the client with the handle " + clientHandle, e)

} catch (MqttException e) {

Log.e(TAG, "Failed to subscribe to" + topic

+ " the client with the handle " + clientHandle, e)

}

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上。好比你给好友发送一封电子邮件，发送完成后你可以去做别的事情，收件人也不必立刻响应，可以在自己有空的时候查看邮件，是一个典型的异者弯步发布/订阅场景。而另一种典型的同步请求/回答场景，可以用接打电话的场景来类比。

MQTT的设计遵循以下的原则：

为了满足不同的场景，MQTT支持三种不同级别的服务质量（Quality of Service，QoS）为不同场景提供消息可靠性：

MQTT拥有14种不同的消息类型：

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

MQTT会构建底层网络传输：它将建立客户端到服务器的连接，提供两者之间的一个有序的、无损的樱嫌蔽、基于字节流的双向传输。

当应用数据通过MQTT网络发送时，MQTT会把与之相关的服务质量（QoS）和主题名（Topic）相关连。

一个使用MQTT协议的应用程序或者设备，它总是建立到服务器的网络连接。客户端可以：

MQTT服务器以称为"消息代理"（Broker），可以是一个应用程序或一台设备。它是位于消息发布者和订阅者之间，它可以：

订阅包含主题筛选器（Topic Filter）和最大服务质量（QoS）。订阅会与一个会话（Session）关联。一个会话可以包含多个订阅。每一个会话中的每个订阅都有一个不同的主题筛选器。

每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。

连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。

一个对主题名通配符筛选器，在订阅表达式脊州中使用，表示订阅所匹配到的多个主题。

消息订阅者所具体接收的内容。

MQTT协议中定义了一些方法（也被称为动作），来于表示对确定资源所进行 *** 作。这个资源可以代表预先存在的数据或动态生成数据，这取决于服务器的实现。通常来说，资源指服务器上的文件或输出。主要方法有：

MQTT协议实现android消息推送,我想每个Android开发人员对它应该都是比较熟悉的。MQ 遥测传输 (MQTT) 是轻量级基于代理的发布/订阅的消息传输协议,设计思想是开放、简单、轻量、易于实现。这些特点使它适用于受限环境。例如,但不仅限于此:网络代价昂贵,带宽低、不可靠。在嵌入设备中运行,处理器和内存资源有限。该协议的特点有:使用发布/订阅消息模式,提供一对多的消息发布,解除应用程序耦合。对负载内容屏蔽的消息传输。

使用 TCP/IP 提供网络乎汪山连接。有三种消息发布服务质量:“至多一次”,消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况,环境传感器数据,丢失一次读记录无所谓,因为不久后还会有第二次发送。“至少一次”,确保消息到达,但消息重复可能会发生。“只有一次”,确保消息到达一次。这一级别可用于如下情况,在计费系统中,消息重复或丢失会导致不正确的结果。小型传输,开销很小(固定长度的头部岁中是 2 字节),协议交换最小化,以降低网络流量。使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制MQTT最简单的使用包括两种,一种是发消息,一种是订阅消息。

发消息就是向一个固定IP地址的某个主题发送消息(publish)订阅消息是向服务器端订阅某些主题,当其他客户端向服务器的这个主题广播消息时,那么所有订阅这个主题的客户端就都能收到了MQTT是一项消息传递技术,由IBM再2001年发布。总结一下,机制就是使用一个代理服务器message broker,客户端client连接上这个服务器,然后告诉服务器说,我可以接收哪些类型的消陵州息,同时,client也可以发布自己的消息,这些消息根据协议的内容,可以被其他client获取。只要手机客户端,连上服务器,然后就可以接收和发布消息了,不用自己写socket什么了,低带宽,低耗电量,代码量也少,很简单吧。

package com.pig.test.mqtt002 003 import com.ibm.mqtt.MqttClient004 import com.ibm.mqtt.MqttException005 import com.ibm.mqtt.MqttSimpleCallback006 007 public class SubscribeClient {008 private final static String CONNECTION_STRING = "tcp://192.168.1.60:1883"009 private final static boolean CLEAN_START = true010 private final static short KEEP_ALIVE = 30//低耗网络,但是又需要及时获取数据,心跳30s011 private final static String CLIENT_ID = "client1"012 private final static String[] TOPICS = {013 "Test/TestTopics/Topic1",014 "Test/TestTopics/Topic2",015 "Test/TestTopics/Topic3",016 "tokudu/client1"017 }018 private final static int[] QOS_VALUES = {0, 0, 2, 0}019 020 //////////////////021 private MqttClient mqttClient = null022 023 public SubscribeClient(String i){024 try {025mqttClient = new MqttClient(CONNECTION_STRING)026SimpleCallbackHandler simpleCallbackHandler = new SimpleCallbackHandler()027mqttClient.registerSimpleHandler(simpleCallbackHandler)//注册接收消息方法028mqttClient.connect(CLIENT_ID+i, CLEAN_START, KEEP_ALIVE)029mqttClient.subscribe(TOPICS, QOS_VALUES)//订阅接主题030 031/**032 * 完成订阅后,可以增加心跳,保持网络通畅,也可以发布自己的消息033 */034 035mqttClient.publish(PUBLISH_TOPICS, "keepalive".getBytes(), QOS_VALUES[0], true)036 037 } catch (MqttException e) {038// TODO Auto-generated catch block039e.printStackTrace()040 }041 }042 043 /**044 * 简单回调函数,处理client接收到的主题消息045 * @author pig046 *047 */048 class SimpleCallbackHandler implements MqttSimpleCallback{

049 050 /**051

* 当客户机和broker意外断开时触发052* 可以再此处理重新订阅053*/054 @Override055 public void connectionLost() throws Exception {056// TODO Auto-generated method stub057System.out.println("客户机和broker已经断开")058 }059 060 /**061

* 客户端订阅消息后,该方法负责回调接收处理消息062

*/063 @Override064 public void publishArrived(String topicName, byte[] payload, int Qos, booleanretained) throws Exception {065// TODO Auto-generated method stub066System.out.println("订阅主题: " + topicName)067System.out.println("消息数据: " + new String(payload))068System.out.println("消息级别(0,1,2): " + Qos)069System.out.println("是否是实时发送的消息(false=实时,true=服务器上保留的最后消息): " + retained)070 }071072 }073 074 /**075

* 高级回调076

* @author pig077

*078 */079 class AdvancedCallbackHandler implements MqttSimpleCallback{080 081 @Override082 public void connectionLost() throws Exception {083// TODO Auto-generated method stub084 085 }086 087 @Override088 public void publishArrived(String arg0, byte[] arg1, int arg2,089 boolean arg3) throws Exception {090// TODO Auto-generated method stub091 092 }093094 }095 096 /**097

* @param args098 */099 public static void main(String[] args) {100 // TODO Auto-generated method stub101new SubscribeClient("" + i)102 103 }104 105 }

---