代码收藏家 阿里云OSS与物联网（IOT）集成使用指南
文章目录
前言
本文详细讲解了阿里云OSS和IOT创建流程和Java实现思路，较为浅显说明了Apache Qpid框架的使用。
一、文件上传
在介绍OSS之前首先介绍下文件上传和本地储存相关内容
2. 前端三要素
提交表单时这表单项中所有数据都会被提交到后台
2. 后端接收处理注意事项
由于设置了Miltipart/form-data编码格式，表单项中的每一项都会产生一个临时文件。
3. 本地存储
MultipartFile类中的提供了transserTo方法来将问存到磁盘，注意文件名不能写死，设置为某个固定文件名，每次存储都会被覆盖掉，所以需要借用UUID的randomUUID()生成一个随机唯一的UUID，通过这个UUID进行拼接来确保我们文件名的唯一性
代码实现：
上传文件大小配置
#配置单个文件最大上传大小
spring:
servlet:
multipart:
max-file-size: 10MB
#配置单个请求最大上传大小(一次请求可以上传多个文件)
spring:
servlet:
multipart:
max-request-size: 100MB
#文件大小的配置都是spring.servlet.multipart.max,其中一个是file应该是request
二、阿里对象存储 OSS
实际开发中会选择存的公司服务器或者云存储上，以下以阿里云为例说明，云存储操作思路和方式是一致的
1. 介绍
2. 入门程序
将下面代码复制到集成开发工具中运行即可
package com.itheima;
import com.aliyun.oss.ClientException;
import com.aliyun.oss.OSS;
import com.aliyun.oss.common.auth.*;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.OSSException;
import com.aliyun.oss.model.PutObjectRequest;
import com.aliyun.oss.model.PutObjectResult;
import java.io.FileInputStream;
import java.io.InputStream;
public class Demo {
public static void main(String[] args) throws Exception {
// Endpoint以华东1（杭州）为例，其它Region请按实际情况填写。
String endpoint = "https://oss-cn-beijing.aliyuncs.com";
// 从环境变量中获取访问凭证。运行本代码示例之前，请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
EnvironmentVariableCredentialsProvider credentialsProvider = CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider();
// 填写Bucket名称，例如examplebucket。
String bucketName = "oss-testfile-001";
// 填写Object完整路径，完整路径中不能包含Bucket名称，例如exampledir/exampleobject.txt。
String objectName = "01.jpg";
// 填写本地文件的完整路径，例如D:\\localpath\\examplefile.txt。
// 如果未指定本地路径，则默认从示例程序所属项目对应本地路径中上传文件流。
String filePath= "E:\\24383\\Pictures\\Saved Pictures\\微信图片_20240803113433.jpg";
// 创建OSSClient实例。
OSS ossClient = new OSSClientBuilder().build(endpoint, credentialsProvider);
try {
InputStream inputStream = new FileInputStream(filePath);
// 创建PutObjectRequest对象。
PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, objectName, inputStream);
// 创建PutObject请求。
PutObjectResult result = ossClient.putObject(putObjectRequest);
} catch (OSSException oe) {
System.out.println("Caught an OSSException, which means your request made it to OSS, "
+ "but was rejected with an error response for some reason.");
System.out.println("Error Message:" + oe.getErrorMessage());
System.out.println("Error Code:" + oe.getErrorCode());
System.out.println("Request ID:" + oe.getRequestId());
System.out.println("Host ID:" + oe.getHostId());
} catch (ClientException ce) {
System.out.println("Caught an ClientException, which means the client encountered "
+ "a serious internal problem while trying to communicate with OSS, "
+ "such as not being able to access the network.");
System.out.println("Error Message:" + ce.getMessage());
} finally {
if (ossClient != null) {
ossClient.shutdown();
}
}
}
}
3. 集成至项目
- 实例：
下图请求流程，用户上传图片后被change事件监听，向后端发送请求，请求携带图片路径，后端将图片上传到阿里云，然后将其云路径响应给前端。
- 接口文档分析
- 接收文件—存储—返回路径 根据官方提供的方法改造(开发者会提供)
- 但是这样写配置信息是有问题的，例如当我们跟换云存储地址的时候，需要找到每个用到旧地址的类，这个时候就需要使用配置文件了
- 通过@Value注解读取配置文件内容，修改的配置文件内容就可以修改全局（这种方式用的比较多，属性过多是可以封装到类中并使用@ConfigurationProperties注解标注）
- 阿里云文件访问路径规则： https://BucketName.Endpoint/ObjectName
三、阿里云IOT
1. 介绍
1.1. 什么是物联网
把所有物品通过信息传感设备与互联网连接起来，进行信息交换，即物物相息，以实现智能化识别和管理。
物联网（英文：Internet of Things，缩写：IoT）起源于传媒领域，是信息科技产业的第三次革命。物联网是指通过信息传感设备，按约定的协议，将任何物体与网络相连接，物体通过信息传播媒介进行信息交换和通信，以实现智能化识别、定位、跟踪、监管等功能。
常见的应用场景：
1.2. IOT简介
产品文档：https://help.aliyun.com/zh/iot/product-overview/?spm=a2c4g.11186623.0.0.32d844a6NPRO9e
阿里云物联网平台是一个集成了设备管理、数据安全通信、消息订阅和数据服务等能力的一体化平台。向下支持连接海量设备，采集设备数据上云；向上提供云端API，服务端可通过调用云端API将指令下发至设备端，实现远程控制。
我们作为一个开发者，基本的设备与后台调度思路，如下：
更多的介绍可以阅读官方产品文档
1.2.1. 开通物联网平台
- 开通阿里云账号
前往阿里云官网注册账号。如果已有注册账号，请跳过此步骤。
进入阿里云首页后，如果没有阿里云的账户需要先进行注册，才可以进行登录。由于注册较为简单，课程和讲义不在进行体现（注册可以使用多种方式，如淘宝账号、支付宝账号等…）。
需要实名认证和活体认证。 - 开通物理网平台
登录账号以后，我们可以在产品中搜索物联网平台
打开之后，点击管理控制台
如果没有开通的话，会提示你开通物联网平台，如下图，直接开通即可
- 申请公共实例
在IOT中分为了两种实例，一个是公共实例，另外一个是企业实例，不同的实例收费标准和功能是不一样的
申请使用公共实例使用，如下图
注意：地域必须选择上海才能申请公共实例1.3. 产品
一旦拥有了公共实例，我们就可以使用临时实例来进行开发，我们先来介绍产品和设备
- 创建产品
https://help.aliyun.com/zh/iot/user-guide/create-a-product?spm=a2c4g.11186623.0.0.6ac7133dbo6zWV
产品：设备的集合，通常指一组具有相同功能的设备。物联网平台为每个产品颁发全局唯一的ProductKey。
简单说就是某一类产品，比如，手表、大门通道门禁、紧急呼叫报警器、滞留报警器、跌倒报警器
现在我们可以创建产品，找到产品–>创建产品
如下图，输入产品名称，然后选择平台提供好的分类，其他选择默认即可，然后确认创建
创建成功之后，如下图
在产品列表中也可以查看，刚刚创建的产品
- 物模型
https://help.aliyun.com/zh/iot/user-guide/add-a-tsl-feature?spm=a2c4g.11186623.0.0.9abf6ec2jkNsER
产品创建好之后，可以给产品添加物模型，也就是给产品定义功能。
比如我们刚才创建的手表产品，可以定义功能，功能也可以分为两类，一个是监测手表本身，一个是因为指标数据
像这些耗电量、血压、血氧数据都属于产品的功能，也叫做物模型
在IOT平台的物模型中，分为了三类：
创建物模型流程，如下图：
注意：为已发布产品添加物模型时要先撤销发布1.4. 设备
https://help.aliyun.com/zh/iot/user-guide/create-a-device?spm=a2c4g.11186623.0.0.4f1b12ackrzdOs
产品是设备的集合，通常指一组具有相同功能的设备。创建产品完成后，需在产品下添加设备，获取设备证书。您可在物联网平台上，同时创建一个或多个设备
前提条件：设备是绑定在产品上的，所以必须先创建产品才行
操作步骤：
- 在左侧导航栏，选择设备管理> 设备。
- 在设备页面，单击添加设备。
- 在添加设备对话框中，输入设备信息，单击确认。
如下图：
在添加设备的时候有三个参数，解释如下：
创建设备成功后，会自动弹出添加完成对话框。您可以查看、复制设备证书信息。设备证书由设备的ProductKey、DeviceName和DeviceSecret组成，是设备与物联网平台进行通信的重要身份认证。
1.5. 设备数据上报
https://help.aliyun.com/zh/iot/user-guide/devices-retrieve-certificates/?spm=a2c4g.11186623.0.0.671b53felMSmzL
物理设备可通过两种方式获取物联网平台颁发的设备证书（ProductKey、DeviceName和DeviceSecret）：设备厂商在产线上将证书烧录到设备上和设备上电联网后从厂商云服务中获取证书。
物联网烧录：是指将特定的程序或数据写入物联网设备中的过程。这些设备可能包括智能家居设备、智能穿戴设备、智能传感器等。通过烧录，可以实现设备的特定功能，例如控制灯光、监测温度、收集数据等。物联网烧录需要使用专业的工具和技术，确保烧录信息的完整性和准确性，以保证设备的正常工作和稳定性。
我们在开发阶段可以使用联网的电脑，来模拟设备的数据上报，比较简答的方式可以使用node来进行链接上报数据，参考代码如下：
const mqtt = require('aliyun-iot-mqtt');
// 1. 设备身份信息
var options = {
productKey: "j0rk1AN61hM",
deviceName: "watch001",
deviceSecret: "ea94110e5495bb04b0a7b35b9535a50c",
host: "iot-06z00frq8umvkx2.mqtt.iothub.aliyuncs.com"
};
// 2. 建立MQTT连接
const client = mqtt.getAliyunIotMqttClient(options);
//订阅云端指令Topic
client.subscribe(`/${options.productKey}/${options.deviceName}/user/get`)
client.subscribe(`/sys/${options.productKey}/${options.deviceName}/thing/event/property/post_reply`)
client.on('message', function (topic, message) {
console.log("topic " + topic)
console.log("message " + message)
})
setInterval(function () {
// 3.定时上报温湿度数据
client.publish(`/sys/${options.productKey}/${options.deviceName}/thing/event/property/post`, getPostData(), { qos: 0 });
}, 5 * 1000);
var power = 1000;
function getPostData () {
const payloadJson = {
id: Date.now(),
version: "1.0",
params: {
PowerConsumption: power--
},
method: "thing.event.property.post"
}
console.log("payloadJson " + JSON.stringify(payloadJson))
return JSON.stringify(payloadJson);
}
把上述代码保存到一个文件夹下，以js为后缀名，如：iot_device_01.js
然后在js所在的文件夹下，打开cmd窗口，分别执行
node -i
node iot_device_01.js
启动之后的效果如下：
设备启动后，可以在物联网平台查看刚才创建的设备，现在已在线
找到物模型数据，可以看到上报之后的数据
2. IOT接口对接（Java环境操作设备信息—增删改查）
2.1. 环境集成
IOT平台目前已经给提供了完整的SDK，我们可以快速集成到项目进行接口的调用，集成方式链接如下：
https://help.aliyun.com/zh/iot/developer-reference/use-iot-platform-sdk-for-java-1?spm=a2c4g.11186623.0.0.4de85c09n2V2vb
导入依赖
<!-- https://mvnrepository.com/artifact/com.aliyun/iot20180120 -->
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>iot20180120</artifactId>
<version>3.0.8</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>tea-openapi</artifactId>
<version>0.2.2</version>
</dependency>
创建Client的Bean
package com.zzyl.config;
import com.aliyun.iot20180120.Client;
import com.aliyun.teaopenapi.models.Config;
import com.zzyl.properties.AliIoTConfigProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class IotClientConfig {
@Autowired
//该类封装了阿里账号密钥等信息的相关配置
private AliIoTConfigProperties aliIoTConfigProperties;
@Bean
//参考官方文档
public Client instance() throws Exception {
Config config = new Config();
config.accessKeyId = aliIoTConfigProperties.getAccessKeyId();
config.accessKeySecret = aliIoTConfigProperties.getAccessKeySecret();
// 您的可用区ID 默认上海
config.regionId = aliIoTConfigProperties.getRegionId();
return new Client(config);
}
}
2.2. Client解析
在Java中，通过IAcsClient接口可以访问阿里云提供的各种服务API。IAcsClient是一个抽象类，它定义了与阿里云服务交互的基本方法。这里的Client为IAcsClient的一个实现类
其定义参考阿里的官方文档
2.3. Client调用API分析
集成SDK后，我们可以直接通过Client对象来调用阿里提供的各种API
- 根据官方文档找到对应API
- 通过Client示例调用(注意：实际方法为文档首字母小写)
- 调用时需要什么参数就参考文档封装什么
提示：阿里提供的API较为完善，其每个方法都有一个对应的参数类，来接收请求参数，其参数类为方法名+request，调用方法时只需要将请求参数封装到请求类中，传入即可
示例
/**
* 根据产品key批量修改指定的设备昵称
* @param deviceDto
*/
@Override
public void batchUpdateDeviceNickname(DeviceDto deviceDto) throws Exception {
//修改阿里IOT昵称
3.
//封装需要修改的设备信息
BatchUpdateDeviceNicknameRequest.BatchUpdateDeviceNicknameRequestDeviceNicknameInfo requestDeviceNicknameInfo
= new BatchUpdateDeviceNicknameRequest.BatchUpdateDeviceNicknameRequestDeviceNicknameInfo();
requestDeviceNicknameInfo.setIotId(deviceDto.getIotId());
requestDeviceNicknameInfo.setNickname(deviceDto.getNickname());
//将设备信息装入集合
List<BatchUpdateDeviceNicknameRequest.BatchUpdateDeviceNicknameRequestDeviceNicknameInfo> list = new ArrayList<>();
list.add(requestDeviceNicknameInfo);
2.
//发送请求修改设备昵称
BatchUpdateDeviceNicknameRequest request = new BatchUpdateDeviceNicknameRequest();
request.setIotInstanceId(aliIoTConfigProperties.getIotInstanceId());
request.setDeviceNicknameInfo(list);
1.
BatchUpdateDeviceNicknameResponse batchUpdateDeviceNicknameResponse = client.batchUpdateDeviceNickname(request);
5.
if (batchUpdateDeviceNicknameResponse.getBody().getSuccess()) {
//修改数据库信息
Device device = BeanUtil.toBean(deviceDto, Device.class);
deviceMapper.updateByPrimaryKeySelective(device);
}
}
- 找到官方文档的目标方法，并调用
2. 调用时发现该方法需要接收一个BatchUpdateDeviceNicknameRequest实例，该实例就是阿里封装好的对应该方法的请求参数封装类，我们创建通过setXxx封装即可
3. 创建BatchUpdateDeviceNicknameRequest时，发现该实例set需要一个BatchUpdateDeviceNicknameRequestDeviceNicknameInfo的集合
4. 在结合官方文档进行分析可知
BatchUpdateDeviceNicknameRequest该示例需要封装IotInstanceId，和DeviceNicknameInfo的集合，每个DeviceNicknameInfo中需要分组的请求信息如下红框
5. 通过请求方法返回结果—xxxResponse调用getBody()获取响应信息,getDate()获取响应体即可解析结果（JSON都不需要自己动手转换）
提示：getBody().getXxx() Xxx为返回结果名称，即可获取对应字段内容
6. 总结：
参考官方文档和代码提示，需要什么封装什么
3. IOT消息采集和处理
我们再来回顾一下IOT基本的工作原理
从图中，我们可以看到，当设备上报数据到IOT平台之后，业务系统可以从IOT平台中拉取数据到自己的数据库，方便在业务应用中去展示。
3.1. 异步处理基础概念
步和异步的概念，在我们之前讲解js课程的时候已经学习过，基本的思路是一样的
3.2.消息队列的基本概念
我们知道了同步和异步的区别之后，我们来理解一下消息队列的概念，消息队列都是以异步的方式进行请求
在IOT中数据流转是这样的，如下图
3.3. AMQP—Apache Qpid
在IOT中数据接收用到的是AMQP来处理消息。
我们这次并不会详细讲解这些软件的使用，其中关于RabbitMQ后期的课程中会详细的，重点的去讲解。
今天我们快速使用Apache Qpid软件来接收IOT中的数据，如下图
- 设备数据消费
在IOT官方文档中，已经提供了对应的接收数据的解决方案，如下链接：SDK对接 - 官网Java SDK接入
参考官网提供的pom依赖，如下：
<!-- amqp 1.0 qpid client -->
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-jms-client</artifactId>
<version>0.57.0</version>
</dependency>
<!-- util for base64-->
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.10</version>
</dependency>
下载地址：https://linkkit-export.oss-cn-shanghai.aliyuncs.com/amqp/amqp-demo.zip
我们可以修改里面的参数，包含以下几个重要参数：
修改之后的代码：
package com.aliyun.iotx.demo;
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
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);
/**
* 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例使用环境变量获取 AccessKey 的方式进行调用，仅供参考
*/
private static String accessKey = "LTAI5tDQKg9F61aJhbmhqVRK";
private static String accessSecret = "LYUKZH7HQGBoD025pmSq0fQsREaOYD";;
private static String consumerGroupId = "eraicKJm98cQR0hHgsxb000100";
//iotInstanceId：实例ID。若是2021年07月30日之前（不含当日）开通的公共实例，请填空字符串。
private static String iotInstanceId = "iot-06z00frq8umvkx2";
//控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。
private static String clientId;
static {
try {
clientId = InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
//${YourHost}为接入域名，请参见AMQP客户端接入说明文档。
private static String host = "iot-06z00frq8umvkx2.amqp.iothub.aliyuncs.com";
// 指定单个进程启动的连接数
// 单个连接消费速率有限，请参考使用限制，最大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 + "×tamp=" + 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(600 * 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() {
@Override
public 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() {
@Override
public 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() {
/**
* 连接成功建立。
*/
@Override
public void onConnectionEstablished(URI remoteURI) {
logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
}
/**
* 尝试过最大重试次数之后，最终连接失败。
*/
@Override
public void onConnectionFailure(Throwable error) {
logger.error("onConnectionFailure, {}", error.getMessage());
}
/**
* 连接中断。
*/
@Override
public void onConnectionInterrupted(URI remoteURI) {
logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
}
/**
* 连接中断后又自动重连上。
*/
@Override
public void onConnectionRestored(URI remoteURI) {
logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
}
@Override
public void onInboundMessage(JmsInboundMessageDispatch envelope) {}
@Override
public void onSessionClosed(Session session, Throwable cause) {}
@Override
public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}
@Override
public 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);
}
}
接收消息订阅
第一：找到 消息转发>服务端订阅->消费者组列表
第二：创建一个自己的消费者组
创建好之后可以查看到已经创建好的消费者组，并且自动生成了消费者组id
入刚刚创建的消费者组，然后点击订阅产品，然后创建订阅
第三：需要选择消费者组与推送消息类型（设备上报数据），如下图
修改demo代码中的消费者组，改为自己创建的消费者组ID
测试
demo代码中绑定对应的消费者组
另外再demo代码中休眠时间改长一些
启动后台代码，可以在日志中查看消费者到的数据
3.4. SDK改造
- 线程池配置
package com.zzyl.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@Configuration
public class ThreadPoolConfig {
/**
* 核心线程池大小
*/
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
/**
* 最大可创建的线程数
*/
private static final int MAX_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
/**
* 队列最大长度
*/
private static final int QUEUE_CAPACITY = 50000;
/**
* 线程池维护线程所允许的空闲时间
*/
private static final int KEEP_ALIVE_SECONDS = 60;
@Bean
public ExecutorService executorService(){
AtomicInteger c = new AtomicInteger(1);
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(QUEUE_CAPACITY);
return new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
KEEP_ALIVE_SECONDS,
TimeUnit.MILLISECONDS,
queue,
r -> new Thread(r, "zzyl-pool-" + c.getAndIncrement()),
new ThreadPoolExecutor.DiscardPolicy()
);
}
}
ⅰ. ArrayBlockingQueue;
ⅱ. LinkedBlockingQueue;
ⅲ. SynchronousQueue;
ⅰ. ThreadPoolExecutor.AbortPolicy：丢弃任务并抛出RejectedExecutionException异常。
ⅱ. ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
ⅲ. ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常
ⅳ. ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
- 实现ApplicationRunner接口之后的AmqpClient
package com.zzyl.job;
import com.zzyl.properties.AliIoTConfigProperties;
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;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.concurrent.ExecutorService;
@Component
public class AmqpClient implements ApplicationRunner {
private final static Logger logger = LoggerFactory.getLogger(AmqpClient.class);
@Autowired
private AliIoTConfigProperties aliIoTConfigProperties;
//控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。
private static String clientId;
static {
try {
clientId = InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
// 指定单个进程启动的连接数
// 单个连接消费速率有限，请参考使用限制，最大64个连接
// 连接数和消费速率及rebalance相关，建议每500QPS增加一个连接
private static int connectionCount = 64;
//业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。
@Autowired
private ExecutorService executorService;
public void start() 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=" + aliIoTConfigProperties.getAccessKeyId()
+ ",iotInstanceId=" + aliIoTConfigProperties.getIotInstanceId()
+ ",consumerGroupId=" + aliIoTConfigProperties.getConsumerGroupId()
+ "|";
//计算签名，password组装方法，请参见AMQP客户端接入说明文档。
String signContent = "authId=" + aliIoTConfigProperties.getAccessKeyId() + "×tamp=" + timeStamp;
String password = doSign(signContent, aliIoTConfigProperties.getAccessKeySecret(), signMethod);
String connectionUrl = "failover:(amqps://" + aliIoTConfigProperties.getHost() + ":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 is started successfully, and will exit after server shutdown ");
}
private MessageListener messageListener = message -> {
try {
//异步处理收到的消息，确保onMessage函数里没有耗时逻辑
executorService.submit(() -> processMessage(message));
} catch (Exception e) {
logger.error("submit task occurs exception ", e);
}
};
/**
* 在这里处理您收到消息后的具体业务逻辑。
*/
private void processMessage(Message message) {
try {
byte[] body = message.getBody(byte[].class);
String contentStr = 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 = " + contentStr);
} catch (Exception e) {
logger.error("processMessage occurs error ", e);
}
}
private JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
/**
* 连接成功建立。
*/
@Override
public void onConnectionEstablished(URI remoteURI) {
logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
}
/**
* 尝试过最大重试次数之后，最终连接失败。
*/
@Override
public void onConnectionFailure(Throwable error) {
logger.error("onConnectionFailure, {}", error.getMessage());
}
/**
* 连接中断。
*/
@Override
public void onConnectionInterrupted(URI remoteURI) {
logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
}
/**
* 连接中断后又自动重连上。
*/
@Override
public void onConnectionRestored(URI remoteURI) {
logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
}
@Override
public void onInboundMessage(JmsInboundMessageDispatch envelope) {
}
@Override
public void onSessionClosed(Session session, Throwable cause) {
}
@Override
public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {
}
@Override
public 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);
}
@Override
public void run(ApplicationArguments args) throws Exception {
start();
}
}
- 接收设备端数据
整体的接收数据思路如下：
修改AmqpClient类中的processMessage，最终代码如下：
package com.zzyl.job;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.json.JSONUtil;
import com.zzyl.entity.DeviceData;
import com.zzyl.mapper.DeviceDataMapper;
import com.zzyl.mapper.DeviceMapper;
import com.zzyl.properties.AliIoTConfigProperties;
import com.zzyl.utils.ObjectUtil;
import com.zzyl.vo.DeviceVo;
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;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.concurrent.ExecutorService;
@Component
public class AmqpClient implements ApplicationRunner {
private final static Logger logger = LoggerFactory.getLogger(AmqpClient.class);
@Autowired
private AliIoTConfigProperties aliIoTConfigProperties;
//控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。
private static String clientId;
static {
try {
clientId = InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
e.printStackTrace();
}
}
// 指定单个进程启动的连接数
// 单个连接消费速率有限，请参考使用限制，最大64个连接
// 连接数和消费速率及rebalance相关，建议每500QPS增加一个连接
private static int connectionCount = 64;
//业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。
@Autowired
private ExecutorService executorService;
public void start() 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=" + aliIoTConfigProperties.getAccessKeyId()
+ ",iotInstanceId=" + aliIoTConfigProperties.getIotInstanceId()
+ ",consumerGroupId=" + aliIoTConfigProperties.getConsumerGroupId()
+ "|";
//计算签名，password组装方法，请参见AMQP客户端接入说明文档。
String signContent = "authId=" + aliIoTConfigProperties.getAccessKeyId() + "×tamp=" + timeStamp;
String password = doSign(signContent, aliIoTConfigProperties.getAccessKeySecret(), signMethod);
String connectionUrl = "failover:(amqps://" + aliIoTConfigProperties.getHost() + ":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 is started successfully, and will exit after server shutdown ");
}
private MessageListener messageListener = message -> {
try {
//异步处理收到的消息，确保onMessage函数里没有耗时逻辑
executorService.submit(() -> processMessage(message));
} catch (Exception e) {
logger.error("submit task occurs exception ", e);
}
};
@Autowired
private DeviceMapper deviceMapper;
@Autowired
private DeviceDataMapper deviceDataMapper;
/**
* 在这里处理您收到消息后的具体业务逻辑。
*/
private void processMessage(Message message) {
try {
byte[] body = message.getBody(byte[].class);
String contentStr = 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 = " + contentStr);
//转换接收到的数据
Content content = JSONUtil.toBean(contentStr, Content.class);
//查询设备数据
DeviceVo deviceVo = deviceMapper.selectByIotId(content.getIotId());
if(ObjectUtil.isEmpty(deviceVo)){
logger.error("设备数据为空，iotId:{}",content.getIotId());
return;
}
List<DeviceData> deviceDataList = new ArrayList<>();
//保存设备数据
content.getItems().forEach((key,item)->{
DeviceData deviceData = DeviceData.builder()
.alarmTime(LocalDateTimeUtil.of(item.getTime()))
.deviceName(content.getDeviceName())
.productKey(content.getProductKey())
.iotId(content.getIotId())
.nickname(deviceVo.getNickname())
.functionId(key)
.accessLocation(deviceVo.getRemark())
.locationType(deviceVo.getLocationType())
.physicalLocationType(deviceVo.getPhysicalLocationType())
.deviceDescription(deviceVo.getDeviceDescription())
.dataValue(item.getValue() + "")
.productName(deviceVo.getProductName())
.build();
deviceDataList.add(deviceData);
});
//批量保存
deviceDataMapper.batchInsert(deviceDataList);
} catch (Exception e) {
logger.error("processMessage occurs error ", e);
}
}
private JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
/**
* 连接成功建立。
*/
@Override
public void onConnectionEstablished(URI remoteURI) {
logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
}
/**
* 尝试过最大重试次数之后，最终连接失败。
*/
@Override
public void onConnectionFailure(Throwable error) {
logger.error("onConnectionFailure, {}", error.getMessage());
}
/**
* 连接中断。
*/
@Override
public void onConnectionInterrupted(URI remoteURI) {
logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
}
/**
* 连接中断后又自动重连上。
*/
@Override
public void onConnectionRestored(URI remoteURI) {
logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
}
@Override
public void onInboundMessage(JmsInboundMessageDispatch envelope) {
}
@Override
public void onSessionClosed(Session session, Throwable cause) {
}
@Override
public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {
}
@Override
public 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);
}
@Override
public void run(ApplicationArguments args) throws Exception {
start();
}
}
作者：风清已存在
