swordmeng
3 weeks ago
21 changed files with 387 additions and 470 deletions
@ -1,4 +1,4 @@ |
|||
package com.huaxing.pojo.entity; |
|||
package com.huaxing.pojo.dto; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@ -1,66 +1,93 @@ |
|||
package com.huaxing.data.database.domain; |
|||
package com.huaxing.pojo.dto; |
|||
|
|||
import com.huaxing.pojo.entity.TableColumnDTO; |
|||
import lombok.Data; |
|||
|
|||
import javax.validation.constraints.NotEmpty; |
|||
import java.util.Iterator; |
|||
import java.util.List; |
|||
import java.util.Objects; |
|||
|
|||
/** |
|||
* @ProjectName: iot-data-bridge |
|||
* @Package: com.huaxing.data.database.domain |
|||
* @ProjectName: data-bridge |
|||
* @Package: com.huaxing.pojo.entity |
|||
* @ClassName: TableDTO |
|||
* @Author: swordmeng8@163.com |
|||
* @Description: 表操作类 |
|||
* @Date: 2025/1/15 16:03 |
|||
* @Description: 表结构接参类 |
|||
* @Date: 2025/2/12 17:40 |
|||
* @Version: 1.0 |
|||
*/ |
|||
|
|||
@Data |
|||
public class TableDTO { |
|||
|
|||
/** |
|||
* 数据库名 |
|||
*/ |
|||
@NotEmpty(message = "数据库名不能为空") |
|||
private String databaseName; |
|||
|
|||
/** |
|||
* 表名 |
|||
* dfs表名 |
|||
*/ |
|||
private String tableName; |
|||
@NotEmpty(message = "Dfs表名不能为空") |
|||
private String dfsTableName; |
|||
|
|||
/** |
|||
* stream表名 |
|||
*/ |
|||
@NotEmpty(message = "Stream表名不能为空") |
|||
private String streamTableName; |
|||
|
|||
/** |
|||
* 表描述 |
|||
*/ |
|||
private String tableDesc; |
|||
|
|||
/** |
|||
* 表列信息 |
|||
*/ |
|||
private List<TableColumnDTO> tableColumnList; |
|||
|
|||
/** |
|||
* 获取列名的拼接字符串并返回 |
|||
* |
|||
* @param tableColumnList |
|||
* @return String |
|||
*/ |
|||
public String handleTableColumnName(List<TableColumnDTO> tableColumnList) { |
|||
return tableColumnList.stream().map(TableColumnDTO::getColumnName).reduce((a, b) -> a + "`" + b).get(); |
|||
} |
|||
|
|||
/** |
|||
* 获取列类型的拼接字符串并返回 |
|||
* |
|||
* @param tableColumnList |
|||
* @return String |
|||
*/ |
|||
public String handleTableColumnType(List<TableColumnDTO> tableColumnList) { |
|||
return tableColumnList.stream().map(TableColumnDTO::getColumnType).reduce((a, b) -> a + "," + b).get(); |
|||
} |
|||
|
|||
/** |
|||
* 获取列与描述的拼接字符串并返回 |
|||
* |
|||
* @param tableColumnList |
|||
* @return String |
|||
*/ |
|||
public String handleTableColumnDesc(List<TableColumnDTO> tableColumnList) { |
|||
String columnDesc = ""; |
|||
StringBuilder columnDesc = new StringBuilder(); |
|||
// 获取列与描述的拼接字符串并返回
|
|||
Iterator<TableColumnDTO> iterator = tableColumnList.iterator(); |
|||
while (iterator.hasNext()) { |
|||
TableColumnDTO tableColumn = iterator.next(); |
|||
if (Objects.isNull(tableColumn.getColumnDesc())) { |
|||
columnDesc += tableColumn.getColumnName() + ":" + "\"-\""; |
|||
columnDesc.append(tableColumn.getColumnName()).append(":").append("\"-\""); |
|||
} else { |
|||
columnDesc += tableColumn.getColumnName() + ":\"" + tableColumn.getColumnDesc()+"\""; |
|||
columnDesc.append(tableColumn.getColumnName()).append(":\"").append(tableColumn.getColumnDesc()).append("\""); |
|||
} |
|||
if (iterator.hasNext()) { |
|||
columnDesc += ","; |
|||
columnDesc.append(","); |
|||
} |
|||
} |
|||
return columnDesc; |
|||
return columnDesc.toString(); |
|||
} |
|||
|
|||
} |
|||
@ -1,16 +0,0 @@ |
|||
package com.huaxing.pojo.entity; |
|||
|
|||
import lombok.Data; |
|||
|
|||
/** |
|||
* @ProjectName: data-bridge |
|||
* @Package: com.huaxing.pojo.entity |
|||
* @ClassName: TableDTO |
|||
* @Author: swordmeng8@163.com |
|||
* @Description: 表结构接参类 |
|||
* @Date: 2025/2/12 17:40 |
|||
* @Version: 1.0 |
|||
*/ |
|||
@Data |
|||
public class TableDTO { |
|||
} |
|||
@ -1,240 +0,0 @@ |
|||
package com.huaxing.data.database.controller; |
|||
|
|||
import com.huaxing.data.storage.domain.DataAnalysisDTO; |
|||
import com.huaxing.data.storage.service.IDeviceDataQueryDfsService; |
|||
import com.huaxing.data.storage.service.IDeviceDataQueryStreamService; |
|||
import com.huaxing.data.storage.service.IDeviceDataStoredService; |
|||
import com.huaxing.data.database.domain.TableDTO; |
|||
import com.huaxing.data.database.service.ITableStructureService; |
|||
import com.huaxing.common.util.JacksonUtil; |
|||
import com.huaxing.data.database.template.ISqlTemplateService; |
|||
import com.huaxing.mqtt.processor.MqttMessageSender; |
|||
import com.huaxing.common.result.ResultVo; |
|||
import com.huaxing.pojo.entity.TableColumnDTO; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
|
|||
import java.util.*; |
|||
|
|||
|
|||
/** |
|||
* @ProjectName: iot-data-bridge |
|||
* @Package: com.huaxing.data.storage.controller |
|||
* @ClassName: TestController |
|||
* @Author: swordmeng8@163.com |
|||
* @Description: 测试controller |
|||
* @Date: 2025/1/15 17:08 |
|||
* @Version: 1.0 |
|||
*/ |
|||
@Slf4j |
|||
@RestController |
|||
@RequestMapping("/api/database") |
|||
@SuppressWarnings("all") |
|||
public class DatabaseTestController { |
|||
|
|||
@Autowired |
|||
private MqttMessageSender messageSender; |
|||
final IDeviceDataStoredService dataStoredService; |
|||
final IDeviceDataQueryDfsService dataQueryDfsService; |
|||
final IDeviceDataQueryStreamService dataQueryStreamService; |
|||
final ITableStructureService tableStructureService; |
|||
|
|||
|
|||
// =============================================== 测试构造器 ======================================
|
|||
public DatabaseTestController(IDeviceDataStoredService dataStoredService, IDeviceDataQueryDfsService dataQueryDfsService, IDeviceDataQueryStreamService dataQueryStreamService, ITableStructureService tableStructureService, ISqlTemplateService tableTemplateService) { |
|||
this.dataStoredService = dataStoredService; |
|||
this.dataQueryDfsService = dataQueryDfsService; |
|||
this.dataQueryStreamService = dataQueryStreamService; |
|||
this.tableStructureService = tableStructureService; |
|||
} |
|||
|
|||
|
|||
// =============================================== 测试插入数据 ======================================
|
|||
|
|||
/** |
|||
* 测试数据插入 |
|||
*/ |
|||
@GetMapping(value = "/testInsert") |
|||
public void testInsert() { |
|||
String tableName = "MyTable2Stream"; |
|||
String sql = "INSERT INTO " + tableName + " (time, projectId, deviceId, test) VALUES (2025.01.11 00:00:00,'48', '0jZU2102_0806_0011', 'testVal')"; |
|||
dataStoredService.execute(sql); |
|||
log.info("SUCCESS"); |
|||
} |
|||
|
|||
// =============================================== 测试Dfs表查询 ======================================
|
|||
|
|||
/** |
|||
* 测试Dfs表查询 |
|||
*/ |
|||
@GetMapping(value = "/testSelectDfs") |
|||
public ResultVo<List<Map<String, Object>>> testSelectDfs() { |
|||
String dbPath = "dfs://ZbDB"; |
|||
String sql = String.format("select * from loadTable('%s','%s')", dbPath, "ZbWaterMeter1Dfs"); |
|||
return ResultVo.ok(dataQueryDfsService.selectList(sql)); |
|||
} |
|||
|
|||
// =============================================== 测试订阅流表查询 ======================================
|
|||
|
|||
/** |
|||
* 测试订阅流表查询 |
|||
*/ |
|||
@GetMapping(value = "/testSelectStream") |
|||
public ResultVo<List<Map<String, Object>>> testSelectStream() { |
|||
String sql = "select * from WaterMeterTset1Stream"; |
|||
return ResultVo.ok(dataQueryStreamService.selectList(sql)); |
|||
} |
|||
|
|||
// =============================================== 给指定的流表增加列字段 ======================================
|
|||
|
|||
/** |
|||
* 测试给指定的流表增加列字段 |
|||
*/ |
|||
@GetMapping(value = "/testStreamAddColumn") |
|||
public void testStreamAddColumn() { |
|||
String tableName = "WaterMeterTsetStream"; |
|||
String columnName = "test"; |
|||
String columnType = "STRING"; |
|||
String columnDesc = "test"; |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setTableName(tableName); |
|||
tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc))); |
|||
try { |
|||
tableStructureService.addStreamColumns(tableDTO); |
|||
log.info("SUCCESS"); |
|||
} catch (Exception e) { |
|||
log.error("FAIL"); |
|||
e.printStackTrace(); |
|||
} |
|||
} |
|||
|
|||
// =============================================== 给指定的Dfs表增加列字段 ======================================
|
|||
/** |
|||
* 测试给指定的Dfs表增加列字段 |
|||
*/ |
|||
@GetMapping(value = "/testDfsAddColumn") |
|||
public void testDfsAddColumn() { |
|||
String tableName = "WaterMeterTsetDfs"; |
|||
String columnName = "test"; |
|||
String columnType = "STRING"; |
|||
String columnDesc = "test描述"; |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setTableName(tableName); |
|||
tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc))); |
|||
try { |
|||
tableStructureService.addDfsColumns(tableDTO); |
|||
log.info("SUCCESS"); |
|||
} catch (Exception e) { |
|||
log.error("FAIL"); |
|||
e.printStackTrace(); |
|||
} |
|||
} |
|||
|
|||
|
|||
// ===================================== mqtt消息发送-测试数据入库 START ==============================
|
|||
|
|||
/** |
|||
* 测试发送mqtt消息 |
|||
*/ |
|||
@GetMapping(value = "/testSendMessage") |
|||
public void testSendMessage() { |
|||
for (int i = 0; i < 2; i++) { |
|||
List<Map<String, Object>> dataList = new ArrayList<>(); |
|||
dataList.add(handleMapByIndex(i)); |
|||
DataAnalysisDTO dataAnalysisDTO = DataAnalysisDTO.builder().tableName( "WaterMeterTsetStream").dataList(dataList).build(); |
|||
messageSender.send("iot/test1/in-storage", JacksonUtil.objectStr(dataAnalysisDTO)); |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 根据索引生成测试数据 |
|||
* @param index 索引 |
|||
* @return |
|||
*/ |
|||
private Map<String, Object> handleMapByIndex(int index) { |
|||
Map<String, Object> map = new HashMap<>(); |
|||
map.put("time", new Date()); |
|||
map.put("projectId", "0jZU2102"); |
|||
map.put("deviceId", "0jZU2102_0806_0011"); |
|||
map.put("WM_WFA", 124.656 + index); |
|||
map.put("WM_WFA_Unit", "m³"); |
|||
map.put("test", "test"); |
|||
return map; |
|||
} |
|||
|
|||
// =============================================== 创建Dfs表 Stream流表 取消订阅 添加订阅 创建 ======================================
|
|||
|
|||
/** |
|||
* 测试创建Dfs表 |
|||
*/ |
|||
@GetMapping(value = "/testCreateDfsTable") |
|||
public void testCreateDfsTable() { |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setDatabaseName("ZbDB"); |
|||
tableDTO.setTableName("Test1"); |
|||
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|||
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|||
tableColumnList.add(new TableColumnDTO("test1", "STRING","test1")); |
|||
tableColumnList.add(new TableColumnDTO("test2", "STRING","test2")); |
|||
tableDTO.setTableColumnList(tableColumnList); |
|||
tableStructureService.createDfsTable(tableDTO); |
|||
} |
|||
|
|||
/** |
|||
* 测试创建流表 |
|||
*/ |
|||
@GetMapping(value = "/testCreateStreamTable") |
|||
public void testCreateStreamTable() { |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setDatabaseName("ZbDB"); |
|||
tableDTO.setTableName("Test1"); |
|||
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|||
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|||
tableColumnList.add(new TableColumnDTO("test1", "STRING","test1")); |
|||
tableColumnList.add(new TableColumnDTO("test2", "STRING","test2")); |
|||
tableDTO.setTableColumnList(tableColumnList); |
|||
tableStructureService.createStreamTable(tableDTO); |
|||
} |
|||
|
|||
/** |
|||
* 取消订阅流表 |
|||
*/ |
|||
@GetMapping(value = "/testUnsubscribeStreamTable") |
|||
public void testUnsubscribeStreamTable() { |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setDatabaseName("ZbDB"); |
|||
tableDTO.setTableName("Test1"); |
|||
tableStructureService.unsubscribeStreamTable(tableDTO); |
|||
} |
|||
|
|||
/** |
|||
* 订阅流表 |
|||
*/ |
|||
@GetMapping(value = "/testSubscribeStreamTable") |
|||
public void testSubscribeStreamTable() { |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setDatabaseName("ZbDB"); |
|||
tableDTO.setTableName("Test1"); |
|||
tableStructureService.subscribeStreamTable(tableDTO); |
|||
} |
|||
|
|||
/** |
|||
* 创建Dfs和Stream表并添加订阅 |
|||
*/ |
|||
@GetMapping(value = "/testCreateDfsAndStreamTableSubscribe") |
|||
public void testCreateDfsAndStreamTableSubscribe() { |
|||
TableDTO tableDTO = new TableDTO(); |
|||
tableDTO.setDatabaseName("ZbDB"); |
|||
tableDTO.setTableName("MyTe2"); |
|||
tableDTO.setTableDesc("MyTe2测试表"); |
|||
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|||
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|||
tableDTO.setTableColumnList(tableColumnList); |
|||
tableStructureService.createDfsAndStreamTableSubscribe(tableDTO); |
|||
} |
|||
|
|||
|
|||
} |
|||
Loading…
Reference in new issue