TDengine 入门教程⑪——基于Spring Boot+Alibaba Druid框架的智能电表项目的后端时序数据库开发实战

男娘i 2023-10-07 20:19 20阅读 0赞

#### 文章目录 ####

*  一、前文
 *  二、工程依赖
 *  三、配置数据源
 *  四、数据库连接池
 *  五、电表数据实体类
 *  六、数据库建表
 *  七、业务Service层
 *  八、总结

## 一、前文 ##

> [TDengine 入门教程——导读][TDengine]

>  *  本开发实战配置多数据源，主数据库是MySQL，从数据库是TDengine。
>  *  原本想要双数据源都基于Spring Boot+Alibaba Druid+Apache MyBatis，但是遇到了点问题。
>  *  所以就改成主数据库MySQL是Spring Boot+Alibaba Druid+Apache MyBatis。
>  *  从数据库直接Spring Boot+Alibaba Druid。

> 本博文主要讲述基于Spring Boot+Alibaba Druid框架的，智能电表项目的，时序数据库后端开发实战内容

## 二、工程依赖 ##

*  注意Maven依赖的版本，必须跟服务器部署的TDengine版本一致。  
    `pom.xml`

<dependency>
                <groupId>com.taosdata.jdbc</groupId>
                <artifactId>taos-jdbcdriver</artifactId>
                <version>3.0.0</version>
            </dependency>

## 三、配置数据源 ##

*  MySQL数据库驱动：`com.mysql.cj.jdbc.Driver`
 *  TDengine数据库驱动：`com.taosdata.jdbc.rs.RestfulDriver`
 *  MySQL配置检测连接是否有效：`SELECT 1 FROM DUAL`
 *  TDengine配置检测连接是否有效：`select server_status();`

`application-druid.yml`

# 数据源配置
    spring:
        datasource:
            type: com.alibaba.druid.pool.DruidDataSource
            druid:
                # 主库数据源
                master:
                    url: jdbc:mysql://localhost:3306/keyueli?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
                    username: root
                    password: password
                    driverClassName: com.mysql.cj.jdbc.Driver
                    # 配置检测连接是否有效
                    validationQuery: SELECT 1 FROM DUAL
                # 从库数据源
                slave:
                    url: jdbc:TAOS-RS://localhost:6041/keyueli?timezone=UTC-8&charset=UTF-8&locale=en_US.UTF-8
                    username: root
                    password: password
                    driverClassName: com.taosdata.jdbc.rs.RestfulDriver
                    # 配置检测连接是否有效
                    validationQuery: select server_status();
                # 初始连接数
                initialSize: 5
                # 最小连接池数量
                minIdle: 10
                # 最大连接池数量
                maxActive: 20
                # 配置获取连接等待超时的时间
                maxWait: 60000
                # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
                timeBetweenEvictionRunsMillis: 60000
                # 配置一个连接在池中最小生存的时间，单位是毫秒
                minEvictableIdleTimeMillis: 300000
                # 配置一个连接在池中最大生存的时间，单位是毫秒
                maxEvictableIdleTimeMillis: 900000
                testWhileIdle: true
                testOnBorrow: false
                testOnReturn: false
                webStatFilter:
                    enabled: true
                statViewServlet:
                    enabled: true
                    # 设置白名单，不填则允许所有访问
                    allow:
                    url-pattern: /druid/*
                    # 控制台管理用户名和密码
                    login-username:
                    login-password:
                filter:
                    stat:
                        enabled: true
                        # 慢SQL记录
                        log-slow-sql: true
                        slow-sql-millis: 1000
                        merge-sql: true
                    #                    merge-sql: false
                    wall:
                        config:
                            multi-statement-allow: true

## 四、数据库连接池 ##

*  `alibaba.druid.pool`自定义数据库连接池；
 *  `@Value`从`application-druid.yml`获得从数据库的配置信息；
 *  `@Bean`从启动初始化的时候就得到一个`DataSource`数据库连接池；

`DruidPoolBuilder.java`

import com.alibaba.druid.pool.DruidDataSource;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.sql.DataSource;
    import java.sql.*;
    
    /**
     * druid 自定义数据库连接池
     *
     * @author keyueli
     */
    @Configuration
    public class DruidPoolBuilder {
        
        private static final Logger log = LoggerFactory.getLogger(DruidPoolBuilder.class);
    
        @Value("${spring.datasource.druid.initialSize}")
        private int initialSize;
    
        @Value("${spring.datasource.druid.minIdle}")
        private int minIdle;
    
        @Value("${spring.datasource.druid.maxActive}")
        private int maxActive;
    
        @Value("${spring.datasource.druid.maxWait}")
        private int maxWait;
    
        @Value("${spring.datasource.druid.slave.url}")
        private String url;
    
        @Value("${spring.datasource.druid.slave.username}")
        private String username;
    
        @Value("${spring.datasource.druid.slave.password}")
        private String password;
    
        @Value("${spring.datasource.druid.slave.driverClassName}")
        private String driverClassName;
    
        @Value("${spring.datasource.druid.slave.validationQuery}")
        private String validationQuery;
    
        DruidDataSource dataSource;
    
        @Bean
        public DataSource getDataSource() {
        
            log.info("url = "+url);
            dataSource = new DruidDataSource();
    
            dataSource.setUrl(url);
            dataSource.setUsername(username);
            dataSource.setPassword(password);
            dataSource.setDriverClassName(driverClassName);
            dataSource.setValidationQuery(validationQuery);
    
            dataSource.setInitialSize(initialSize);
            dataSource.setMinIdle(minIdle);
            dataSource.setMaxActive(maxActive);
            dataSource.setMaxWait(maxWait);
            return dataSource;
        }
    
        public int executeUpdate(String sql) {
        
            log.info("SQL >>> " + sql);
            try (Connection conn = dataSource.getConnection()) {
        
                try (Statement stmt = conn.createStatement()) {
        
                    return stmt.executeUpdate(sql);
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            } catch (SQLException e) {
        
                e.printStackTrace();
            }
            return 0;
        }
    }

## 五、电表数据实体类 ##

一条电表数据包含：

*  子表名：`houseId`。因为电表数据是跟着房间走。房间的租户/电表/等等信息都会发生变化，唯独房间id不会发生变化。
 *  动态数据（数据点）：
    
     *  `ts`：当前时间
     *  `addr`：电表地址
     *  `lastValue`：上次读数
     *  `currentValue`：当前读数
     *  `useValue`：用电量
     *  `unitPrice`：电费单价
     *  `usePrice`：用电费
     *  `balances`：余额
     *  `recordType`：类型, 0:正常上报; 1:第一次上报; 2:充值; 3:修改电费单价; 4:电表电费归零
 *  静态数据（标签）：
    
     *  `userId`：用户id
     *  `groupId`：分组id

import java.sql.Timestamp;
    public class WattHourRecord
    {
        
        // TDengine子表表名 房间id
        private Long houseId;
    
        //当前时间
        private Timestamp ts;
    
        //电表地址
        private String addr;
    
        //上次读数
        private float lastValue;
    
        //当前读数
        private float currentValue;
    
        //用电量
        private float useValue;
    
        //电费单价
        private float unitPrice;
    
        //用电费
        private float usePrice;
    
        //余额
        private float balances;
    
        //类型, 0:正常上报; 1:第一次上报; 2:充值; 3:修改电费单价; 4:电表电费归零
        private int recordType;
    
        private int userId;
    
        private int groupId;
    
        public void updateUseValue(){
        
            this.useValue = this.currentValue - this.lastValue;
        }
    
        public void updateUsePrice(){
        
            this.usePrice = this.unitPrice * this.useValue;
        }
    
        public Long getHouseId() {
        
            return houseId;
        }
    
        public void setHouseId(Long houseId) {
        
            this.houseId = houseId;
        }
    
        public Timestamp getTs() {
        
            return ts;
        }
    
        public void setTs(Timestamp ts) {
        
            this.ts = ts;
        }
    
        public String getAddr() {
        
            return addr;
        }
    
        public void setAddr(String addr) {
        
            this.addr = addr;
        }
    
        public float getLastValue() {
        
            return lastValue;
        }
    
        public void setLastValue(float lastValue) {
        
            this.lastValue = lastValue;
        }
    
        public float getCurrentValue() {
        
            return currentValue;
        }
    
        public void setCurrentValue(float currentValue) {
        
            this.currentValue = currentValue;
        }
    
        public float getUseValue() {
        
            return useValue;
        }
    
        public void setUseValue(float useValue) {
        
            this.useValue = useValue;
        }
    
        public float getUnitPrice() {
        
            return unitPrice;
        }
    
        public void setUnitPrice(float unitPrice) {
        
            this.unitPrice = unitPrice;
        }
    
        public float getUsePrice() {
        
            return usePrice;
        }
    
        public void setUsePrice(float usePrice) {
        
            this.usePrice = usePrice;
        }
    
        public float getBalances() {
        
            return balances;
        }
    
        public void setBalances(float balances) {
        
            this.balances = balances;
        }
    
        public int getRecordType() {
        
            return recordType;
        }
    
        public void setRecordType(int recordType) {
        
            this.recordType = recordType;
        }
    
        public int getUserId() {
        
            return userId;
        }
    
        public void setUserId(int userId) {
        
            this.userId = userId;
        }
    
        public int getGroupId() {
        
            return groupId;
        }
    
        public void setGroupId(int groupId) {
        
            this.groupId = groupId;
        }
    
        @Override
        public String toString() {
        
            return "WattHourRecord{" +
                    "houseId=" + houseId +
                    ", ts=" + ts +
                    ", addr='" + addr + '\'' +
                    ", lastValue=" + lastValue +
                    ", currentValue=" + currentValue +
                    ", useValue=" + useValue +
                    ", unitPrice=" + unitPrice +
                    ", usePrice=" + usePrice +
                    ", balances=" + balances +
                    ", recordType=" + recordType +
                    ", userId=" + userId +
                    ", groupId=" + groupId +
                    '}';
        }
    }

## 六、数据库建表 ##

*  系统运行之前需要先建表（超级表）

CREATE TABLE keyueli.watt_hour_record(ts timestamp, addr nchar(12), last_value float, current_value float, use_value float, unit_price float, use_price float, balances float, record_type int) tags(user_id int, group_id int);

## 七、业务Service层 ##

主要实现几个函数

*  `sum(Long houseId, Long startTime, Long endTime)`：获取某个房间电表的某个时间段的电量和、电费和。
 *  `lastOne(Long houseId)`：获取某个房间电表的最新数据。
 *  `select(Long houseId, Long startTime, Long endTime, int limit, int offset)`：分页获取某个房间电表的某个时间段的电表数据列表。
 *  `insert(WattHourRecord wattHourRecord);`：插入一条电表数据到TDengine。
 *  `alter(WattHourRecord wattHourRecord);`：修改电表数据的静态数据（标签）。

`IWattHourRecordService.java`

import com.keyueli.common.core.domain.entity.WattHourRecord;
    import java.util.List;
    
    public interface IWattHourRecordService {
        
    
        WattHourRecord sum(Long houseId, Long startTime, Long endTime);
    
        WattHourRecord lastOne(Long houseId);
    
        List<WattHourRecord> select(Long houseId, Long startTime, Long endTime, int limit, int offset);
    
        int insert(WattHourRecord wattHourRecord);
    
        int alter(WattHourRecord wattHourRecord);
    }

`WattHourRecordServiceImpl .java`

import com.keyueli.common.core.domain.entity.WattHourRecord;
    import com.keyueli.common.utils.DruidPoolBuilder;
    import com.keyueli.system.service.IWattHourRecordService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.sql.Connection;
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.sql.Statement;
    import java.util.ArrayList;
    import java.util.List;
    
    @Service
    //@DataSource(value = DataSourceType.SLAVE)
    public class WattHourRecordServiceImpl implements IWattHourRecordService {
        
        private static final Logger log = LoggerFactory.getLogger(WattHourRecordServiceImpl.class);
    
    
        @Autowired
        private DruidPoolBuilder druidPoolBuilder;
    
    // 建表  CREATE TABLE keyueli.watt_hour_record(ts timestamp, addr nchar(12), last_value float, current_value float, use_value float, unit_price float, use_price float, balances float, record_type int) tags(note nchar(32));
    
    
        @Override
        public WattHourRecord sum(Long houseId, Long startTime, Long endTime) {
        
            String sql = String.format("select sum(use_value) as use_value, sum(use_price) as use_price from k%d " +
                    "where ts>%d and ts<%d;", houseId, startTime, endTime);
            log.info("SQL >>> " + sql);
            try (Connection conn = druidPoolBuilder.getDataSource().getConnection()) {
        
                try (Statement stmt = conn.createStatement()) {
        
                    ResultSet resultSet = stmt.executeQuery(sql);
                    WattHourRecord wattHourRecord = new WattHourRecord();
                    if (resultSet.next()){
        
                        wattHourRecord.setHouseId(houseId);
                        wattHourRecord.setUseValue(resultSet.getFloat("use_value"));
                        wattHourRecord.setUsePrice(resultSet.getFloat("use_price"));
                    }
                    return wattHourRecord;
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            } catch (SQLException e) {
        
                e.printStackTrace();
            }
            return null;
        }
    
        @Override
        public WattHourRecord lastOne(Long houseId) {
        
            String sql = String.format("select last_row(ts) as ts, last_row(addr) as addr, last_row(last_value) as last_value, last_row(current_value) as current_value, last_row(use_value) as use_value, " +
                    "last_row(unit_price) as unit_price, last_row(use_price) as use_price, last_row(balances) as balances, last_row(record_type) as record_type, " +
                    "last_row(group_id) as group_id, last_row(user_id) as user_id from k%d;", houseId);
            log.info("SQL >>> " + sql);
            try (Connection conn = druidPoolBuilder.getDataSource().getConnection()) {
        
                try (Statement stmt = conn.createStatement()) {
        
                    ResultSet resultSet = stmt.executeQuery(sql);
                    WattHourRecord wattHourRecord = new WattHourRecord();
                    if (resultSet.next()){
        
                        wattHourRecord.setHouseId(houseId);
                        wattHourRecord.setTs(resultSet.getTimestamp("ts"));
                        wattHourRecord.setAddr(resultSet.getString("addr"));
                        wattHourRecord.setLastValue(resultSet.getFloat("last_value"));
                        wattHourRecord.setCurrentValue(resultSet.getFloat("current_value"));
                        wattHourRecord.setUseValue(resultSet.getFloat("use_value"));
                        wattHourRecord.setUnitPrice(resultSet.getFloat("unit_price"));
                        wattHourRecord.setUsePrice(resultSet.getFloat("use_price"));
                        wattHourRecord.setBalances(resultSet.getFloat("balances"));
                        wattHourRecord.setRecordType(resultSet.getInt("record_type"));
                        wattHourRecord.setGroupId(resultSet.getInt("group_id"));
                        wattHourRecord.setUserId(resultSet.getInt("user_id"));
                    }
                    log.info(wattHourRecord.toString());
                    return wattHourRecord;
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            } catch (SQLException e) {
        
                e.printStackTrace();
            }
            return null;
        }
    
        @Override
        public List<WattHourRecord> select(Long houseId, Long startTime, Long endTime, int limit, int offset) {
        
            String sql = String.format("select ts,addr,last_value,current_value,use_value,unit_price,use_price,balances,record_type,group_id,user_id from k%d order by ts desc limit %d offset %d", houseId, limit, offset);
            log.info("SQL >>> " + sql);
            try (Connection conn = druidPoolBuilder.getDataSource().getConnection()) {
        
                try (Statement stmt = conn.createStatement()) {
        
                    ResultSet resultSet = stmt.executeQuery(sql);
                    List<WattHourRecord> wattHourRecordList = new ArrayList<>();
                    while (resultSet.next()){
        
                        WattHourRecord wattHourRecord = new WattHourRecord();
                        wattHourRecord.setHouseId(houseId);
                        wattHourRecord.setTs(resultSet.getTimestamp("ts"));
                        wattHourRecord.setAddr(resultSet.getString("addr"));
                        wattHourRecord.setLastValue(resultSet.getFloat("last_value"));
                        wattHourRecord.setCurrentValue(resultSet.getFloat("current_value"));
                        wattHourRecord.setUseValue(resultSet.getFloat("use_value"));
                        wattHourRecord.setUnitPrice(resultSet.getFloat("unit_price"));
                        wattHourRecord.setUsePrice(resultSet.getFloat("use_price"));
                        wattHourRecord.setBalances(resultSet.getFloat("balances"));
                        wattHourRecord.setRecordType(resultSet.getInt("record_type"));
                        wattHourRecord.setGroupId(resultSet.getInt("group_id"));
                        wattHourRecord.setUserId(resultSet.getInt("user_id"));
                        wattHourRecordList.add(wattHourRecord);
                        log.info(wattHourRecord.toString());
                    }
                    return wattHourRecordList;
                } catch (SQLException e) {
        
                    e.printStackTrace();
                }
            } catch (SQLException e) {
        
                e.printStackTrace();
            }
            return null;
        }
    
        @Override
        public int insert(WattHourRecord wattHourRecord) {
        
            log.info(wattHourRecord.toString());
            String sql = String.format("insert into k%d USING keyueli.watt_hour_record tags('%s', %d, %d) values(%d, '%s', %f, %f, %f, %f, %f, %f, %d)", wattHourRecord.getHouseId(), "", wattHourRecord.getGroupId(), wattHourRecord.getUserId(),
                    wattHourRecord.getTs().getTime(), wattHourRecord.getAddr(), wattHourRecord.getLastValue(), wattHourRecord.getCurrentValue(), wattHourRecord.getUseValue(),
                    wattHourRecord.getUnitPrice(), wattHourRecord.getUsePrice(), wattHourRecord.getBalances(), wattHourRecord.getRecordType());
            return druidPoolBuilder.executeUpdate(sql);
        }
    
        @Override
        public int alter(WattHourRecord wattHourRecord) {
        
            log.info(wattHourRecord.toString());
            String sql = String.format("ALTER TABLE k%d set tag group_id=%d", wattHourRecord.getHouseId(), wattHourRecord.getGroupId());
            druidPoolBuilder.executeUpdate(sql);
            sql = String.format("ALTER TABLE k%d set tag user_id=%d", wattHourRecord.getHouseId(), wattHourRecord.getUserId());
            return druidPoolBuilder.executeUpdate(sql);
        }
    }

## 八、总结 ##

*  设备通过MQTT上报数据，通过SpringBoot将数据写入到时序数据库TDengine
 *  用户通过App/小程序查询电表数据，通过https接口，通过SpringBoot的Controller层再调用Service层代码函数，实现电表数据记录查询和电表数据统计查询。
 *  本架构缺少MyBatis实现XML、SQL等对象关系映射功能。SQL直接写在Service层代码中，虽然略显草率，但是功能实现更为直接，瑕不掩瑜。

***觉得好，就一键三连呗（点赞+收藏+关注）***

[TDengine]: https://blog.csdn.net/kangweijian/article/details/126733066