使用java整合iotdb2.0.4+springboot3+mybatis-plus+dynamic-datasource
·
1. IOTDB说明
Apache IoTDB 是一款工业物联网时序数据库管理系统,采用端边云协同的轻量化架构,支持一体化的物联网时序数据收集、存储、管理与分析 ,具有多协议兼容、超高压缩比、高通量读写、工业级稳定、极简运维等特点。
官网链接: https://iotdb.apache.org/zh/
下载链接: https://iotdb.apache.org/zh/Download/
2. 整合IOTDB
2.1 相关配置
2.1.1 pom文件
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.rk</groupId>
<artifactId>iot</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.4.8</version>
<relativePath/>
</parent>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!--SpringBoot Web依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- pom.xml 示例 -->
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-spring-boot-starter</artifactId>
<version>2.0.3</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.28</version>
</dependency>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-spring-boot3-starter</artifactId>
<version>3.5.12</version>
</dependency>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>dynamic-datasource-spring-boot3-starter</artifactId>
<version>4.3.1</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid-spring-boot-starter</artifactId>
<version>1.2.27</version>
</dependency>
<!-- 基础JDBC驱动 -->
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-jdbc</artifactId>
<version>2.0.4</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.github.xiaoymin</groupId>
<artifactId>knife4j-openapi3-jakarta-spring-boot-starter</artifactId>
<version>4.4.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.0.0-M6</version>
<configuration>
<argLine>--add-opens=java.base/java.nio=ALL-UNNAMED</argLine>
</configuration>
</plugin>
</plugins>
</build>
</project>
2.1.2 配置文件application.yml
################### 项目配置 ###################
server:
port: 8888
servlet:
context-path: /iot
tomcat:
uri-encoding: UTF-8
max-http-form-post-size: 104857600
spring:
datasource:
type: com.alibaba.druid.pool.DruidDataSource
druid:
initial-size: 10
max-active: 500
max-pool-prepared-statement-per-connection-size: 20
max-wait: 60000
min-evictable-idle-time-millis: 300000
min-idle: 10
pool-prepared-statements: true
#配置监控页功能
stat-view-servlet:
enabled: true
login-password: admin
login-username: admin
reset-enable: true
url-pattern: /druid/*
test-on-borrow: false
test-on-return: false
test-while-idle: true
time-between-eviction-runs-millis: 60000
validation-query: SELECT 'x'
#第一个是web监控 第二个是防火墙 开启几个功能组件
filters: stat,wall
filter:
#对上面filters里面的stat的详细配置
stat:
# 慢查询时间 毫秒 超过当前时间均为慢查询
slow-sql-millis: 1000
# 是否记录慢查询
log-slow-sql: true
enabled: true
wall:
enabled: true
config:
#例子 所有更新sql都会被拦截
update-allow: true
# 不允许删除
delete-allow: true
# 不允许删除表
drop-table-allow: false
# 监控web应用
web-stat-filter:
enabled: true
exclusions: '"*.js , *.gif ,*.jpg ,*.png ,*.css ,*.ico , /druid/*"'
profile-enable: true
session-stat-enable: false
session-stat-max-count: 1000
url-pattern: /*
dynamic:
primary: master
# strict: false
datasource:
master:
url: jdbc:mysql://127.0.0.1:3306/test?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
username: root
password: root
driver-class-name: com.mysql.cj.jdbc.Driver
IOTDB:
url: jdbc:iotdb://192.168.3.12:6667/test?sql_dialect=table
username: root
password: root
driver-class-name: org.apache.iotdb.jdbc.IoTDBDriver
jackson:
time-zone: GMT+8
data:
redis:
host: 127.0.0.1
port: 6379
password: 123456
database: 10
lettuce:
pool:
max-active: 20
max-idle: 10
max-wait: -1ms
# properties:
# spring.json.trusted.packages: "com.rk.iot" # 反序列化信任包
mybatis-plus:
configuration:
#log-impl: org.apache.ibatis.logging.stdout.StdOutImpl #开启sql打印
call-setters-on-nulls: true
mapperLocations:
- classpath*:/mapper/*.xml
# springdoc-openapi项目配置
springdoc:
swagger-ui:
path: /swagger-ui.html
tags-sorter: alpha
operations-sorter: alpha
api-docs:
path: /v3/api-docs
group-configs:
- group: 'default'
paths-to-match: '/**'
packages-to-scan: org.rk.iot
# knife4j的增强配置,不需要增强可以不配
knife4j:
enable: true
setting:
language: zh_cn
# iotdbSession配置
iotdb:
session:
node_urls: 192.168.3.12:6667
username: root
password: root
sql-dialect: table
database: test
max-size: 10
2.1.3 实体类IotEntity
@Data
public class IotEntity {
@TableField("time")
private Date time;
@TableField("tag_id")
private String tagId;
@TableField("la")
private Float la;
}
2.2 JDBC方式
2.2.1 mapper
package org.rk.iot.mapper;
import com.baomidou.dynamic.datasource.annotation.DS;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.rk.iot.entity.IotEntity;
import java.util.Date;
import java.util.List;
@Mapper
@DS("IOTDB")
public interface IotDbMapper {
void insertBatch(@Param("entities") List<IotEntity> entityList);
List<IotEntity> selectData(@Param("tagId") String tagId, @Param("begin") Date begin,@Param("end") Date end);
}
XML文件
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="org.rk.iot.mapper.IotDbMapper">
<insert id="insertBatch">
insert into record
(time
,tag_id
,la)
VALUES
<foreach collection="entities" separator="," item="entity" index="index">
(#{entity.time}
, #{entity.tagId}
, #{entity.la})
</foreach>
</insert>
<select id="selectData" resultType="org.rk.iot.entity.IotEntity">
select time,tag_id,value,energy from record
<where>
and tag_id = #{tagId}
<if test="begin!=null">
and time >= #{begin}
</if>
<if test="end!=null">
and time<#{end}
</if>
</where>
</select>
</mapper>
2.3 iotSession方式
异步批量写入数据
private static final int BATCH_SIZE = 5000;
private static ExecutorService executor = Executors.newFixedThreadPool(20);
@Autowired
private IotDbMapper iotDbMapper;
@Autowired
private ITableSessionPool tableSessionPool;
private void writeDataIoIotDB(List<IotEntity> iotEntities,Map<Object,Object> map) {
List<String> columnNameList = Arrays.asList("tag_id", "la");
List<TSDataType> dataTypeList =
Arrays.asList(
TSDataType.STRING,
TSDataType.FLOAT);
List<ColumnCategory> columnTypeList =
new ArrayList<>(
Arrays.asList(
ColumnCategory.TAG,
ColumnCategory.FIELD
));
List<CompletableFuture<Void>> futureList = Lists.partition(iotEntities, BATCH_SIZE).stream().map(batch -> CompletableFuture.runAsync(() -> {
Tablet tablet = new Tablet("record", columnNameList, dataTypeList, columnTypeList, BATCH_SIZE);
for (IotEntity iotEntity : batch) {
Object la = map.get(iotEntity.getLa());
if (la == null){
continue;
}
int rowIndex = tablet.getRowSize();
tablet.addTimestamp(rowIndex, iotEntity.getTime().getTime());
tablet.addValue("tag_id", rowIndex, iotEntity.getTagId());
tablet.addValue("la", rowIndex, la);
}
try (ITableSession session = tableSessionPool.getSession()){
session.insert(tablet);
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
log.error(e.getMessage());
tableSessionPool.close();
}
}, executor)).collect(Collectors.toList());
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
}
魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。
更多推荐


所有评论(0)