UDI
/
udi-wms-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 同步任务和线程池参数的自定义配置

Browse Source
dev_2.5_scan^2
chenhc 3 months ago
parent 7415b0c02d
commit 4ce7bcb950
6 changed files with 146 additions and 63 deletions
Show Stats Download Patch File Download Diff File

70
src/main/java/com/glxp/api/config/TaskPoolConfig.java
Unescape Escape View File

@ -1,5 +1,6 @@
package com.glxp.api.config;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@ -11,30 +12,65 @@ import java.util.concurrent.ThreadPoolExecutor;
/**
* 配置线程池
*/
@Configuration
@EnableAsync
@Configuration
@EnableConfigurationProperties(ThreadPoolConfigProperties.class)
public class TaskPoolConfig {
// @Bean(name = "taskExecutor")
// @Primary
// public ThreadPoolTaskExecutor taskExecutor() {
// // 获取当前主机的cpu核心数
// int threadCount = Runtime.getRuntime().availableProcessors();
// ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
// //核心池的大小
// taskExecutor.setCorePoolSize(threadCount);
// //线程池最大线程数
// taskExecutor.setMaxPoolSize(threadCount * 2);
// //队列最大长度
// taskExecutor.setQueueCapacity(200);
// //线程空闲时间
// taskExecutor.setKeepAliveSeconds(60);
// //配置线程前缀
// taskExecutor.setThreadNamePrefix("custom_executor");
// //配置拒绝策略
// taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// //执行初始化
// taskExecutor.initialize();
// return taskExecutor;
// }
@Bean(name = "taskExecutor")
@Primary
public ThreadPoolTaskExecutor taskExecutor() {
// 获取当前主机的cpu核心数
int threadCount = Runtime.getRuntime().availableProcessors();
public ThreadPoolTaskExecutor taskExecutor(ThreadPoolConfigProperties properties) {
// 动态获取CPU核心数
final int cpuCores = Runtime.getRuntime().availableProcessors();
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
//核心池的大小
taskExecutor.setCorePoolSize(threadCount);
//线程池最大线程数
taskExecutor.setMaxPoolSize(threadCount * 2);
//队列最大长度
taskExecutor.setQueueCapacity(200);
//线程空闲时间
taskExecutor.setKeepAliveSeconds(60);
//配置线程前缀
taskExecutor.setThreadNamePrefix("custom_executor");
//配置拒绝策略
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//执行初始化
// 核心线程数（配置优先，默认CPU核心数）
taskExecutor.setCorePoolSize(
properties.getCorePoolSize() != null ?
properties.getCorePoolSize() : cpuCores
);
// 最大线程数（配置优先，默认CPU核心数*2）
taskExecutor.setMaxPoolSize(
properties.getMaxPoolSize() != null ?
properties.getMaxPoolSize() : cpuCores * 2
);
// 其他固定配置项
taskExecutor.setQueueCapacity(properties.getQueueCapacity());
taskExecutor.setKeepAliveSeconds(properties.getKeepAliveSeconds());
taskExecutor.setThreadNamePrefix(properties.getThreadNamePrefix());
// 拒绝策略保持原样
taskExecutor.setRejectedExecutionHandler(
new ThreadPoolExecutor.CallerRunsPolicy()
);
taskExecutor.initialize();
return taskExecutor;
}
}

36
src/main/java/com/glxp/api/config/ThreadPoolConfigProperties.java
Unescape Escape View File

@ -0,0 +1,36 @@
package com.glxp.api.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
// 1. 创建配置类（建议放在config包下）
@ConfigurationProperties(prefix = "thread.pool")
@Data
public class ThreadPoolConfigProperties {
/**
* 核心线程数（默认CPU核心数）
*/
private Integer corePoolSize;
/**
* 最大线程数（默认CPU核心数*2）
*/
private Integer maxPoolSize;
/**
* 队列容量（默认200）
*/
private int queueCapacity = 200;
/**
* 线程空闲时间（秒，默认60）
*/
private int keepAliveSeconds = 60;
/**
* 线程名前缀（默认"custom_executor_"）
*/
private String threadNamePrefix = "custom_executor_";
}

2
src/main/java/com/glxp/api/service/sync/WebSocketComponent.java
Unescape Escape View File

@ -65,7 +65,7 @@ public class WebSocketComponent implements CommandLineRunner {
}
}
if (message.equals(SocketMsgType.DL_ALL_DATA)) {
heartTaskService.pullData();
heartTaskService.pullData(null);
}
if (message.contains(SocketMsgType.STAT_DATA)) {
cn.hutool.json.JSONObject obj = JSONUtil.parseObj(message);

72
src/main/java/com/glxp/api/task/SyncHeartService.java
Unescape Escape View File

@ -25,17 +25,20 @@ public class SyncHeartService {
private SyncDataSetDao syncDataSetDao;
public void syncProcess() {
//查询数据同步设置
pushData();
pushOrder();
pullData();
}
public void pushData() {
SyncDataSetEntity syncDataSetEntity = syncDataSetDao.selectSet();
if (!syncDataSetEntity.isDownstreamEnable()) {
return;
}
//查询数据同步设置
pushData(syncDataSetEntity);
pushOrder(syncDataSetEntity);
pullData(syncDataSetEntity);
}
public void pushData(SyncDataSetEntity syncDataSetEntity) {
//定时上传最近更新基础数据至上游轮询时间
long timeInterval1 = syncDataSetEntity.getSyncTime() * 6 * 1000L;
long curTime1 = System.currentTimeMillis();
@ -75,11 +78,8 @@ public class SyncHeartService {
});
}
public void pushOrder() {
SyncDataSetEntity syncDataSetEntity = syncDataSetDao.selectSet();
if (!syncDataSetEntity.isDownstreamEnable()) {
return;
}
public void pushOrder(SyncDataSetEntity syncDataSetEntity) {
//定时上传最近更新单据数据至上游轮询时间
long timeInterval2 = syncDataSetEntity.getOrderSyncTime() * 6 * 1000L;
long curTime2 = System.currentTimeMillis();
@ -100,23 +100,43 @@ public class SyncHeartService {
}
}
public void pullData() {
try {
heartService.dlAllOrder();
} catch (Exception e) {
e.printStackTrace();
log.error(ExceptionUtils.getStackTrace(e));
public void pullData(SyncDataSetEntity syncDataSetEntity) {
if (syncDataSetEntity == null) {
syncDataSetEntity = syncDataSetDao.selectSet();
if (!syncDataSetEntity.isDownstreamEnable()) {
return;
}
}
try {
heartService.dlAllDiProducts();
} catch (Exception e) {
e.printStackTrace();
log.error(ExceptionUtils.getStackTrace(e));
//定时下载上游最近更新数据轮询时间
long timeInterval = syncDataSetEntity.getSyncDownloadTime() * 6 * 100;
long curTime = System.currentTimeMillis();
Long lastTime = (Long) redisUtil.get("SPS_SYNC_DOWNLOAD_DATA");
if (lastTime == null) {
lastTime = System.currentTimeMillis();
redisUtil.set("SPS_SYNC_DOWNLOAD_DATA", lastTime);
}
Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
heartService.pullData(i);
});
redisUtil.set("SPS_SYNC_DOWNLOAD_DATA", System.currentTimeMillis());
if (curTime - lastTime > timeInterval) {
try {
heartService.dlAllOrder();
} catch (Exception e) {
e.printStackTrace();
log.error(ExceptionUtils.getStackTrace(e));
}
try {
heartService.dlAllDiProducts();
} catch (Exception e) {
e.printStackTrace();
log.error(ExceptionUtils.getStackTrace(e));
}
Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
heartService.pullData(i);
});
redisUtil.set("SPS_SYNC_DOWNLOAD_DATA", System.currentTimeMillis());
}
}
}

19
src/main/java/com/glxp/api/task/SyncHeartTask.java
Unescape Escape View File

@ -50,26 +50,7 @@ public class SyncHeartTask implements SchedulingConfigurer {
private void process() {
log.info("数据同步心跳--");
syncHeartService.syncProcess();
SyncDataSetEntity syncDataSetEntity = syncDataSetDao.selectSet();
if (!syncDataSetEntity.isDownstreamEnable()) {
return;
}
//定时下载上游最近更新数据轮询时间
long timeInterval = syncDataSetEntity.getSyncDownloadTime() * 6 * 100;
long curTime = System.currentTimeMillis();
Long lastTime = (Long) redisUtil.get("SPS_SYNC_DOWNLOAD_DATA");
if (lastTime == null) {
lastTime = System.currentTimeMillis();
redisUtil.set("SPS_SYNC_DOWNLOAD_DATA", lastTime);
}
if (curTime - lastTime > timeInterval) {
syncHeartService.pullData();
}
}

10
src/main/resources/application.yml
Unescape Escape View File

@ -41,3 +41,13 @@ WEB_SUB_TITLE: 医院UDI管理系统
WEB_VERSION:
WEB_LOGO:
#http://127.0.0.1:9991/udiwms/image/getImage?type=image1&name=dima.jpg
# 生产环境推荐配置（8核CPU示例）
thread:
pool:
core-pool-size: 8 # 与CPU核心一致
max-pool-size: 16 # 核心数*2
queue-capacity: 1000 # 根据业务负载调整
keep-alive-seconds: 160 # 大流量场景适当延长

Write Preview
Loading…
Cancel
Save