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

修改超时时间

master
Cui YongXing 2024-09-09 16:32:59 +08:00
parent 713a354e4f
commit 7c6f337941
14 changed files with 45 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cloud-task-common/pom.xml
View File

@ -29,5 +29,10 @@
<groupId>com.muyu</groupId>
<artifactId>cloud-rule-common</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-etl</artifactId>
<version>1.0.0</version>
</dependency>
</dependencies>
</project>

4
cloud-task-common/src/main/java/com/muyu/common/domian/basic/BasicTask.java
View File

@ -1,7 +1,7 @@
package com.muyu.common.domian.basic;
import com.muyu.common.domain.DataValue;
import com.muyu.common.domian.basic.abstracts.DataTaskHandler;
public interface BasicTask<T> {
void set(T dataValue);

2
cloud-task-common/src/main/java/com/muyu/common/domian/basic/abstracts/DataTaskHandler.java
View File

@ -1,6 +1,6 @@
package com.muyu.common.domian.basic.abstracts;
import com.muyu.common.domain.DataValue;
public class DataTaskHandler {
private static final ThreadLocal<Object> threadLocal = new ThreadLocal<>();

3
cloud-task-common/src/main/java/com/muyu/common/domian/basic/abstracts/DataTaskHandlerRowHandler.java
View File

@ -1,6 +1,7 @@
package com.muyu.common.domian.basic.abstracts;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
public class DataTaskHandlerRowHandler {
public DataTaskHandlerRowHandler() {

3
cloud-task-common/src/main/java/com/muyu/common/domian/basic/abstracts/DataTaskValueHandler.java
View File

@ -1,6 +1,7 @@
package com.muyu.common.domian.basic.abstracts;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
public class DataTaskValueHandler {
public DataTaskValueHandler() {

6
cloud-task-remote/pom.xml
View File

@ -25,6 +25,12 @@
<groupId>com.muyu</groupId>
<artifactId>cloud-rule-common</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-etl</artifactId>
<version>1.0.0</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

3
cloud-task-remote/src/main/java/com/muyu/remote/feign/DatasourceFeign.java
View File

@ -1,7 +1,8 @@
package com.muyu.remote.feign;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
import com.muyu.remote.feign.Factory.DatasourceFeignFactory;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.cloud.openfeign.FeignClient;

2
cloud-task-remote/src/main/java/com/muyu/remote/feign/Factory/DatasourceFeignFactory.java
View File

@ -1,7 +1,7 @@
package com.muyu.remote.feign.Factory;
import com.muyu.common.core.domain.Result;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
import com.muyu.remote.feign.DatasourceFeign;
import lombok.extern.log4j.Log4j2;
import org.springframework.cloud.openfeign.FallbackFactory;

2
cloud-task-remote/src/main/java/com/muyu/remote/feign/Factory/RuleFactory.java
View File

@ -1,7 +1,7 @@
package com.muyu.remote.feign.Factory;
import com.muyu.common.core.domain.Result;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
import com.muyu.remote.feign.DatasourceFeign;
import com.muyu.remote.feign.RuleFeign;
import com.muyu.rule.common.domain.RuleEngineVersion;

3
cloud-task-remote/src/main/java/com/muyu/remote/feign/RuleFeign.java
View File

@ -1,7 +1,8 @@
package com.muyu.remote.feign;
import com.muyu.common.core.domain.Result;
import com.muyu.common.domain.DataValue;
import com.muyu.etl.domain.DataValue;
import com.muyu.remote.feign.Factory.RuleFactory;
import com.muyu.rule.common.domain.RuleEngineVersion;
import io.swagger.v3.oas.annotations.Operation;

6
cloud-task-server/pom.xml
View File

@ -100,7 +100,11 @@
<groupId>com.muyu</groupId>
<artifactId>cloud-datasources-common</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-datasources-client</artifactId>
<version>1.0.0</version>
</dependency>
</dependencies>
<build>

23
cloud-task-server/src/main/java/com/muyu/task/server/service/impl/TaskInfoServiceImpl.java
View File

@ -4,14 +4,19 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.client.mysql.MySqlDataSource;
import com.muyu.client.mysql.MySqlQuery;
import com.muyu.common.core.domain.Result;
import com.muyu.common.core.utils.StringUtils;
import com.muyu.common.domain.DataValue;
import com.muyu.common.data.base.BaseQuery;
import com.muyu.common.domian.*;
import com.muyu.common.domian.basic.abstracts.DataTaskAbstracts;
import com.muyu.common.domian.enums.Weight;
import com.muyu.common.domian.req.TaskInfoListReq;
import com.muyu.common.domian.resp.TaskInfoResp;
import com.muyu.common.pool.MysqlPool;
import com.muyu.etl.domain.DataValue;
import com.muyu.remote.feign.DatasourceFeign;
import com.muyu.remote.feign.RuleFeign;
import com.muyu.rule.common.domain.RuleEngineVersion;
@ -324,7 +329,8 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
return joint;
}
@Resource
private MySqlDataSource mySqlDataSource;
@NotNull
private void getString(Long pageNum,
String fieName,
@ -338,12 +344,13 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
Long taskId) {
String sqlSelect = " SELECT " + fieName + " FROM " + joint + " LIMIT " + PAGE_SIZE + " OFFSET " + pageNum;
log.info(sqlSelect);
mySqlDataSource.setQuery(MySqlQuery.builder().dataSourceId(basicId).sql(sqlSelect).one(one).two(two).build());
DataValue[][] rows = mySqlDataSource.getRows();
//log.info("执行{}查询的方法",sqlSelect);
Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect, one, two);
log.info(tableValueResult);
DataValue[][] data = tableValueResult.getData();
// Result<DataValue[][]> tableValueResult = datasourceFeign.findTableValueToArray(basicId, sqlSelect, one, two);
//log.info(tableValueResult);
//
// DataValue[][] data = tableValueResult.getData();
log.info("执行{}查询的方法结束", sqlSelect);
// for (DataValue[] datum : data) {
// for (DataValue dataValue : datum) {
@ -354,7 +361,7 @@ public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> i
// }
// Result result = datasourceFeign.addProduct(newBasicId, tableId, data);
// log.info("{}添加结束", result);
executeTheRule(data, map, newBasicId, tableId, taskId);
executeTheRule(rows, map, newBasicId, tableId, taskId);
}

8
cloud-task-server/src/main/java/com/muyu/task/server/service/impl/TaskInputServiceImpl.java
View File

@ -1,13 +1,9 @@
package com.muyu.task.server.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.common.core.domain.Result;
import com.muyu.common.core.utils.StringUtils;
import com.muyu.common.domain.DataValue;
import com.muyu.common.domian.NodeJoint;
import com.muyu.common.domian.TaskInput;
import com.muyu.common.domian.req.TaskInputListReq;
import com.muyu.remote.feign.DatasourceFeign;
@ -17,10 +13,6 @@ import com.muyu.task.server.service.NodeTableService;
import com.muyu.task.server.service.TaskInputService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.HashSet;
import java.util.List;
@Service
public class TaskInputServiceImpl extends ServiceImpl<TaskInputMapper, TaskInput> implements TaskInputService {

2
cloud-task-server/src/main/java/com/muyu/task/server/thread/OptimizedPrioritizedThreadPool.java
View File

@ -37,7 +37,7 @@ public class OptimizedPrioritizedThreadPool {
// 创建固定大小的线程池
executor = new ThreadPoolExecutor(
totalThreads, totalThreads,
80L, TimeUnit.SECONDS,
180L, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>()
);
highPrioritySemaphore = new Semaphore(defaultHighThreads);