IDEA连接阿里云物联网平台获取设备上传数据

官方文档： https://help.aliyun.com/document_detail/143601.html?spm=a2c4g.11174283.6.646.28ba1668pisoQW

一、物联网平台准备

1、准备消费组
打开物联网平台 -> 规则引擎 -> 服务端订阅 -> 消费组列表
点击创建消费组,输入名称。

创建完成，为创建的消费组添加订阅。点击查看，打开订阅产品一栏，创建订阅。

2、准备AccessKey

创建完成

二、IDEA端

1、添加依赖

       <!-- amqp 1.0 qpid client --><dependency><groupId>org.apache.qpid</groupId><artifactId>qpid-jms-client</artifactId><version>0.56.0</version></dependency><!-- util for base64--><dependency><groupId>commons-codec</groupId><artifactId>commons-codec</artifactId><version>1.10</version></dependency><!-- ================================================= --><!-- 日志及相关依赖（用slf4j+logback） --><!-- ================================================= --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.5</version></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.3</version></dependency>

2、创建AmqpClient类
贴上官方代码：

import java.net.URI;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;import org.apache.commons.codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class AmqpClient {private final static Logger logger = LoggerFactory.getLogger(AmqpClient.class);private static String accessKey = "${YourAccessKey}";private static String accessSecret = "${YourAccessSecret}";private static String consumerGroupId = "${YourConsumerGroupId}";//iotInstanceId：企业版实例请填写实例ID，公共实例请填空字符串""。private static String iotInstanceId = "${YourIotInstanceId}";//控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。private static String clientId = "${YourClientId}";//${YourHost}为接入域名，请参见AMQP客户端接入说明文档。private static String host = "${YourHost}";// 指定单个进程启动的连接数// 单个连接消费速率有限，请参考使用限制，最大64个连接// 连接数和消费速率及rebalance相关，建议每500QPS增加一个连接private static int connectionCount = 4;//业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。private final static ExecutorService executorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,new LinkedBlockingQueue(50000));public static void main(String[] args) throws Exception {List<Connection> connections = new ArrayList<>();//参数说明，请参见AMQP客户端接入说明文档。for (int i = 0; i < connectionCount; i++) {long timeStamp = System.currentTimeMillis();//签名方法：支持hmacmd5、hmacsha1和hmacsha256。String signMethod = "hmacsha1";//userName组装方法，请参见AMQP客户端接入说明文档。String userName = clientId + "-" + i + "|authMode=aksign"+ ",signMethod=" + signMethod+ ",timestamp=" + timeStamp+ ",authId=" + accessKey+ ",iotInstanceId=" + iotInstanceId+ ",consumerGroupId=" + consumerGroupId+ "|";//计算签名，password组装方法，请参见AMQP客户端接入说明文档。String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;String password = doSign(signContent, accessSecret, signMethod);String connectionUrl = "failover:(amqps://" + host + ":5671?amqp.idleTimeout=80000)"+ "?failover.reconnectDelay=30";Hashtable<String, String> hashtable = new Hashtable<>();hashtable.put("connectionfactory.SBCF", connectionUrl);hashtable.put("queue.QUEUE", "default");hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");Context context = new InitialContext(hashtable);ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");Destination queue = (Destination)context.lookup("QUEUE");// 创建连接。Connection connection = cf.createConnection(userName, password);connections.add(connection);((JmsConnection)connection).addConnectionListener(myJmsConnectionListener);// 创建会话。// Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。// Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);connection.start();// 创建Receiver连接。MessageConsumer consumer = session.createConsumer(queue);consumer.setMessageListener(messageListener);}logger.info("amqp demo is started successfully, and will exit after 60s ");// 结束程序运行 Thread.sleep(60 * 1000);logger.info("run shutdown");connections.forEach(c-> {try {c.close();} catch (JMSException e) {logger.error("failed to close connection", e);}});executorService.shutdown();if (executorService.awaitTermination(10, TimeUnit.SECONDS)) {logger.info("shutdown success");} else {logger.info("failed to handle messages");}}private static MessageListener messageListener = new MessageListener() {@Overridepublic void onMessage(final Message message) {try {//1.收到消息之后一定要ACK。// 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。// 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。// message.acknowledge();//2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。// 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。executorService.submit(new Runnable() {@Overridepublic void run() {processMessage(message);}});} catch (Exception e) {logger.error("submit task occurs exception ", e);}}};/*** 在这里处理您收到消息后的具体业务逻辑。*/private static void processMessage(Message message) {try {byte[] body = message.getBody(byte[].class);String content = new String(body);String topic = message.getStringProperty("topic");String messageId = message.getStringProperty("messageId");logger.info("receive message"+ ",\n topic = " + topic+ ",\n messageId = " + messageId+ ",\n content = " + content);} catch (Exception e) {logger.error("processMessage occurs error ", e);}}private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {/*** 连接成功建立。*/@Overridepublic void onConnectionEstablished(URI remoteURI) {logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);}/*** 尝试过最大重试次数之后，最终连接失败。*/@Overridepublic void onConnectionFailure(Throwable error) {logger.error("onConnectionFailure, {}", error.getMessage());}/*** 连接中断。*/@Overridepublic void onConnectionInterrupted(URI remoteURI) {logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);}/*** 连接中断后又自动重连上。*/@Overridepublic void onConnectionRestored(URI remoteURI) {logger.info("onConnectionRestored, remoteUri:{}", remoteURI);}@Overridepublic void onInboundMessage(JmsInboundMessageDispatch envelope) {}@Overridepublic void onSessionClosed(Session session, Throwable cause) {}@Overridepublic void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}@Overridepublic void onProducerClosed(MessageProducer producer, Throwable cause) {}};/*** 计算签名，password组装方法，请参见AMQP客户端接入说明文档。*/private static String doSign(String toSignString, String secret, String signMethod) throws Exception {SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);Mac mac = Mac.getInstance(signMethod);mac.init(signingKey);byte[] rawHmac = mac.doFinal(toSignString.getBytes());return Base64.encodeBase64String(rawHmac);}
}

需要我们修改的为开头一系列参数。

先放官方文档介绍。

（1）AccessKey
回到刚刚的AccessKey管理页面，点击后面的查看secret，输入验证码。

（2）consumerGroupId
回到刚刚的消费组列表，点击后面的查看，获取消费组ID。

（3）iotInstanceId
如官方所说，企业版实例请填写实例ID，公共实例请填空字符串""。我用的为公共实例，所以用空字符串。
（4）clientId
自己起一个名字，起到标识作用。
（5）host
查看官方文档。

${uid}.iot-amqp.${YourRegionId}.aliyuncs.com

最后修改后我的如下，供参考。

三、测试

修改完毕后IDEA运行该程序。注意启动后有时间限制。

打开物联网平台 -> 监控运维 -> 设备模拟器，选择调试设备，点击属性上报，点击启动设备模拟器。

输入参数，点击发送指令。

回到IDEA，可以看到已经可以接收到消息。

完成。