q792602257/FrameTour-BE
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'render-worker-microservice'

Browse Source 
# Conflicts:
#	src/main/java/com/ycwl/basic/integration/scenic/service/ScenicIntegrationService.java
#	src/main/java/com/ycwl/basic/service/task/impl/TaskTaskServiceImpl.java
This commit is contained in:
Jerry Yan
2025-09-09 11:00:10 +08:00
parent 9ec222a155 aa683a62c4
commit ef61ce9d63
52 changed files with 2711 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/java/com/ycwl/basic/service/mobile/impl/AppScenicServiceImpl.java
View File

@@ -5,7 +5,8 @@ import com.github.pagehelper.PageInfo;
import com.ycwl.basic.integration.common.manager.DeviceConfigManager;
import com.ycwl.basic.integration.scenic.dto.scenic.ScenicV2DTO;
import com.ycwl.basic.integration.device.service.DeviceIntegrationService;
import com.ycwl.basic.integration.device.dto.device.DeviceV2ListResponse;
import com.ycwl.basic.integration.device.dto.device.DeviceV2DTO;
import com.ycwl.basic.integration.common.response.PageResponse;
import com.ycwl.basic.mapper.ExtraDeviceMapper;
import com.ycwl.basic.mapper.ScenicAccountMapper;
import com.ycwl.basic.model.jwt.JwtInfo;
@@ -86,7 +87,7 @@ public class AppScenicServiceImpl implements AppScenicService {
public ApiResponse<ScenicDeviceCountVO> deviceCountByScenicId(Long scenicId) {
JwtInfo worker = JwtTokenUtil.getWorker();
// 通过zt-device服务获取设备统计
DeviceV2ListResponse deviceListResponse = deviceIntegrationService.getScenicActiveDevices(scenicId, 1, 1000);
PageResponse<DeviceV2DTO> deviceListResponse = deviceIntegrationService.getScenicActiveDevices(scenicId, 1, 1000);
ScenicDeviceCountVO scenicDeviceCountVO = new ScenicDeviceCountVO();
if (deviceListResponse != null && deviceListResponse.getList() != null) {
scenicDeviceCountVO.setTotalDeviceCount(deviceListResponse.getList().size());
@@ -120,7 +121,7 @@ public class AppScenicServiceImpl implements AppScenicService {
}
// 通过zt-device服务获取设备统计
DeviceV2ListResponse deviceListResponse = deviceIntegrationService.getScenicActiveDevices(id, 1, 1000);
PageResponse<DeviceV2DTO> deviceListResponse = deviceIntegrationService.getScenicActiveDevices(id, 1, 1000);
ScenicDeviceCountVO scenicDeviceCountVO = new ScenicDeviceCountVO();
if (deviceListResponse != null && deviceListResponse.getList() != null) {
scenicDeviceCountVO.setTotalDeviceCount(deviceListResponse.getList().size());
@@ -276,7 +277,7 @@ public class AppScenicServiceImpl implements AppScenicService {
@Override
public ApiResponse<List<DeviceRespVO>> getDevices(Long scenicId) {
DeviceV2ListResponse deviceV2ListResponse = deviceIntegrationService.listDevices(1, 1000, null, null, null, 1, scenicId);
PageResponse<DeviceV2DTO> deviceV2ListResponse = deviceIntegrationService.listDevices(1, 1000, null, null, null, 1, scenicId);
List<DeviceRespVO> deviceRespVOList = deviceV2ListResponse.getList().stream().map(device -> {
DeviceRespVO deviceRespVO = new DeviceRespVO();
deviceRespVO.setId(device.getId());

22
src/main/java/com/ycwl/basic/service/pc/RenderWorkerService.java
View File

@@ -1,22 +0,0 @@
package com.ycwl.basic.service.pc;
import com.github.pagehelper.PageInfo;
import com.ycwl.basic.model.pc.renderWorker.entity.RenderWorkerEntity;
import com.ycwl.basic.model.pc.renderWorker.req.RenderWorkerReqQuery;
import com.ycwl.basic.utils.ApiResponse;
import java.util.List;
/**
* @Author：longbinbin
* @Date：2024/12/3 15:07
*/
public interface RenderWorkerService {
ApiResponse<PageInfo<RenderWorkerEntity>> pageQuery(RenderWorkerReqQuery renderWorkerReqQuery);
ApiResponse<List<RenderWorkerEntity>> list(RenderWorkerReqQuery renderWorkerReqQuery);
ApiResponse<RenderWorkerEntity> detail(Long id);
ApiResponse<Integer> add(RenderWorkerEntity renderWorker);
ApiResponse<Integer> deleteById(Long id);
ApiResponse<Integer> update(RenderWorkerEntity renderWorker);
ApiResponse<Integer> updateStatus(Long id);
}

139
src/main/java/com/ycwl/basic/service/pc/impl/RenderWorkerServiceImpl.java
View File

@@ -1,139 +0,0 @@
package com.ycwl.basic.service.pc.impl;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.ycwl.basic.mapper.RenderWorkerMapper;
import com.ycwl.basic.model.pc.renderWorker.entity.RenderWorkerEntity;
import com.ycwl.basic.model.pc.renderWorker.req.RenderWorkerReqQuery;
import com.ycwl.basic.model.task.req.ClientStatusReqVo;
import com.ycwl.basic.repository.RenderWorkerRepository;
import com.ycwl.basic.service.pc.RenderWorkerService;
import com.ycwl.basic.utils.ApiResponse;
import com.ycwl.basic.utils.SnowFlakeUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
/**
* @Author：longbinbin
* @Date：2024/12/3 15:09
*/
@Service
public class RenderWorkerServiceImpl implements RenderWorkerService {
@Autowired
private RenderWorkerMapper renderWorkerMapper;
@Autowired
private RenderWorkerRepository renderWorkerRepository;
@Override
public ApiResponse<PageInfo<RenderWorkerEntity>> pageQuery(RenderWorkerReqQuery renderWorkerReqQuery) {
PageHelper.startPage(renderWorkerReqQuery.getPageNum(), renderWorkerReqQuery.getPageSize());
List<RenderWorkerEntity> list = renderWorkerMapper.list(renderWorkerReqQuery);
list.forEach(worker -> {
ClientStatusReqVo clientStatus = renderWorkerRepository.getWorkerHostStatus(worker.getId());
if (clientStatus == null) {
return;
}
worker.setCpuCount(clientStatus.getCpu_count());
worker.setCpuUsage(clientStatus.getCpu_usage());
// 上报的是字节，存储的是兆
worker.setMemoryAvailable(clientStatus.getMemory_available().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setMemoryTotal(clientStatus.getMemory_total().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setPlatform(clientStatus.getPlatform());
worker.setRuntimeVersion(clientStatus.getRuntime_version());
worker.setSupportFeature(String.join(",", clientStatus.getSupport_feature()));
worker.setVersion(clientStatus.getVersion());
worker.setUpdateAt(clientStatus.getUpdateAt());
});
PageInfo<RenderWorkerEntity> pageInfo = new PageInfo<>(list);
return ApiResponse.success(pageInfo);
}
@Override
public ApiResponse<List<RenderWorkerEntity>> list(RenderWorkerReqQuery renderWorkerReqQuery) {
List<RenderWorkerEntity> list = renderWorkerMapper.list(renderWorkerReqQuery);
list.forEach(worker -> {
ClientStatusReqVo clientStatus = renderWorkerRepository.getWorkerHostStatus(worker.getId());
if (clientStatus == null) {
return;
}
worker.setCpuCount(clientStatus.getCpu_count());
worker.setCpuUsage(clientStatus.getCpu_usage());
// 上报的是字节，存储的是兆
worker.setMemoryAvailable(clientStatus.getMemory_available().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setMemoryTotal(clientStatus.getMemory_total().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setPlatform(clientStatus.getPlatform());
worker.setRuntimeVersion(clientStatus.getRuntime_version());
worker.setSupportFeature(String.join(",", clientStatus.getSupport_feature()));
worker.setVersion(clientStatus.getVersion());
worker.setUpdateAt(clientStatus.getUpdateAt());
});
return ApiResponse.success(list);
}
@Override
public ApiResponse<RenderWorkerEntity> detail(Long id) {
RenderWorkerEntity worker = renderWorkerMapper.getById(id);
ClientStatusReqVo clientStatus = renderWorkerRepository.getWorkerHostStatus(worker.getId());
if (clientStatus != null) {
worker.setCpuCount(clientStatus.getCpu_count());
worker.setCpuUsage(clientStatus.getCpu_usage());
// 上报的是字节，存储的是兆
worker.setMemoryAvailable(clientStatus.getMemory_available().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setMemoryTotal(clientStatus.getMemory_total().divide(BigDecimal.valueOf(1024 * 1024), RoundingMode.CEILING));
worker.setPlatform(clientStatus.getPlatform());
worker.setRuntimeVersion(clientStatus.getRuntime_version());
worker.setSupportFeature(String.join(",", clientStatus.getSupport_feature()));
worker.setVersion(clientStatus.getVersion());
worker.setUpdateAt(clientStatus.getUpdateAt());
}
return ApiResponse.success(worker);
}
@Override
public ApiResponse<Integer> add(RenderWorkerEntity renderWorker) {
renderWorker.setId(SnowFlakeUtil.getLongId());
if (StringUtils.isEmpty(renderWorker.getAccessKey())) {
renderWorker.setAccessKey(SnowFlakeUtil.getId());
}
renderWorker.setStatus(0);
int add = renderWorkerMapper.add(renderWorker);
if (add == 0) {
return ApiResponse.fail("渲染机添加失败");
}else {
return ApiResponse.success(add);
}
}
@Override
public ApiResponse<Integer> deleteById(Long id) {
renderWorkerRepository.clearCache(id);
return ApiResponse.success(renderWorkerMapper.deleteById(id));
}
@Override
public ApiResponse<Integer> update(RenderWorkerEntity renderWorker) {
renderWorkerRepository.clearCache(renderWorker.getId());
int update = renderWorkerMapper.update(renderWorker);
if (update == 0) {
return ApiResponse.fail("渲染机修改失败");
}else {
return ApiResponse.success(update);
}
}
@Override
public ApiResponse<Integer> updateStatus(Long id) {
renderWorkerRepository.clearCache(id);
return ApiResponse.success(renderWorkerMapper.updateStatus(id));
}
}

53
src/main/java/com/ycwl/basic/service/task/impl/TaskTaskServiceImpl.java
View File

@@ -2,7 +2,8 @@ package com.ycwl.basic.service.task.impl;
import cn.hutool.core.date.DateUtil;
import cn.hutool.crypto.digest.MD5;
import com.ycwl.basic.integration.common.manager.DeviceConfigManager;
import com.ycwl.basic.integration.common.manager.RenderWorkerConfigManager;
import com.ycwl.basic.integration.common.manager.ScenicConfigManager;
import com.ycwl.basic.utils.JacksonUtil;
import com.ycwl.basic.biz.OrderBiz;
import com.ycwl.basic.biz.TaskStatusBiz;
@@ -12,7 +13,6 @@ import com.ycwl.basic.constant.TaskConstant;
import com.ycwl.basic.mapper.FaceMapper;
import com.ycwl.basic.mapper.FaceSampleMapper;
import com.ycwl.basic.mapper.MemberMapper;
import com.ycwl.basic.mapper.RenderWorkerMapper;
import com.ycwl.basic.mapper.SourceMapper;
import com.ycwl.basic.mapper.TaskMapper;
import com.ycwl.basic.mapper.TemplateMapper;
@@ -83,6 +83,8 @@ import java.util.stream.Collectors;
@Slf4j
@Service
public class TaskTaskServiceImpl implements TaskService {
private static final String WORKER_SELF_HOSTED_CACHE_KEY = "worker_self_hosted_scenic:%s";
private static final int CACHE_EXPIRE_MINUTES = 3;
@Autowired
private TaskMapper taskMapper;
@Autowired
@@ -121,13 +123,40 @@ public class TaskTaskServiceImpl implements TaskService {
private DeviceRepository deviceRepository;
@Autowired
private VideoReUploader videoReUploader;
@Autowired
private RedisTemplate<String, String> redisTemplate;
private RenderWorkerEntity getWorker(@NonNull WorkerAuthReqVo req) {
String accessKey = req.getAccessKey();
if (accessKey == null) {
return null;
}
return repository.getWorkerByAccessKey(accessKey);
RenderWorkerEntity worker = repository.getWorkerByAccessKey(accessKey);
if (worker == null) {
return null;
}
if (worker.getStatus() != 1) {
return null;
}
return worker;
}
private boolean isWorkerSelfHostedScenic(Long scenicId) {
String cacheKey = String.format(WORKER_SELF_HOSTED_CACHE_KEY, scenicId);
String cachedValue = redisTemplate.opsForValue().get(cacheKey);
if (cachedValue != null) {
return Boolean.parseBoolean(cachedValue);
}
// 缓存中没有，查询配置
ScenicConfigManager config = scenicRepository.getScenicConfigManager(scenicId);
boolean workerSelfHostedScenic = Boolean.TRUE.equals(config.getBoolean("worker_self_hosted"));
// 缓存结果，设置30分钟过期
redisTemplate.opsForValue().set(cacheKey, String.valueOf(workerSelfHostedScenic), CACHE_EXPIRE_MINUTES, TimeUnit.MINUTES);
return workerSelfHostedScenic;
}
@Override
@@ -136,7 +165,6 @@ public class TaskTaskServiceImpl implements TaskService {
if (worker == null) {
return null;
}
worker.setOnline(1);
worker.setName(null);
worker.setStatus(null);
// get status
@@ -162,21 +190,25 @@ public class TaskTaskServiceImpl implements TaskService {
} else {
updTemplateList = templateRepository.getAllEnabledTemplateList();
}
RenderWorkerConfigManager configManager = repository.getWorkerConfigManager(worker.getId());
try {
if (lock.tryLock(2, TimeUnit.SECONDS)) {
try {
List<TaskRespVO> taskList;
if (worker.getScenicOnly() != null) {
taskList = taskMapper.selectNotRunningByScenicList(worker.getScenicOnly());
if (Strings.isNotBlank(configManager.getString("scenic_only"))) {
taskList = taskMapper.selectNotRunningByScenicList(configManager.getString("scenic_only"));
} else {
taskList = taskMapper.selectNotRunning();
var _taskList = taskMapper.selectNotRunning();
taskList = _taskList.stream().filter(task -> {
boolean workerSelfHostedScenic = isWorkerSelfHostedScenic(task.getScenicId());
return !workerSelfHostedScenic;
}).limit(1).toList();
}
resp.setTasks(taskList);
resp.setTemplates(updTemplateList);
taskList.forEach(task -> {
taskMapper.assignToWorker(task.getId(), worker.getId());
videoTaskRepository.clearTaskCache(task.getId());
repository.clearCache(worker.getId());
});
} finally {
lock.unlock();
@@ -551,10 +583,11 @@ public class TaskTaskServiceImpl implements TaskService {
if (worker == null) {
return null;
}
RenderWorkerConfigManager config = repository.getWorkerConfigManager(worker.getId());
IStorageAdapter adapter;
try {
adapter = StorageFactory.get(worker.getStoreType());
adapter.loadConfig(JacksonUtil.parseObject(worker.getStoreConfigJson(), Map.class));
adapter = StorageFactory.get(config.getString("store_type"));
adapter.loadConfig(config.getObject("store_config_json", Map.class));
} catch (Exception e) {
adapter = scenicService.getScenicStorageAdapter(task.getScenicId());
}