TreeData
/
cloud-etl-source
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改方法

master
lwj 2024-09-10 11:03:29 +08:00
parent 587f61aab5
commit 9b123a8e5d
13 changed files with 78 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cloud-etl-client/src/main/java/com/muyu/basic/SourceConfig.java
View File

@ -3,5 +3,5 @@ package com.muyu.basic;
import com.muyu.domain.Source;
public interface SourceConfig {
public Source queryById(Long id);
public Source findSourceById(Long id);
}

6
cloud-etl-client/src/main/java/com/muyu/basic/impl/SourceConfigImpl.java
View File

@ -11,9 +11,9 @@ public class SourceConfigImpl implements SourceConfig {
@Autowired
private SourceRemote sourceRemote;
@Override
public Source queryById(Long id) {
return sourceRemote.queryById(id).getData();
public Source findSourceById(Long id) {
return sourceRemote.findSourceById(id).getData();
}
}

4
cloud-etl-client/src/main/java/com/muyu/mysql/MySqlDataSource.java
View File

@ -22,7 +22,7 @@ public class MySqlDataSource extends BaseDataAbsSource {
@Autowired
private SourceConfig sourceRemote;
public SourceConfig sourceRemote;
// SourceService sourceService = SpringUtils.getBean(SourceService.class);
@ -359,7 +359,7 @@ public class MySqlDataSource extends BaseDataAbsSource {
Long dataSourceId = query.getDataSourceId();
ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
Source dataSources = sourceRemote.queryById(dataSourceId);
Source dataSources = sourceRemote.findSourceById(dataSourceId);
HikariConfig hikariConfig = new HikariConfig();
hikariConfig.setPoolName("HikariCP 连接池");

2
cloud-etl-client/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1,2 @@
com.muyu.mysql.MySqlDataSource
com.muyu.basic.impl.SourceConfigImpl

6
cloud-etl-common/pom.xml
View File

@ -24,6 +24,12 @@
<artifactId>cloud-common-core</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-etl</artifactId>
<version>1.0.0</version>
</dependency>
</dependencies>
</project>

7
cloud-etl-remote/src/main/java/com/muyu/remote/SourceRemote.java
View File

@ -8,9 +8,10 @@ import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@EnableFeignClients
@FeignClient(value = "cloud-etl-datasources",fallbackFactory = SourceFactory.class )
@FeignClient(value = "cloud-source",fallbackFactory = SourceFactory.class )
public interface SourceRemote {
@GetMapping("datasources/{id}")
public Result<Source> queryById(@PathVariable("id") Long id);
@GetMapping("/source/findSourceById/{id}")
public Result<Source> findSourceById(@PathVariable("id") Long id);
}

2
cloud-etl-remote/src/main/java/com/muyu/remote/factory/SourceFactory.java
View File

@ -12,7 +12,7 @@ public class SourceFactory implements FallbackFactory<SourceRemote> {
public SourceRemote create(Throwable cause) {
return new SourceRemote() {
@Override
public Result<Source> queryById(Long id) {
public Result<Source> findSourceById(Long id) {
return null;
}
};

2
cloud-etl-remote/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1,2 @@
com.muyu.remote.SourceRemote
com.muyu.remote.factory.SourceFactory

38
cloud-etl-server/pom.xml
View File

@ -48,10 +48,10 @@
<version>2.9.0</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-core</artifactId>
</dependency>
<!-- <dependency>-->
<!-- <groupId>com.muyu</groupId>-->
<!-- <artifactId>cloud-common-core</artifactId>-->
<!-- </dependency>-->
<dependency>
<groupId>com.muyu</groupId>
@ -59,6 +59,13 @@
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-etl</artifactId>
<version>1.0.0</version>
<scope>compile</scope>
</dependency>
<!-- SpringCloud Alibaba Nacos -->
<dependency>
@ -91,10 +98,10 @@
</dependency>
<!-- MuYu Common DataSource -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-datasource</artifactId>
</dependency>
<!-- <dependency>-->
<!-- <groupId>com.muyu</groupId>-->
<!-- <artifactId>cloud-common-datasource</artifactId>-->
<!-- </dependency>-->
<!-- MuYu Common DataScope -->
<dependency>
@ -130,8 +137,23 @@
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-datasource</artifactId>
<version>3.6.3</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-etl-client</artifactId>
<version>1.0.0</version>
<scope>compile</scope>
</dependency>
<!-- <dependency>-->
<!-- <groupId>com.muyu</groupId>-->
<!-- <artifactId>cloud-etl-client</artifactId>-->
<!-- <version>1.0.0</version>-->
<!-- <scope>compile</scope>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>com.muyu</groupId>-->
<!-- <artifactId>cloud-common-rabbit</artifactId>-->

12
cloud-etl-server/src/main/java/com/muyu/cloud/etl/controller/SourceController.java
View File

@ -51,6 +51,18 @@ public class SourceController extends BaseController {
}
/**
* 获取远程调用基本信息
*/
@GetMapping("/findSourceById/{id}")
public Result<Source> findSourceById(@PathVariable("id") Long id) {
Source source= sourceService.getById(id);
return success(source);
}
/**
* 新增基础信息
*/

5
cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/DataValueServiceImpl.java
View File

@ -1,8 +1,6 @@
package com.muyu.cloud.etl.service.impl;
import com.muyu.Hikari.HikariPool;
import com.muyu.cloud.etl.mysql.MySqlDataSource;
import com.muyu.cloud.etl.mysql.MySqlQuery;
import com.muyu.cloud.etl.service.DataValueService;
import com.muyu.cloud.etl.service.SourceService;
import com.muyu.cloud.etl.service.TableInfoService;
@ -11,6 +9,8 @@ import com.muyu.domain.DataValue;
import com.muyu.domain.Source;
import com.muyu.domain.TableInfo;
import com.muyu.domain.enums.DataType;
import com.muyu.mysql.MySqlDataSource;
import com.muyu.mysql.MySqlQuery;
import com.zaxxer.hikari.HikariDataSource;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
@ -318,4 +318,5 @@ public class DataValueServiceImpl implements DataValueService {
return rows;
}
}

18
cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/ProductServiceImpl.java
View File

@ -496,14 +496,16 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
hikariConfig.setPassword(dataSources.getPassword());
hikariConfig.setMinimumIdle(2);
hikariConfig.setMaximumPoolSize(10);
hikariConfig.setMaxLifetime(300000); // 5 minutes
hikariConfig.setConnectionTimeout(30000); // 30 seconds
HikariDataSource hikariDataSource = new HikariDataSource(hikariConfig);
// HikariDataSource hikariDataSource = HikariPool.getHikariDataSource(dataSources);
ExecutorService executorService = Executors.newFixedThreadPool(8);
ExecutorService executorService = Executors.newFixedThreadPool(6);
AtomicInteger addCount = new AtomicInteger();
// 分割数据为较小的批次
List<DataValue[][]> batches = splitData(listList, 3000);
List<DataValue[][]> batches = splitData(listList, 5000);
try (Connection conn = hikariDataSource.getConnection()) {
conn.setAutoCommit(false); // 开启事务
@ -537,7 +539,11 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.HOURS);
if (!executorService.awaitTermination(1, TimeUnit.HOURS)) {
log.warn("Executor service did not terminate within the timeout.");
executorService.shutdownNow();
}
// executorService.awaitTermination(1, TimeUnit.HOURS);
conn.commit(); // 提交事务
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@ -554,7 +560,6 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
private String buildBatchInsertSQL(String tableName, DataValue[][] batch) {
StringBuilder columns = new StringBuilder("(");
StringBuilder values = new StringBuilder("VALUES ");
// 构建字段名
for (DataValue dataValue : batch[0]) {
String key = dataValue.getKey();
@ -562,7 +567,6 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
}
// 删除最后一个逗号和空格
columns.delete(columns.length() - 2, columns.length());
// 构建值部分
for (DataValue[] dataValueList : batch) {
values.append("(");
@ -575,7 +579,6 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
}
// 删除最后一个逗号
values.delete(values.length() - 2, values.length());
// 完成 SQL 插入语句
String sql = "INSERT INTO " + tableName + " " + columns.toString() + ") " + values.toString();
return sql;
@ -586,7 +589,6 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
// 根据业务需求处理 null 值
return "NULL"; // 或者其他默认值
}
if (type == DataType.VARCHAR || type == DataType.TEXT) {
return "'" + value.toString().replace("'", "''") + "'";
} else if (type == DataType.BIGINT) {
@ -617,10 +619,8 @@ public int addProduct(Long basicId, Long tableId, DataValue[][] listList) {
DataValue[][] batch = Arrays.copyOfRange(listList, start, end);
batches.add(batch);
}
return batches;
}
// 关闭数据源
private void close(HikariDataSource dataSource) {
dataSource.close();

2
cloud-etl-server/src/main/resources/bootstrap.yml
View File

@ -10,6 +10,8 @@ nacos:
namespace: cloud-2112
# Spring
spring:
main:
allow-bean-definition-overriding: true
application:
# 应用名称