探索 IBM Bluemix 上的 MQTT 和物联网服务

构建您的应用程序所需的准备工作

“MQTT 是一个基于 TCP/IP 协议的简单、轻量型的发布/订阅消息协议，是新兴 IoT 领域的理想之选。”

第 1 步. 设置 Bluemix IoT 服务

在仪表板上创建一个应用程序

添加 IoT 服务

启动 IoT 服务控制台

注册设备

注册应用程序

第 2 步. 创建一个设备端程序

连接到 IoT 服务

	public void connect(String serverHost, String clientId, String authmethod,
			String authtoken, boolean isSSL) {
		// check if client is already connected
		if (!isMqttConnected()) {
			String connectionUri = null;
			
			//tcp://<org-id>.messaging.internetofthings.ibmcloud.com:1883
			//ssl://<org-id>.messaging.internetofthings.ibmcloud.com:8883
			if (isSSL) {
				connectionUri = "ssl://" + serverHost + ":" + DEFAULT_SSL_PORT;
			} else {
				connectionUri = "tcp://" + serverHost + ":" + DEFAULT_TCP_PORT;
			}
                       
			if (client != null) {
				try {
					client.disconnect();
				} catch (MqttException e) {
					e.printStackTrace();
				}
				client = null;
			}
                       
			try {
				client = new MqttClient(connectionUri, clientId);
			} catch (MqttException e) {
				e.printStackTrace();
			}
                       
			client.setCallback(this);
                       
			// create MqttConnectOptions and set the clean session flag
			MqttConnectOptions options = new MqttConnectOptions();
			options.setCleanSession(true);
                      
			options.setUserName(authmethod);
			options.setPassword(authtoken.toCharArray());
                      
			//If SSL is used, do not forget to use TLSv1.2
			if (isSSL) {
				java.util.Properties sslClientProps = new java.util.Properties();
				sslClientProps.setProperty("com.ibm.ssl.protocol", "TLSv1.2");
				options.setSSLProperties(sslClientProps);
			}
                      
			try {
				// connect
				client.connect(options);
				System.out.println("Connected to " + connectionUri);
			} catch (MqttException e) {
				e.printStackTrace();
			}
		}
	}

               //Read properties from the conf file
		Properties props = MqttUtil.readProperties("device.conf");
               
		String org = props.getProperty("org");
		String id = props.getProperty("deviceid");
		String authmethod = "use-token-auth";
		String authtoken = props.getProperty("token");
		//isSSL property
		String sslStr = props.getProperty("isSSL");
		boolean isSSL = false;
		if (sslStr.equals("T")) {
			isSSL = true;
		}
               
		System.out.println("org: " + org);
		System.out.println("id: " + id);
		System.out.println("authmethod: " + authmethod);
		System.out.println("authtoken: " + authtoken);
		System.out.println("isSSL: " + isSSL);
               
		String serverHost = org + MqttUtil.SERVER_SUFFIX;
               
		//Format: d:<orgid>:<type-id>:<divice-id>
		String clientId = "d:" + org + ":" + MqttUtil.DEFAULT_DEVICE_TYPE + ":"
				+ id;
		handler = new DeviceMqttHandler();
		handler.connect(serverHost, clientId, authmethod, authtoken, isSSL);

发布事件

                      //Format the Json String
			JSONObject contObj = new JSONObject();
			JSONObject jsonObj = new JSONObject();
			try {
				contObj.put("count", count);
				contObj.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
						.format(new Date()));
				jsonObj.put("d", contObj);
			} catch (JSONException e1) {
				e1.printStackTrace();
			}
                       
			System.out.println("Send count as " + count);
                      
			//Publish device events to the app
			//iot-2/evt/<event-id>/fmt/<format>
			handler.publish("iot-2/evt/" + MqttUtil.DEFAULT_EVENT_ID
					+ "/fmt/json", jsonObj.toString(), false, 0);

{
    "d": {
        "count": 3,
        "time": "2014-12-30 16:14:59"
    }
}

订阅命令

               //Subscribe the Command events
		//iot-2/cmd/<cmd-type>/fmt/<format-id>
		handler.subscribe("iot-2/cmd/" + MqttUtil.DEFAULT_CMD_ID + "/fmt/json",
				0);

               public void messageArrived(String topic, MqttMessage mqttMessage)
				throws Exception {
			super.messageArrived(topic, mqttMessage);
			
			//Check whether the event is a command event from app
			if (topic.equals("iot-2/cmd/" + MqttUtil.DEFAULT_CMD_ID
					+ "/fmt/json")) {
				String payload = new String(mqttMessage.getPayload());
				JSONObject jsonObject = new JSONObject(payload);
				String cmd = jsonObject.getString("cmd");
				//Reset the count
				if (cmd != null && cmd.equals("reset")) {
					int resetcount = jsonObject.getInt("count");
					count = resetcount;
					System.out.println("Count is reset to " + resetcount);
				}
			}
		}

第 3 步. 创建应用程序端程序

连接到 IoT 服务

               //Read properties from the conf file
		Properties props = MqttUtil.readProperties("app.conf");
               
		String org = props.getProperty("org");
		String id = props.getProperty("appid");
		String authmethod = props.getProperty("key");
		String authtoken = props.getProperty("token");
		//isSSL property
		String sslStr = props.getProperty("isSSL");
		boolean isSSL = false;
		if (sslStr.equals("T")) {
			isSSL = true;
		}
		
		System.out.println("org: " + org);
		System.out.println("id: " + id);
		System.out.println("authmethod: " + authmethod);
		System.out.println("authtoken" + authtoken);
		System.out.println("isSSL: " + isSSL);
               
		//Format: a:<orgid>:<app-id>
		String clientId = "a:" + org + ":" + id;
		String serverHost = org + MqttUtil.SERVER_SUFFIX;
               
		handler = new AppMqttHandler();
		handler.connect(serverHost, clientId, authmethod, authtoken, isSSL);

订阅事件

               handler.subscribe("iot-2/type/" + MqttUtil.DEFAULT_DEVICE_TYPE
				+ "/id/+/mon", 0);
		
		//Subscribe Device Events
		//iot-2/type/<type-id>/id/<device-id>/evt/<event-id>/fmt/<format-id>
		handler.subscribe("iot-2/type/" + MqttUtil.DEFAULT_DEVICE_TYPE
				+ "/id/+/evt/" + MqttUtil.DEFAULT_EVENT_ID + "/fmt/json", 0);

             public void messageArrived(String topic, MqttMessage mqttMessage)
				throws Exception {
			super.messageArrived(topic, mqttMessage);
                       
			Matcher matcher = pattern.matcher(topic);
			if (matcher.matches()) {
				String deviceid = matcher.group(1);
				String payload = new String(mqttMessage.getPayload());
				
				//Parse the payload in Json Format
				JSONObject jsonObject = new JSONObject(payload);
				JSONObject contObj = jsonObject.getJSONObject("d");
				int count = contObj.getInt("count");
				System.out.println("Receive count " + count + " from device "
						+ deviceid);
                         
				//If count >=4, start a new thread to reset the count
				if (count >= 4) {
					new ResetCountThread(deviceid, 0).start();
				}
			}
		}

发布命令

                      JSONObject jsonObj = new JSONObject();
			try {
				jsonObj.put("cmd", "reset");
				jsonObj.put("count", count);
				jsonObj.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
						.format(new Date()));
			} catch (JSONException e) {
				e.printStackTrace();
			}
			System.out.println("Reset count for device " + deviceid);
			
			//Publish command to one specific device
			//iot-2/type/<type-id>/id/<device-id>/cmd/<cmd-id>/fmt/<format-id>
			handler.publish("iot-2/type/" + MqttUtil.DEFAULT_DEVICE_TYPE
					+ "/id/" + deviceid + "/cmd/" + MqttUtil.DEFAULT_CMD_ID
					+ "/fmt/json", jsonObj.toString(), false, 0);

第 4 步. 运行应用程序

构建项目

更新应用程序端配置文件

#Configuration files for App Side Applications
           
#The org field is the same org field as the Device side
org=<org>
           
# A unique id you choose it by yourself, maybe, abcdefg123456
appid=<appid>
           
# The key field from App Keys info you copied previously
key=<Key>
           
# The Auth Token field from App Keys info you copied previously
token=<Auth_Token>
           
#T or F, T means using SSL, while F means not
isSSL=F

运行应用程序端程序

更新设备端配置文件

#Configuration files for Device Side Applications
           
#The org field from Device info you copied previously
org=<org>
           
#The id field from Device info you copied previously
deviceid=<id>
           
#The auth-token field from Device info you copied previously
token=<auth-token>
           
#T or F, T means using SSL, while F means not
isSSL=F

运行设备端程序

检查正在运行的应用程序

了解更多信息

通过 Node-RED 执行应用程序端编程（可选）

结束语