| | |
| | | package com.linghu.controller; |
| | | |
| | | import java.time.Duration; |
| | | import java.time.LocalDateTime; |
| | | import java.util.ArrayList; |
| | | import java.util.Date; |
| | | import java.util.List; |
| | | import java.util.Map; |
| | | import java.time.format.DateTimeFormatter; |
| | | import java.util.*; |
| | | import java.util.concurrent.BlockingQueue; |
| | | import java.util.concurrent.LinkedBlockingQueue; |
| | | import java.util.concurrent.locks.ReentrantLock; |
| | | import java.util.stream.Collectors; |
| | | |
| | | import javax.servlet.http.HttpServletRequest; |
| | | |
| | | import com.fasterxml.jackson.core.JsonProcessingException; |
| | | import com.linghu.model.dto.*; |
| | | import com.linghu.model.entity.*; |
| | | import com.linghu.service.*; |
| | | import lombok.extern.slf4j.Slf4j; |
| | | import org.springframework.beans.factory.annotation.Autowired; |
| | | import org.springframework.beans.factory.annotation.Value; |
| | | import org.springframework.core.ParameterizedTypeReference; |
| | | import org.springframework.dao.DuplicateKeyException; |
| | | import org.springframework.http.*; |
| | | import org.springframework.web.client.RestTemplate; |
| | | import org.springframework.web.reactive.function.client.WebClient; |
| | | import org.springframework.http.client.HttpComponentsClientHttpRequestFactory; |
| | | |
| | | import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; |
| | | import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; |
| | | import com.linghu.model.common.ResponseResult; |
| | | import com.linghu.model.dto.SearchTaskRequest; |
| | | import com.linghu.model.entity.Keyword; |
| | | import com.linghu.model.entity.Question; |
| | | import com.linghu.model.entity.User; |
| | | import com.linghu.service.KeywordService; |
| | | import com.linghu.service.QuestionService; |
| | | import com.linghu.service.ReferenceService; |
| | | import com.linghu.utils.JwtUtils; |
| | | import com.linghu.model.dto.SearchTaskResponse; |
| | | import com.linghu.model.dto.TaskStatusResponse; |
| | | import com.linghu.model.dto.TaskCancelResponse; |
| | | |
| | | import io.jsonwebtoken.lang.Collections; |
| | | import io.swagger.annotations.Api; |
| | | import io.swagger.annotations.ApiOperation; |
| | | import reactor.core.publisher.Flux; |
| | | import reactor.core.publisher.Mono; |
| | | |
| | | import org.springframework.web.bind.annotation.*; |
| | | import org.springframework.web.bind.annotation.* ; |
| | | import org.springframework.http.HttpStatus; |
| | | import com.linghu.model.dto.TaskResultResponse; |
| | | import com.linghu.model.dto.TaskResultResponse.QuestionResult; |
| | | import com.linghu.model.dto.TaskResultResponse.UserResult; |
| | | import com.linghu.model.entity.Reference; |
| | | import java.util.stream.Collectors; |
| | | import reactor.core.scheduler.Schedulers; |
| | | |
| | | @RestController |
| | | @RequestMapping("/collect") |
| | | @Api(value = "采集接口", tags = "采集管理") |
| | | @Slf4j |
| | | public class CollectController { |
| | | |
| | | @Autowired |
| | |
| | | private KeywordService keywordService; |
| | | @Autowired |
| | | private QuestionService questionService; |
| | | @Autowired |
| | | private KeywordTaskService keywordTaskService; |
| | | @Autowired |
| | | private PlatformService platformService; |
| | | @Autowired |
| | | private TypeService typeService; |
| | | @Autowired |
| | | private UserService userService; |
| | | @Autowired |
| | | private OrderService orderService; |
| | | // 1. 使用线程安全的队列实现 |
| | | private final BlockingQueue<SearchTaskRequest> taskQueue = new LinkedBlockingQueue<>(); |
| | | // 添加队列访问锁 |
| | | private final ReentrantLock queueLock = new ReentrantLock(); |
| | | private static boolean isProcessing = false; |
| | | |
| | | @PostMapping("/search") |
| | | @ApiOperation(value = "开始采集") |
| | | public Mono<ResponseResult<SearchTaskResponse>> createSearchTask( |
| | | public Mono<ResponseResult<?>> createSearchTask( |
| | | @RequestBody SearchTaskRequest searchTaskRequest, |
| | | HttpServletRequest request) { |
| | | String token = request.getHeader("Authorization"); |
| | | User user = jwtUtils.parseToken(token); |
| | | List<User> users = new ArrayList<>(); |
| | | users.add(user); |
| | | searchTaskRequest.setUsers(users); |
| | | HttpServletRequest request) throws JsonProcessingException { |
| | | |
| | | return webClient.post() |
| | | .uri(baseUrl + "/search") |
| | | .contentType(MediaType.APPLICATION_JSON) |
| | | .bodyValue(searchTaskRequest) |
| | | .retrieve() |
| | | .bodyToMono(new ParameterizedTypeReference<ResponseResult<SearchTaskResponse>>() { |
| | | }) |
| | | .flatMap(responseResult -> { |
| | | // 提取任务ID |
| | | SearchTaskResponse taskResponse = responseResult.getData(); |
| | | if (taskResponse != null && taskResponse.getTask_id() != null) { |
| | | // 保存任务ID到关键词 |
| | | LambdaUpdateWrapper<Keyword> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | updateWrapper.eq(Keyword::getKeyword_id, searchTaskRequest.getKeyword_id()); |
| | | updateWrapper.set(Keyword::getTask_id, taskResponse.getTask_id()); |
| | | keywordService.update(updateWrapper); |
| | | // 可选:更新响应中的其他信息 |
| | | // taskResponse.setMessage("任务已提交并保存,ID: " + taskResponse.getTaskId()); |
| | | // 首先检查服务器资源 |
| | | return getServerResource() |
| | | .flatMap(resourceResponse -> { |
| | | double cpuUsage = parseUsage(resourceResponse.getCpu_usage_percent()); |
| | | double memoryUsage = parseUsage(resourceResponse.getMemory_usage_percent()); |
| | | |
| | | if (cpuUsage >= 90.0 || memoryUsage >= 90.0) { |
| | | |
| | | |
| | | String errorMsg = String.format("服务器资源不足,请稍后再试"); |
| | | |
| | | log.warn(errorMsg); |
| | | return Mono.just(ResponseResult.error(503, errorMsg)); |
| | | } |
| | | return Mono.just(responseResult); |
| | | |
| | | // 将新的任务请求加入队列 |
| | | // taskQueue.add(searchTaskRequest); |
| | | queueLock.lock(); |
| | | try { |
| | | taskQueue.add(searchTaskRequest); |
| | | } finally { |
| | | queueLock.unlock(); |
| | | } |
| | | |
| | | // 如果当前没有任务在处理中,则启动任务队列的处理 |
| | | if (!isProcessing) { |
| | | processNextTaskInQueue(); |
| | | } |
| | | |
| | | // 返回响应,通知用户任务已开始 |
| | | return Mono.just(ResponseResult.success("任务已加入队列,正在处理...")); |
| | | }) |
| | | .onErrorResume(e -> { |
| | | return Mono.just(ResponseResult.error("调用失败: " + e.getMessage())); |
| | | log.error("检查服务器资源失败: {}", e.getMessage(), e); |
| | | return Mono.just(ResponseResult.error("检查服务器资源失败: " + e.getMessage())); |
| | | }); |
| | | } |
| | | |
| | | private void processNextTaskInQueue() { |
| | | // 设置为正在处理 |
| | | isProcessing = true; |
| | | |
| | | // 从队列中取出下一个任务 |
| | | SearchTaskRequest nextTaskRequest = taskQueue.poll(); |
| | | |
| | | if (nextTaskRequest != null) { |
| | | // 处理任务 |
| | | executeBatchTask(nextTaskRequest) |
| | | .doFinally(signal -> { |
| | | // 完成后,继续处理下一个任务 |
| | | isProcessing = false; |
| | | if (!taskQueue.isEmpty()) { |
| | | processNextTaskInQueue(); // 继续处理队列中的下一个任务 |
| | | } |
| | | }) |
| | | .subscribe(); |
| | | } |
| | | } |
| | | |
| | | private Mono<ResponseResult<String>> executeBatchTask(SearchTaskRequest searchTaskRequest) { |
| | | log.info("开始处理任务:{}", searchTaskRequest); |
| | | log.info("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++"); |
| | | Integer keywordId = searchTaskRequest.getKeyword_id(); |
| | | int maxConcurrentUsers = searchTaskRequest.getConfig() != null ? |
| | | searchTaskRequest.getConfig().getMax_concurrent_users() : 3; |
| | | List<List<UserDto>> userBatches = splitUsersIntoBatches(searchTaskRequest.getUsers(), maxConcurrentUsers, keywordId); |
| | | |
| | | Queue<List<UserDto>> batchQueue = new LinkedList<>(userBatches); // 用队列存储批次 |
| | | |
| | | return Mono.just(ResponseResult.success("第一个批次已开始")) |
| | | .doOnTerminate(() -> { |
| | | // 启动后台任务,继续处理批次 |
| | | executeBatchTask(batchQueue, searchTaskRequest, keywordId) |
| | | .subscribe(); // 使用subscribe()启动后台任务 |
| | | }); |
| | | } |
| | | |
| | | private Mono<ResponseResult<?>> executeBatchTask(Queue<List<UserDto>> batchQueue, SearchTaskRequest searchTaskRequest, Integer keywordId) { |
| | | // 如果队列为空,说明所有批次已经完成 |
| | | if (batchQueue.isEmpty()) { |
| | | return Mono.just(ResponseResult.success("所有批次已完成")); |
| | | } |
| | | |
| | | List<UserDto> currentBatch = batchQueue.poll(); // 从队列中获取当前批次 |
| | | SearchTaskRequest batchRequest = new SearchTaskRequest(); |
| | | batchRequest.setUsers(currentBatch); |
| | | batchRequest.setQuestions(searchTaskRequest.getQuestions()); |
| | | batchRequest.setConfig(searchTaskRequest.getConfig()); |
| | | batchRequest.setSave_to_database(searchTaskRequest.getSave_to_database()); |
| | | batchRequest.setWebhook_url(searchTaskRequest.getWebhook_url()); |
| | | batchRequest.setKeyword_id(keywordId); |
| | | |
| | | return createSingleBatchTask(batchRequest) |
| | | .flatMap(taskResponse -> { |
| | | // if (taskResponse != null && taskResponse.getTask_id() != null) { |
| | | // // 保存任务关联到数据库 |
| | | // return saveKeywordTasks(keywordId, taskResponse) |
| | | // .then(waitForTaskCompletion(taskResponse.getTask_id(), batchQueue, searchTaskRequest, keywordId)); |
| | | // } else { |
| | | // return Mono.just(ResponseResult.error("创建批次任务失败")); |
| | | // } |
| | | if (taskResponse != null && taskResponse.getTask_id() != null) { |
| | | // 直接等待任务完成,不再保存任务关联信息 |
| | | return waitForTaskCompletion(taskResponse.getTask_id(), batchQueue, searchTaskRequest, keywordId); |
| | | } else { |
| | | return Mono.just(ResponseResult.error("创建批次任务失败")); |
| | | } |
| | | }); |
| | | } |
| | | |
| | | // private Mono<Void> saveKeywordTasks(Integer keywordId, SearchTaskResponse taskResponse) { |
| | | // if (taskResponse == null || taskResponse.getTask_id() == null) { |
| | | // return Mono.error(new RuntimeException("任务响应无效或任务ID为空")); |
| | | // } |
| | | // |
| | | // KeywordTask keywordTask = new KeywordTask(); |
| | | // keywordTask.setKeyword_id(keywordId); |
| | | // keywordTask.setTask_id(taskResponse.getTask_id()); |
| | | // keywordTask.setStatus("pending"); |
| | | // |
| | | // // 将 MyBatis-Plus 的同步方法包装为 Mono<Void> |
| | | // return Mono.fromRunnable(() -> { |
| | | // boolean success = keywordTaskService.saveOrUpdate(keywordTask); |
| | | // if (!success) { |
| | | // throw new RuntimeException("保存关键词任务关联失败"); |
| | | // } |
| | | // }) |
| | | // .doFinally(signalType -> log.info("成功保存关键词任务关联: Task ID {}", taskResponse.getTask_id())) |
| | | // .then(); |
| | | // } |
| | | private Mono<ResponseResult<?>> waitForTaskCompletion(String taskId, Queue<List<UserDto>> batchQueue, SearchTaskRequest searchTaskRequest, Integer keywordId) { |
| | | // 查询任务状态 |
| | | return getTaskStatus(taskId) |
| | | .flatMap(statusResponse -> { |
| | | // 如果任务状态是"submitted"或"running",继续轮询 |
| | | if (!"completed".equalsIgnoreCase(statusResponse.getStatus()) && !"failed".equalsIgnoreCase(statusResponse.getStatus()) && !"cancelled".equalsIgnoreCase(statusResponse.getStatus()) ) { |
| | | return Mono.delay(Duration.ofSeconds(5)) // 延迟 5 秒后再次查询 |
| | | .flatMap(aLong -> waitForTaskCompletion(taskId, batchQueue, searchTaskRequest, keywordId)); // 递归调用继续等待 |
| | | } else { |
| | | // 如果状态为其他状态,则继续处理下一个批次 |
| | | return executeBatchTask(batchQueue, searchTaskRequest, keywordId); |
| | | } |
| | | }) |
| | | .onErrorResume(e -> { |
| | | // 处理查询任务状态时的错误 |
| | | return Mono.just(ResponseResult.error("查询任务状态失败: " + e.getMessage())); |
| | | }); |
| | | } |
| | | @ApiOperation(value = "查询任务状态") |
| | | @GetMapping("/status") |
| | | public Mono<TaskStatusResponse> getTaskStatus(String taskId) { |
| | | return webClient.get() |
| | | .uri(baseUrl + "/tasks/" + taskId) |
| | | .uri(baseUrl + "/api/v1/tasks/" + taskId) |
| | | .accept(MediaType.APPLICATION_JSON) |
| | | .retrieve() |
| | | .onStatus(HttpStatus::is4xxClientError, response -> response.bodyToMono(String.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException("任务不存在: " + errorBody)))) |
| | | .onStatus(HttpStatus::isError, response -> response.bodyToMono(TaskStatusResponse.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException(errorBody.getDetail())))) |
| | | .bodyToMono(TaskStatusResponse.class) |
| | | .flatMap(result -> { |
| | | TaskStatusResponse taskStatusResponse = result; |
| | | if (taskStatusResponse != null && taskStatusResponse.getStatus() != null) { |
| | | List<Question> updateQuestions = taskStatusResponse.getQuestions_status().stream() |
| | | .map(qs -> { |
| | | Question question = new Question(); |
| | | question.setQuestion_id(qs.getQuestion_id()); |
| | | question.setStatus(qs.getStatus()); |
| | | return question; |
| | | }).collect(Collectors.toList()); |
| | | |
| | | questionService.updateBatchById(updateQuestions); |
| | | } |
| | | return Mono.just(result); |
| | | .onErrorResume(e -> { |
| | | // 处理错误,创建一个自定义的错误响应对象 |
| | | TaskStatusResponse errorResponse = new TaskStatusResponse(); |
| | | errorResponse.setStatus("ERROR"); |
| | | errorResponse.setMessage(e.getMessage()); |
| | | errorResponse.setDetail(e.getMessage()); |
| | | return Mono.just(errorResponse); |
| | | }); |
| | | } |
| | | |
| | | @PostMapping("/cancel/{taskId}") |
| | | |
| | | |
| | | // 添加一个辅助方法来安全地将字符串转换为double |
| | | private double parseUsage(String usageStr) { |
| | | try { |
| | | if (usageStr != null) { |
| | | // 移除可能存在的百分号 |
| | | usageStr = usageStr.replace("%", "").trim(); |
| | | return Double.parseDouble(usageStr); |
| | | } |
| | | return 0.0; |
| | | } catch (NumberFormatException e) { |
| | | log.error("解析资源使用率失败: {}", e.getMessage()); |
| | | return 0.0; |
| | | } |
| | | } |
| | | /*@PostMapping("/search") |
| | | @ApiOperation(value = "开始采集") |
| | | public Mono<ResponseResult<?>> createSearchTask( |
| | | @RequestBody SearchTaskRequest searchTaskRequest, |
| | | HttpServletRequest request) throws JsonProcessingException { |
| | | |
| | | // 首先检查服务器资源 |
| | | return getServerResource() |
| | | .flatMap(resourceResponse -> { |
| | | // 将字符串类型的使用率转换为double类型 |
| | | double cpuUsage = parseUsage(resourceResponse.getCpu_usage_percent()); |
| | | double memoryUsage = parseUsage(resourceResponse.getMemory_usage_percent()); |
| | | // 检查CPU和内存使用率 |
| | | if (cpuUsage >= 90.0 || memoryUsage >= 90.0) { |
| | | String errorMsg = String.format("服务器资源不足:CPU使用率 %.1f%%,内存使用率 %.1f%%", |
| | | resourceResponse.getCpu_usage_percent(), resourceResponse.getMemory_usage_percent()); |
| | | log.warn(errorMsg); |
| | | return Mono.just(ResponseResult.error(503, errorMsg)); |
| | | } |
| | | Integer keywordId = searchTaskRequest.getKeyword_id(); |
| | | |
| | | int maxConcurrentUsers = searchTaskRequest.getConfig() != null ? |
| | | searchTaskRequest.getConfig().getMax_concurrent_users() : 3; |
| | | List<List<UserDto>> userBatches = splitUsersIntoBatches(searchTaskRequest.getUsers(), maxConcurrentUsers,keywordId); |
| | | |
| | | return Flux.fromIterable(userBatches) |
| | | .flatMap(batch -> { |
| | | SearchTaskRequest batchRequest = new SearchTaskRequest(); |
| | | batchRequest.setUsers(batch); |
| | | batchRequest.setQuestions(searchTaskRequest.getQuestions()); |
| | | batchRequest.setConfig(searchTaskRequest.getConfig()); |
| | | batchRequest.setSave_to_database(searchTaskRequest.getSave_to_database()); |
| | | batchRequest.setWebhook_url(searchTaskRequest.getWebhook_url()); |
| | | batchRequest.setKeyword_id(keywordId); |
| | | |
| | | return createSingleBatchTask(batchRequest) |
| | | .delaySubscription(Duration.ofSeconds(2)); // 批次之间添加延迟 |
| | | }, 1) // 限制并发数为1,确保顺序执行 |
| | | .collectList() // 收集所有批次的响应 |
| | | .flatMap(responses -> |
| | | saveKeywordTasks(keywordId, responses) // 保存关联关系 |
| | | .thenReturn(responses) // 返回原始响应 |
| | | ) |
| | | .map(responses -> ResponseResult.success(responses)) // 使用ResponseResult包装结果 |
| | | .onErrorResume(e -> { |
| | | log.error("创建搜索任务失败: {}", e.getMessage(), e); |
| | | return Mono.just(ResponseResult.error("创建搜索任务失败: " + e.getMessage())); |
| | | }); |
| | | }) |
| | | .onErrorResume(e -> { |
| | | log.error("检查服务器资源失败: {}", e.getMessage(), e); |
| | | return Mono.just(ResponseResult.error("检查服务器资源失败: " + e.getMessage())); |
| | | }); |
| | | }*/ |
| | | |
| | | // private Mono<Void> saveKeywordTasks(Integer keywordId, List<SearchTaskResponse> taskResponses) { |
| | | // List<KeywordTask> keywordTasks = taskResponses.stream() |
| | | // .filter(response -> response.getTask_id() != null) |
| | | // .map(response -> { |
| | | // KeywordTask keywordTask = new KeywordTask(); |
| | | // keywordTask.setKeyword_id(keywordId); |
| | | // keywordTask.setTask_id(response.getTask_id()); |
| | | // keywordTask.setStatus("pending"); |
| | | // return keywordTask; |
| | | // }) |
| | | // .collect(Collectors.toList()); |
| | | // |
| | | // // 将 MyBatis-Plus 的同步方法包装为 Mono<Void> |
| | | // return Mono.fromRunnable(() -> { |
| | | // boolean success = keywordTaskService.saveOrUpdateBatch(keywordTasks); |
| | | // if (!success) { |
| | | //// throw new RuntimeException("保存关键词任务关联失败"); |
| | | // // 添加异常处理 |
| | | // Mono.error( new RuntimeException("保存关键词任务关联失败")); |
| | | // } |
| | | // }) |
| | | // .doFinally(signalType -> log.info("成功保存 {} 个关键词任务关联", keywordTasks.size())) |
| | | // .then(); |
| | | // } |
| | | |
| | | private List<List<UserDto>> splitUsersIntoBatches(List<UserDto> users, int batchSize,Integer keywordId) { |
| | | |
| | | Keyword keyword = keywordService.getById(keywordId); |
| | | if (null==keyword.getNum()){ |
| | | keyword.setNum(0); |
| | | } |
| | | keyword.setNum(keyword.getNum()+1); |
| | | keywordService.updateById(keyword); |
| | | |
| | | List<List<UserDto>> batches = new ArrayList<>(); |
| | | for (int i = 0; i < users.size(); i += batchSize) { |
| | | batches.add(users.subList(i, Math.min(i + batchSize, users.size()))); |
| | | |
| | | } |
| | | for (int i = 0; i < batches.size(); i++){ |
| | | // 创建 KeywordTask 关联,task_id 设置为 null,表示任务尚未开始 |
| | | KeywordTask keywordTask = new KeywordTask(); |
| | | keywordTask.setKeyword_id(keywordId); |
| | | keywordTask.setTask_id(null); // 任务ID为空 |
| | | |
| | | keywordTask.setNum(keyword.getNum());// 任务初始状态为 pending |
| | | keywordTaskService.save(keywordTask); // 保存 KeywordTask |
| | | } |
| | | |
| | | |
| | | return batches; |
| | | } |
| | | |
| | | private Mono<SearchTaskResponse> createSingleBatchTask(SearchTaskRequest batchRequest) { |
| | | return webClient.post() |
| | | .uri(baseUrl + "/api/v1/search") |
| | | .contentType(MediaType.APPLICATION_JSON) |
| | | .bodyValue(batchRequest) |
| | | .retrieve() |
| | | .onStatus(HttpStatus::is4xxClientError, response -> response.bodyToMono(String.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException(errorBody)))) |
| | | .bodyToMono(new ParameterizedTypeReference<SearchTaskResponse>() {}) |
| | | .flatMap(taskResponse -> { |
| | | if (taskResponse != null && taskResponse.getTask_id() != null) { |
| | | |
| | | // 使用 Reactor 的方式更新数据库 |
| | | return Mono.fromRunnable(() -> { |
| | | |
| | | //更新关键词状态 |
| | | LambdaUpdateWrapper<Keyword> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | updateWrapper.eq(Keyword::getKeyword_id, batchRequest.getKeyword_id()); |
| | | updateWrapper.set(Keyword::getStatus, "submitted"); |
| | | updateWrapper.set(Keyword::getTask_id, taskResponse.getTask_id()); |
| | | keywordService.update(updateWrapper); |
| | | //设置轮数 |
| | | Keyword keyword = keywordService.getById(batchRequest.getKeyword_id()); |
| | | // 更新关键词任务与任务ID的关联 |
| | | // 获取与关键词相关的任务,task_id 为 null,确保只取一个任务 |
| | | List<KeywordTask> keywordTasks = keywordTaskService.list(new LambdaQueryWrapper<KeywordTask>() |
| | | .eq(KeywordTask::getKeyword_id, keyword.getKeyword_id()) |
| | | .isNull(KeywordTask::getTask_id)); |
| | | if (keywordTasks.size() > 0) { |
| | | KeywordTask keywordTask = keywordTasks.get(0); |
| | | keywordTask.setTask_id(taskResponse.getTask_id()); |
| | | keywordTask.setStatus("pending"); |
| | | keywordTaskService.updateById(keywordTask); |
| | | } |
| | | // KeywordTask taskToUpdate = keywordTaskService.getOne(new LambdaQueryWrapper<KeywordTask>() |
| | | // .eq(KeywordTask::getKeyword_id, keyword.getKeyword_id()) |
| | | // .eq(KeywordTask::getTask_id, null)); // 确保 task_id 为 null 的任务 |
| | | |
| | | // 更新任务的 task_id 和状态 |
| | | // taskToUpdate.setTask_id(taskResponse.getTask_id()); |
| | | // taskToUpdate.setStatus("submitted"); |
| | | // keywordTaskService.updateById(taskToUpdate); |
| | | |
| | | |
| | | //设置订单进入采集状态 |
| | | LambdaUpdateWrapper<Orders> updateOrderWrapper = new LambdaUpdateWrapper<>(); |
| | | updateOrderWrapper.eq(Orders::getOrder_id, keyword.getOrder_id()) // 确保字段名正确 |
| | | .set(Orders::getStatus, 2); // 直接设置状态值 |
| | | |
| | | boolean success = orderService.update(updateOrderWrapper); |
| | | log.info("订单状态更新结果: {}", success ? "成功" : "失败"); |
| | | |
| | | }).subscribeOn(Schedulers.boundedElastic()) // 在弹性线程池执行 |
| | | .thenReturn(taskResponse); |
| | | } |
| | | return Mono.just(taskResponse); |
| | | }); |
| | | } |
| | | |
| | | // 移除原来的waitForTaskCompletion方法,不再需要同步等待 |
| | | // @ApiOperation(value = "查询任务状态") |
| | | // @GetMapping("/status") |
| | | // public Mono<TaskStatusResponse> getTaskStatus(String taskId) { |
| | | // return webClient.get() |
| | | // .uri(baseUrl + "/api/v1/tasks/" + taskId) |
| | | // .accept(MediaType.APPLICATION_JSON) |
| | | // .retrieve() |
| | | // .onStatus(HttpStatus::isError, response -> response.bodyToMono(TaskStatusResponse.class) |
| | | // .flatMap(errorBody -> Mono.error(new RuntimeException(errorBody.getDetail())))) |
| | | // .bodyToMono(TaskStatusResponse.class) |
| | | // .flatMap(result -> { |
| | | // TaskStatusResponse taskStatusResponse = result; |
| | | // if (taskStatusResponse != null && taskStatusResponse.getStatus() != null) { |
| | | // List<Question> updateQuestions = taskStatusResponse.getQuestions_status().stream() |
| | | // .map(qs -> { |
| | | // Question question = new Question(); |
| | | // question.setQuestion_id(qs.getQuestion_id()); |
| | | // question.setStatus(qs.getStatus()); |
| | | // return question; |
| | | // }).collect(Collectors.toList()); |
| | | // |
| | | // // 包装成响应式操作 |
| | | // return Mono.fromCallable(() -> { |
| | | // questionService.updateBatchById(updateQuestions); |
| | | // return result; |
| | | // }); |
| | | // |
| | | // } |
| | | // return Mono.just(result); |
| | | // }) |
| | | // .onErrorResume(e -> { |
| | | // // 创建一个自定义的错误响应对象 |
| | | // TaskStatusResponse errorResponse = new TaskStatusResponse(); |
| | | // errorResponse.setStatus("ERROR"); |
| | | // errorResponse.setMessage(e.getMessage()); |
| | | // errorResponse.setDetail(e.getMessage()); |
| | | // |
| | | // return Mono.just(errorResponse); |
| | | // }); |
| | | // } |
| | | @PostMapping("/cancel/{keywordId}") |
| | | @ApiOperation(value = "取消任务") |
| | | public Mono<ResponseResult<TaskCancelResponse>> cancelTask(@PathVariable Integer keywordId) { |
| | | // 1. 查询所有与关键词相关的任务 |
| | | List<KeywordTask> tasks = keywordTaskService.list( |
| | | new LambdaQueryWrapper<KeywordTask>().eq(KeywordTask::getKeyword_id, keywordId) |
| | | ); |
| | | |
| | | // 2. 从队列中移除所有相关任务(使用锁保证线程安全) |
| | | List<SearchTaskRequest> removedQueueTasks = removeTasksFromQueueByKeywordId(keywordId); |
| | | |
| | | // 3. 筛选出需要远程取消的任务 |
| | | List<KeywordTask> tasksToCancelRemotely = tasks.stream() |
| | | .filter(task -> task.getTask_id() != null && "pending".equalsIgnoreCase(task.getStatus())) |
| | | .collect(Collectors.toList()); |
| | | |
| | | // 4. 对筛选出的任务发送远程取消请求(并行执行) |
| | | return Flux.fromIterable(tasksToCancelRemotely) |
| | | .flatMap(task -> { |
| | | // 创建状态更新和远程取消的组合操作 |
| | | Mono<Void> updateStatus = updateTaskStatus(task.getTask_id(), "canceled"); |
| | | Mono<ResponseResult<?>> cancelOp = cancelRemoteTask(task.getTask_id()) |
| | | .onErrorResume(e -> { |
| | | log.error("取消任务 {} 失败: {}", task.getTask_id(), e.getMessage()); |
| | | return Mono.just(ResponseResult.error("取消任务失败: " + e.getMessage())); |
| | | }); |
| | | |
| | | // 合并操作:无论远程取消是否成功,都更新状态 |
| | | return Mono.zip(cancelOp, updateStatus) |
| | | .thenReturn(true); |
| | | }, 10) // 设置10的并发度 |
| | | .collectList() |
| | | .thenReturn(ResponseResult.success( |
| | | new TaskCancelResponse( |
| | | String.format("关键词任务已取消,队列中移除 %d 个任务,远程取消 %d 个任务", |
| | | removedQueueTasks.size(), |
| | | tasksToCancelRemotely.size()) |
| | | ) |
| | | )) |
| | | .onErrorResume(e -> { |
| | | log.error("取消关键词任务失败: {}", e.getMessage()); |
| | | return Mono.just(ResponseResult.error(500, "取消关键词任务失败: " + e.getMessage())); |
| | | }); |
| | | } |
| | | |
| | | // @PostMapping("/cancel/{keywordId}") |
| | | // @ApiOperation(value = "取消任务") |
| | | // public Mono<ResponseResult<TaskCancelResponse>> cancelTask(@PathVariable Integer keywordId) { |
| | | // // 1. 查询所有与关键词相关的任务 |
| | | // List<KeywordTask> tasks = keywordTaskService.list( |
| | | // new LambdaQueryWrapper<KeywordTask>().eq(KeywordTask::getKeyword_id, keywordId) |
| | | // ); |
| | | // |
| | | // // 2. 从队列中移除所有相关任务 |
| | | // List<SearchTaskRequest> removedQueueTasks = removeTasksFromQueueByKeywordId(keywordId); |
| | | // |
| | | // // 3. 筛选出需要远程取消的任务(任务ID不为空且状态为pending) |
| | | // List<KeywordTask> tasksToCancelRemotely = tasks.stream() |
| | | // .filter(task -> task.getTask_id() != null && "pending".equalsIgnoreCase(task.getStatus())) |
| | | // .collect(Collectors.toList()); |
| | | // // 检查是否有任务与关键词相关 |
| | | //// if (tasks.isEmpty()) { |
| | | //// return Mono.just(ResponseResult.error("没有找到相关任务")); |
| | | //// } |
| | | // |
| | | // // 4. 对筛选出的任务发送远程取消请求 |
| | | // List<Mono<ResponseResult<?>>> cancelRequests = tasksToCancelRemotely.stream() |
| | | // .map(task -> cancelRemoteTask(task.getTask_id()) |
| | | // .doOnSuccess(response -> { |
| | | // // 更新任务状态为canceled |
| | | // updateTaskStatus(task.getTask_id(), "canceled").subscribe(); |
| | | // }) |
| | | // .onErrorResume(e -> { |
| | | // log.error("取消任务 {} 失败: {}", task.getTask_id(), e.getMessage()); |
| | | // // 即使取消失败,也尝试更新状态 |
| | | // updateTaskStatus(task.getTask_id(), "canceled").subscribe(); |
| | | // return Mono.just(ResponseResult.error("取消任务失败: " + e.getMessage())); |
| | | // })) |
| | | // .collect(Collectors.toList()); |
| | | // |
| | | // // 5. 并行执行所有取消请求 |
| | | // return Flux.fromIterable(cancelRequests) |
| | | // .concatMap(request -> request) // 顺序执行,而非并行 |
| | | // .collectList() |
| | | // .thenReturn(ResponseResult.success( |
| | | // new TaskCancelResponse( |
| | | // String.format("关键词任务已取消,队列中移除 %d 个任务,远程取消 %d 个任务", |
| | | // removedQueueTasks.size(), |
| | | // tasksToCancelRemotely.size()) |
| | | // ) |
| | | // )) |
| | | // .onErrorResume(e -> { |
| | | // log.error("取消关键词任务失败: {}", e.getMessage()); |
| | | // return Mono.just(ResponseResult.error(500, "取消关键词任务失败: " + e.getMessage())); |
| | | // }); |
| | | // } |
| | | // 线程安全的队列移除方法 |
| | | private List<SearchTaskRequest> removeTasksFromQueueByKeywordId(Integer keywordId) { |
| | | List<SearchTaskRequest> removedTasks = new ArrayList<>(); |
| | | |
| | | // 使用锁保证队列操作的原子性 |
| | | queueLock.lock(); |
| | | try { |
| | | Iterator<SearchTaskRequest> iterator = taskQueue.iterator(); |
| | | while (iterator.hasNext()) { |
| | | SearchTaskRequest task = iterator.next(); |
| | | if (task.getKeyword_id() != null && task.getKeyword_id().equals(keywordId)) { |
| | | removedTasks.add(task); |
| | | iterator.remove(); |
| | | } |
| | | } |
| | | } finally { |
| | | queueLock.unlock(); |
| | | } |
| | | |
| | | log.info("从队列中移除了 {} 个与关键词ID {} 相关的任务", removedTasks.size(), keywordId); |
| | | return removedTasks; |
| | | } |
| | | |
| | | // 从队列中移除所有关键词ID匹配的任务 |
| | | // private List<SearchTaskRequest> removeTasksFromQueueByKeywordId(Integer keywordId) { |
| | | // List<SearchTaskRequest> removedTasks = new ArrayList<>(); |
| | | // |
| | | // // 使用迭代器安全地移除元素 |
| | | // Iterator<SearchTaskRequest> iterator = taskQueue.iterator(); |
| | | // while (iterator.hasNext()) { |
| | | // SearchTaskRequest task = iterator.next(); |
| | | // if (task.getKeyword_id() != null && task.getKeyword_id().equals(keywordId)) { |
| | | // removedTasks.add(task); |
| | | // iterator.remove(); |
| | | // } |
| | | // } |
| | | // |
| | | // log.info("从队列中移除了 {} 个与关键词ID {} 相关的任务", removedTasks.size(), keywordId); |
| | | // return removedTasks; |
| | | // } |
| | | // 发送远程取消请求 |
| | | // 发送远程取消请求(使用Java 8兼容的Map创建方式) |
| | | private Mono<ResponseResult<?>> cancelRemoteTask(String taskId) { |
| | | // 使用Collections.singletonMap或手动创建Map |
| | | Map<String, Object> requestBody = new HashMap<>(); |
| | | requestBody.put("status", "pending"); |
| | | |
| | | return webClient.post() |
| | | .uri(baseUrl + "/api/v1/tasks/" + taskId + "/cancel") |
| | | .contentType(MediaType.APPLICATION_JSON) |
| | | .bodyValue(requestBody) |
| | | .retrieve() |
| | | .onStatus(HttpStatus::isError, response -> response.bodyToMono(String.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException("取消失败: " + errorBody)))) |
| | | .bodyToMono(Void.class) |
| | | .thenReturn(ResponseResult.success("任务已取消")); |
| | | } |
| | | |
| | | // 更新单个任务状态 |
| | | private Mono<Void> updateTaskStatus(String taskId, String status) { |
| | | return Mono.fromRunnable(() -> { |
| | | LambdaUpdateWrapper<KeywordTask> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | updateWrapper.eq(KeywordTask::getTask_id, taskId); |
| | | updateWrapper.set(KeywordTask::getStatus, status); |
| | | keywordTaskService.update(updateWrapper); |
| | | }).subscribeOn(Schedulers.boundedElastic()).then(); |
| | | } |
| | | /* @PostMapping("/cancel/{taskId}") |
| | | @ApiOperation(value = "取消任务") |
| | | public Mono<ResponseResult<TaskCancelResponse>> cancelTask(@PathVariable String taskId) { |
| | | return webClient.post() |
| | | .uri(baseUrl + "/tasks/" + taskId + "/cancel") |
| | | .uri(baseUrl + "/api/v1/tasks/" + taskId + "/cancel") |
| | | .contentType(MediaType.APPLICATION_JSON) |
| | | .bodyValue(Collections.emptyMap()) // 添加空请求体 |
| | | .retrieve() |
| | | .onStatus(HttpStatus::isError, response -> { |
| | | .onStatus(HttpStatus::isError, response -> response.bodyToMono(TaskCancelResponse.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException(errorBody.getDetail())))) |
| | | .bodyToMono(TaskCancelResponse.class) |
| | | .flatMap(cancelResponse -> { |
| | | // 更新关键词状态 |
| | | Mono<Void> updateKeyword = Mono.fromRunnable(() -> { |
| | | LambdaUpdateWrapper<Keyword> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | updateWrapper.eq(Keyword::getTask_id, taskId); |
| | | updateWrapper.set(Keyword::getStatus, "canceled"); // 统一使用"canceled" |
| | | keywordService.update(updateWrapper); |
| | | }) |
| | | .subscribeOn(Schedulers.boundedElastic()) |
| | | .then(); |
| | | |
| | | // 更新关键词任务状态 |
| | | Mono<Void> updateKeywordTask = Mono.fromRunnable(() -> { |
| | | LambdaUpdateWrapper<KeywordTask> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | updateWrapper.eq(KeywordTask::getTask_id, taskId); |
| | | updateWrapper.set(KeywordTask::getStatus, "canceled"); // 统一使用"canceled" |
| | | keywordTaskService.update(updateWrapper); |
| | | }) |
| | | .subscribeOn(Schedulers.boundedElastic()) |
| | | .then(); |
| | | |
| | | // 并行执行两个更新操作,并在完成后返回cancelResponse |
| | | return Mono.when(updateKeyword, updateKeywordTask) |
| | | .thenReturn(cancelResponse); |
| | | }) |
| | | .map(data -> ResponseResult.success(data)) |
| | | .onErrorResume(e -> { |
| | | if (e.getMessage().contains("任务不存在")) { |
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
| | | } else if (e.getMessage().contains("无法取消")) { |
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
| | | } |
| | | return Mono.just(ResponseResult.error(500, e.getMessage())); |
| | | }); |
| | | }*/ |
| | | |
| | | @ApiOperation(value = "获取任务结果") |
| | | @GetMapping("/tasks/{taskId}") |
| | | public Mono<TaskResultResponse> getTaskResult(@PathVariable String taskId) { |
| | | return webClient.get() |
| | | .uri(baseUrl + "/api/v1/tasks/" + taskId + "/result") |
| | | .accept(MediaType.APPLICATION_JSON) |
| | | .retrieve() |
| | | .onStatus(HttpStatus::is4xxClientError, response -> { |
| | | if (response.statusCode() == HttpStatus.NOT_FOUND) { |
| | | return response.bodyToMono(String.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException("任务不存在"))); |
| | | } else if (response.statusCode() == HttpStatus.BAD_REQUEST) { |
| | | return response.bodyToMono(String.class) |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException("任务已经完成,无法取消"))); |
| | | .flatMap(errorBody -> Mono.error(new RuntimeException("任务未完成,无法获取结果"))); |
| | | } |
| | | return response.createException().flatMap(Mono::error); |
| | | }) |
| | | .bodyToMono(TaskCancelResponse.class) |
| | | .map(data -> ResponseResult.success(data)) |
| | | .onErrorResume(e -> { |
| | | if (e.getMessage().contains("任务不存在")) { |
| | | return Mono.just(ResponseResult.error(404, "任务不存在")); |
| | | } else if (e.getMessage().contains("无法取消")) { |
| | | return Mono.just(ResponseResult.error(400, "任务已完成,无法取消")); |
| | | .bodyToMono(TaskResultResponse.class) |
| | | .flatMap(responseResult -> { |
| | | TaskResultResponse result = responseResult; |
| | | |
| | | if (result != null && result.getResults() != null) { |
| | | return updateQuestionAndReference(result) |
| | | .thenReturn(responseResult); |
| | | } |
| | | return Mono.just(ResponseResult.error(500, "取消任务失败: " + e.getMessage())); |
| | | return Mono.just(responseResult); |
| | | }) |
| | | .onErrorResume(e -> { |
| | | System.out.println("获取任务结果失败"); |
| | | TaskResultResponse result = new TaskResultResponse(); |
| | | result.setDetail("获取任务结果失败: " + e.getMessage()); |
| | | return Mono.just(result); |
| | | }); |
| | | } |
| | | // @ApiOperation(value = "获取任务结果") |
| | | // @GetMapping("/tasks/{taskId}/result") |
| | | // public Mono<ResponseResult<TaskResultResponse>> getTaskResult(@PathVariable |
| | | // String taskId) { |
| | | // return webClient.get() |
| | | // .uri(baseUrl + "/tasks/" + taskId + "/result") |
| | | // .retrieve() |
| | | // .onStatus(HttpStatus::isError, response -> response.bodyToMono(String.class) |
| | | // .flatMap(errorBody -> Mono.error(new RuntimeException("获取结果失败: " + |
| | | // errorBody)))) |
| | | // .bodyToMono(TaskResultResponse.class) |
| | | // .flatMap(result -> { |
| | | // // 更新keyword状态 |
| | | // LambdaUpdateWrapper<Keyword> keywordWrapper = new LambdaUpdateWrapper<>(); |
| | | // keywordWrapper.eq(Keyword::getTask_id, taskId) |
| | | // .set(Keyword::getStatus, "completed"); |
| | | // keywordService.update(keywordWrapper); |
| | | |
| | | // // 更新question信息并收集references |
| | | // List<Question> updateQuestions = new ArrayList<>(); |
| | | // List<Reference> references = new ArrayList<>(); |
| | | // private Mono<Void> updateQuestionAndReference(TaskResultResponse result) { |
| | | // return Mono.fromRunnable(() -> { |
| | | // // 1. 更新关键词状态 |
| | | // LambdaUpdateWrapper<Keyword> keywordUpdate = new LambdaUpdateWrapper<>(); |
| | | // keywordUpdate.eq(Keyword::getTask_id, result.getTask_id()) |
| | | // .set(Keyword::getStatus, "completed"); |
| | | // keywordService.update(keywordUpdate); |
| | | // |
| | | // // 查询关键词ID |
| | | // LambdaQueryWrapper<Keyword> keywordQuery = new LambdaQueryWrapper<>(); |
| | | // keywordQuery.eq(Keyword::getTask_id, result.getTask_id()); |
| | | // Keyword keyword = keywordService.getOne(keywordQuery); |
| | | // |
| | | // if (keyword == null) { |
| | | // System.out.println("未找到关联的关键词,task_id: " + result.getTask_id()); |
| | | // return; |
| | | // } |
| | | // |
| | | // // 2. 处理每个用户的问题结果 |
| | | // for (UserResult userResult : result.getResults()) { |
| | | // for (QuestionResult questionResult : userResult.getQuestions_results()) { |
| | | // // 2.1 查询问题ID |
| | | // LambdaQueryWrapper<Question> queryWrapper = new LambdaQueryWrapper<>(); |
| | | // queryWrapper.eq(Question::getQuestion, questionResult.getQuestion()) |
| | | // .eq(Question::getKeyword_id, keyword.getKeyword_id()); |
| | | // Question question = questionService.getOne(queryWrapper); |
| | | // |
| | | // if (question != null) { |
| | | // // 更新问题状态 |
| | | // LambdaUpdateWrapper<Question> updateWrapper = new LambdaUpdateWrapper<>(); |
| | | // updateWrapper.eq(Question::getQuestion_id, question.getQuestion_id()) |
| | | // .set(Question::getStatus, questionResult.getStatus()) |
| | | // .set(Question::getResponse, questionResult.getResponse()) |
| | | // .set(Question::getExtracted_count, questionResult.getExtracted_count()) |
| | | // .set(Question::getError, questionResult.getError()) |
| | | // .set(Question::getTimestamp, LocalDateTime.parse( |
| | | // questionResult.getTimestamp(), |
| | | // DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS") |
| | | // )); |
| | | // questionService.update(updateWrapper); |
| | | // |
| | | // // 2.2 保存引用数据 |
| | | // List<Reference> references = questionResult.getReferences().stream() |
| | | // .map(ref -> { |
| | | // Reference reference = new Reference(); |
| | | // reference.setQuestion_id(question.getQuestion_id()); |
| | | // reference.setTitle(ref.getTitle()); |
| | | // reference.setUrl(ref.getUrl()); |
| | | // reference.setDomain(ref.getDomain()); |
| | | // reference.setCreate_time(LocalDateTime.now()); |
| | | // return reference; |
| | | // }) |
| | | // .collect(Collectors.toList()); |
| | | // |
| | | // if (!references.isEmpty()) { |
| | | // referenceService.saveBatch(references); |
| | | // } |
| | | // } else { |
| | | // System.out.println("未找到匹配的问题,question " + question.getQuestion()); |
| | | // |
| | | // } |
| | | // } |
| | | // } |
| | | // }); |
| | | // } |
| | | /** |
| | | * 获取或创建平台(确保同一domain只创建一次) |
| | | * @param domain 平台域名 |
| | | * @return 已存在或新创建的Platform |
| | | */ |
| | | private Platform getOrCreatePlatform(String domain) { |
| | | // 1. 先尝试查询已存在的平台 |
| | | Platform platform = platformService.getPlatformByDomain(domain); |
| | | if (platform != null) { |
| | | return platform; |
| | | } |
| | | |
| | | // result.getResults().forEach(userResult -> { |
| | | // userResult.getQuestions_results().forEach(qResult -> { |
| | | // Question question = new Question(); |
| | | // question.setQuestion_id(qResult.getQuestion_id()); |
| | | // question.setResponse(qResult.getResponse()); |
| | | // question.setStatus(qResult.getStatus()); |
| | | // updateQuestions.add(question); |
| | | // 2. 若不存在,尝试创建(处理并发场景) |
| | | try { |
| | | // 2.1 获取或创建“默认”类型(Type也需避免重复,建议Type表的type_name也加唯一约束) |
| | | Type defaultType = typeService.getOne(new LambdaQueryWrapper<Type>() |
| | | .eq(Type::getType_name, "默认")); |
| | | if (defaultType == null) { |
| | | defaultType = new Type(); |
| | | defaultType.setType_name("默认"); |
| | | typeService.save(defaultType); // 若Type可能重复,此处也需处理DuplicateKeyException |
| | | } |
| | | |
| | | // // 转换references |
| | | // references.addAll(qResult.getReferences().stream() |
| | | // .map(ref -> new Reference( |
| | | // qResult.getQuestion_id(), |
| | | // ref.getTitle(), |
| | | // ref.getUrl(), |
| | | // ref.getDomain(), |
| | | // result.getTask_id(), |
| | | // )) |
| | | // .collect(Collectors.toList())); |
| | | // }); |
| | | // }); |
| | | // 2.2 构建新平台对象 |
| | | Platform newPlatform = new Platform(); |
| | | newPlatform.setDomain(domain); |
| | | newPlatform.setPlatform_name(domain); // 平台名称默认使用域名,可根据实际需求调整 |
| | | newPlatform.setType_id(defaultType.getType_id()); |
| | | newPlatform.setCreate_time(LocalDateTime.now()); // 补充创建时间 |
| | | |
| | | // // 批量更新和插入 |
| | | // if (!updateQuestions.isEmpty()) { |
| | | // questionService.updateBatchById(updateQuestions); |
| | | // } |
| | | // if (!references.isEmpty()) { |
| | | // referenceService.saveBatch(references); |
| | | // } |
| | | // 2.3 尝试保存,若因唯一约束冲突失败,则捕获异常 |
| | | platformService.save(newPlatform); |
| | | return newPlatform; // 保存成功,返回新创建的平台 |
| | | |
| | | // return Mono.just(ResponseResult.success(result)); |
| | | // }) |
| | | // .onErrorResume(e -> Mono.just(ResponseResult.error(e.getMessage()))); |
| | | // } |
| | | } catch (DuplicateKeyException e) { |
| | | // 3. 若捕获到重复键异常,说明并发创建了,重新查询即可(此时数据库中已存在该平台) |
| | | log.warn("平台domain={}已存在,无需重复创建", domain, e); |
| | | return platformService.getPlatformByDomain(domain); // 重新查询,一定能获取到 |
| | | } catch (Exception e) { |
| | | // 处理其他异常(如数据库连接失败等) |
| | | log.error("创建平台失败,domain={}", domain, e); |
| | | throw new RuntimeException("创建平台失败", e); |
| | | } |
| | | } |
| | | private Mono<Void> updateQuestionAndReference(TaskResultResponse result) { |
| | | return Mono.fromRunnable(() -> { |
| | | try { |
| | | //查看每个账号信息的status是否正常 |
| | | |
| | | // @ApiOperation(value = "获取任务结果") |
| | | // @GetMapping("/tasks/{taskId}/result") |
| | | // public Mono<ResponseResult<TaskResultResponse>> |
| | | // getTaskResultlMono(@PathVariable String taskId) { |
| | | // return webClient.get() |
| | | // .uri(baseUrl + "/tasks/" + taskId + "/result") |
| | | // .accept(MediaType.APPLICATION_JSON) |
| | | // .retrieve() |
| | | // .onStatus(HttpStatus::is4xxClientError, response -> { |
| | | // if (response.statusCode() == HttpStatus.NOT_FOUND) { |
| | | // return response.bodyToMono(String.class) |
| | | // .flatMap(errorBody -> Mono.error(new RuntimeException("任务不存在"))); |
| | | // } else if (response.statusCode() == HttpStatus.BAD_REQUEST) { |
| | | // return response.bodyToMono(String.class) |
| | | // .flatMap(errorBody -> Mono.error(new RuntimeException("任务未完成,无法获取结果"))); |
| | | // } |
| | | // return response.createException().flatMap(Mono::error); |
| | | // }) |
| | | // .bodyToMono(new |
| | | // ParameterizedTypeReference<ResponseResult<TaskResultResponse>>() {}) |
| | | // .flatMap(responseResult -> { |
| | | // TaskResultResponse result = responseResult.getData(); |
| | | // if (result != null && result.getResults() != null) { |
| | | // // 处理结果并更新数据库 |
| | | // return updateQuestionAndReference(result) |
| | | // .thenReturn(responseResult); |
| | | // } |
| | | // return Mono.just(responseResult); |
| | | // }) |
| | | // .onErrorResume(e -> { |
| | | // 1. 根据KeywordTask更新关键词状态 |
| | | // 查询关键词ID |
| | | LambdaQueryWrapper<KeywordTask> keywordTaskWrapper = new LambdaQueryWrapper<>(); |
| | | keywordTaskWrapper.eq(KeywordTask::getTask_id, result.getTask_id()); |
| | | KeywordTask keywordTask = keywordTaskService.getOne(keywordTaskWrapper); |
| | | // keywordTask.setStatus("completed"); |
| | | // keywordTaskService.updateById(keywordTask); |
| | | Keyword keyword = keywordService.getById(keywordTask.getKeyword_id()); |
| | | |
| | | // return Mono.just(ResponseResult.error(e.getMessage())); |
| | | // }); |
| | | // } |
| | | if (keyword == null) { |
| | | System.out.println("未找到关联的关键词,task_id: " + result.getTask_id()); |
| | | //报错 |
| | | throw new Exception("未找到关联的关键词,task_id: " + result.getTask_id()); |
| | | // return; |
| | | } |
| | | LambdaQueryWrapper<KeywordTask> keywordTaskWrapper2 = new LambdaQueryWrapper<>(); |
| | | keywordTaskWrapper2.eq(KeywordTask::getKeyword_id, keyword.getKeyword_id()); |
| | | List<KeywordTask> keywordTasks = keywordTaskService.list(keywordTaskWrapper2); |
| | | |
| | | // 更新问题和引用数据 |
| | | // private Mono<Void> updateQuestionAndReference(TaskResultResponse result) { |
| | | // return Mono.fromRunnable(() -> { |
| | | // // 1. 更新关键词状态 |
| | | // LambdaUpdateWrapper<Keyword> keywordUpdate = new LambdaUpdateWrapper<>(); |
| | | // keywordUpdate.eq(Keyword::getTask_id, result.getTask_id()) |
| | | // .set(Keyword::getStatus, "completed"); |
| | | // keywordService.update(keywordUpdate); |
| | | //如果全部为completed 关键词也为completed ,如果关联关系没有任务id,或者状态为running ,关键词为submitted, |
| | | if (keywordTasks.stream().allMatch(task -> "completed".equals(task.getStatus())) ) { |
| | | keyword.setStatus("completed"); |
| | | keywordService.updateById(keyword); |
| | | |
| | | // // 2. 处理每个用户的问题结果 |
| | | // for (UserResult userResult : result.getResults()) { |
| | | // for (QuestionResult questionResult : userResult.getQuestions_results()) { |
| | | // // 2.1 更新问题状态 |
| | | // LambdaUpdateWrapper<Question> questionUpdate = new LambdaUpdateWrapper<>(); |
| | | // questionUpdate.eq(Question::getTa, result.getTask_id()) |
| | | // .eq(Question::getContent, questionResult.getQuestion()) |
| | | // .set(Question::getStatus, questionResult.getStatus()) |
| | | // .set(Question::getResponse, questionResult.getResponse()) |
| | | // .set(Question::getProcessTime, |
| | | // LocalDateTime.parse(questionResult.getTimestamp())); |
| | | // questionService.update(questionUpdate); |
| | | } |
| | | //如果有一个task为failed设置关键词为false |
| | | else if (keywordTasks.stream().anyMatch(task -> "failed".equals(task.getStatus()))) { |
| | | keyword.setStatus("false"); |
| | | keywordService.updateById(keyword); |
| | | } |
| | | |
| | | // // 2.2 保存引用数据 |
| | | // List<Reference> references = questionResult.getReferences().stream() |
| | | // .map(ref -> { |
| | | // Reference reference = new Reference(); |
| | | // reference.setQuestionId(questionService.getOne(questionUpdate).getId()); |
| | | // reference.setTitle(ref.getTitle()); |
| | | // reference.setUrl(ref.getUrl()); |
| | | // reference.setDomain(ref.getDomain()); |
| | | // reference.setCreateTime(LocalDateTime.now()); |
| | | // return reference; |
| | | // }) |
| | | // .collect(Collectors.toList()); |
| | | |
| | | // if (!references.isEmpty()) { |
| | | // referenceService.saveBatch(references); |
| | | // } |
| | | // } |
| | | // } |
| | | // }); |
| | | // } |
| | | |
| | | // |
| | | //// 定义状态优先级:canceled > false > completed |
| | | // String finalStatus = "completed"; // 默认状态为 completed |
| | | // |
| | | // for (KeywordTask task : keywordTasks) { |
| | | // String status = task.getStatus(); |
| | | //// if ("canceled".equals(status)) { |
| | | //// finalStatus = "canceled"; |
| | | //// break; // 遇到 canceled 直接跳出循环,因为优先级最高 |
| | | //// } else |
| | | // if ("false".equals(status)) { |
| | | // finalStatus = "false"; |
| | | // // 不跳出循环,继续检查是否存在 canceled |
| | | // } |
| | | // } |
| | | // 更新关键词状态 |
| | | |
| | | String orderId = keyword.getOrder_id(); |
| | | if (orderId == null || orderId.isEmpty()) { |
| | | System.out.println("关键词[" + keyword.getKeyword_id() + "]未关联订单,跳过订单状态更新"); |
| | | return; |
| | | } |
| | | |
| | | // 2. 查询该订单下的所有关键词 |
| | | LambdaQueryWrapper<Keyword> orderKeywordsWrapper = new LambdaQueryWrapper<>(); |
| | | orderKeywordsWrapper.eq(Keyword::getOrder_id, orderId); |
| | | List<Keyword> orderKeywords = keywordService.list(orderKeywordsWrapper); |
| | | |
| | | if (orderKeywords.isEmpty()) { |
| | | System.out.println("订单[" + orderId + "]下无关键词,跳过状态更新"); |
| | | return; |
| | | } |
| | | // 3. 检查所有关键词的状态是否均为 completed 或 false |
| | | boolean allValid = orderKeywords.stream() |
| | | .allMatch(k -> "completed".equals(k.getStatus()) || "false".equals(k.getStatus()) || "canceled".equals(k.getStatus())); |
| | | |
| | | // 4. 若所有关键词状态均有效,更新订单状态为3 |
| | | if (allValid) { |
| | | Orders orders = orderService.getById(orderId); |
| | | if (orders != null) { |
| | | orders.setStatus(3); // 假设Orders有Integer类型的status字段 |
| | | orderService.updateById(orders); |
| | | System.out.println("订单[" + orderId + "]所有关键词状态符合条件,已更新状态为3"); |
| | | } else { |
| | | System.out.println("未找到订单[" + orderId + "],无法更新状态"); |
| | | } |
| | | } |
| | | |
| | | Orders orders = orderService.getById(keyword.getOrder_id()); |
| | | |
| | | // 2. 批量查询所有问题 |
| | | LambdaQueryWrapper<Question> queryWrapper = new LambdaQueryWrapper<>(); |
| | | queryWrapper.eq(Question::getKeyword_id, keyword.getKeyword_id()); |
| | | List<Question> questions = questionService.list(queryWrapper); |
| | | |
| | | // 构建问题映射表,用于快速查找 |
| | | Map<String, Question> questionMap = questions.stream() |
| | | .collect(Collectors.toMap(Question::getQuestion, q -> q)); |
| | | |
| | | // 3. 收集所有需要更新的问题和引用 |
| | | List<Question> questionsToUpdate = new ArrayList<>(); |
| | | List<Reference> allReferences = new ArrayList<>(); |
| | | List<Reference> resultList = new ArrayList<>(); |
| | | // 遍历结果 |
| | | for (UserResult userResult : result.getResults()) { |
| | | for (QuestionResult questionResult : userResult.getQuestions_results()) { |
| | | try { |
| | | Question question = questionMap.get(questionResult.getQuestion()); |
| | | if (question != null) { |
| | | // 更新问题对象 |
| | | question.setStatus(questionResult.getStatus()); |
| | | question.setResponse(questionResult.getResponse()); |
| | | question.setExtracted_count(questionResult.getExtracted_count()); |
| | | question.setError(questionResult.getError()); |
| | | question.setKeyword_id(keyword.getKeyword_id()); |
| | | |
| | | // 解析时间戳 |
| | | if (questionResult.getTimestamp() != null) { |
| | | DateTimeFormatter formatter = DateTimeFormatter |
| | | .ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS"); |
| | | question.setTimestamp( |
| | | LocalDateTime.parse(questionResult.getTimestamp(), formatter)); |
| | | } |
| | | //更新 |
| | | // questionService.updateById(question); |
| | | |
| | | questionsToUpdate.add(question); |
| | | |
| | | /* List<Reference> references = |
| | | Optional.ofNullable(questionResult.getReferences()) |
| | | .orElse(Collections.emptyList()) |
| | | .stream() |
| | | .map(ref -> { |
| | | Reference reference = new Reference(); |
| | | reference.setQuestion_id(question.getQuestion_id()); |
| | | reference.setTitle(ref.getTitle()); |
| | | reference.setUrl(ref.getUrl()); |
| | | reference.setDomain(ref.getDomain()); |
| | | reference.setNum(keyword.getNum()); |
| | | reference.setTask_id(result.getTask_id()); |
| | | reference.setKeyword_id(keyword.getKeyword_id()); |
| | | //域名和平台id映射 |
| | | reference.setCreate_time(LocalDateTime.now()); |
| | | Platform platform = platformService.getPlatformByDomain(reference.getDomain()); |
| | | if (platform == null) { |
| | | //平台为空 创建平台 类型为“默认” |
| | | Type type = typeService.getOne(new LambdaQueryWrapper<Type>().eq(Type::getType_name,"默认")); |
| | | if (type == null) { |
| | | Type newType = new Type(); |
| | | newType.setType_name("默认"); |
| | | typeService.save(newType); |
| | | type = newType; |
| | | } |
| | | Platform platform1 = new Platform(); |
| | | platform1.setDomain(reference.getDomain()); |
| | | platform1.setPlatform_name(reference.getDomain()); |
| | | platform1.setType_id(type.getType_id()); |
| | | platformService.save(platform1); |
| | | |
| | | reference.setType_id(type.getType_id()); |
| | | reference.setPlatform_id(platform1.getPlatform_id()); |
| | | |
| | | } |
| | | else { |
| | | reference.setPlatform_id(platform.getPlatform_id()); |
| | | Type type = typeService.getById(platform.getType_id()); |
| | | if (type != null){ |
| | | reference.setType_id(type.getType_id()); |
| | | } |
| | | } |
| | | return reference; |
| | | }) |
| | | .collect(Collectors.toList());*/ |
| | | // 初始化引用列表(避免null) |
| | | List<Reference> references = new ArrayList<>(); |
| | | List<TaskResultResponse.Reference> originalReferences = questionResult.getReferences(); |
| | | if (originalReferences == null) { |
| | | originalReferences = Collections.emptyList(); |
| | | } |
| | | |
| | | // 遍历原始引用列表,转换为Reference对象 |
| | | for (TaskResultResponse.Reference ref : originalReferences) { // 注意:需将“原引用类型”替换为实际类型(如QuestionResult中的引用类型) |
| | | Reference reference = new Reference(); |
| | | // 设置基本字段 |
| | | reference.setQuestion_id(question.getQuestion_id()); |
| | | reference.setTitle(ref.getTitle()); |
| | | reference.setUrl(ref.getUrl()); |
| | | reference.setDomain(ref.getDomain()); |
| | | reference.setNum(keyword.getNum()); |
| | | reference.setTask_id(result.getTask_id()); |
| | | reference.setKeyword_id(keyword.getKeyword_id()); |
| | | reference.setCreate_time(LocalDateTime.now()); |
| | | |
| | | /* // 处理平台和类型关联 |
| | | Platform platform = platformService.getPlatformByDomain(reference.getDomain()); |
| | | if (platform == null) { |
| | | // 平台不存在,创建新平台(类型默认为“默认”) |
| | | Type type = typeService.getOne(new LambdaQueryWrapper<Type>().eq(Type::getType_name, "默认")); |
| | | if (type == null) { |
| | | Type newType = new Type(); |
| | | newType.setType_name("默认"); |
| | | typeService.save(newType); |
| | | type = newType; |
| | | } |
| | | Platform platform1 = new Platform(); |
| | | platform1.setDomain(reference.getDomain()); |
| | | platform1.setPlatform_name(reference.getDomain()); |
| | | platform1.setType_id(type.getType_id()); |
| | | platform1.setCreate_time(LocalDateTime.now()); |
| | | platformService.save(platform1); |
| | | |
| | | // 关联新平台和类型 |
| | | reference.setType_id(type.getType_id()); |
| | | reference.setPlatform_id(platform1.getPlatform_id()); |
| | | } else { |
| | | // 平台已存在,直接关联 |
| | | reference.setPlatform_id(platform.getPlatform_id()); |
| | | Type type = typeService.getById(platform.getType_id()); |
| | | if (type != null) { |
| | | reference.setType_id(type.getType_id()); |
| | | } |
| | | } |
| | | */ |
| | | // 关键:使用优化后的方法获取平台,避免重复创建 |
| | | Platform platform = getOrCreatePlatform(ref.getDomain()); |
| | | reference.setPlatform_id(platform.getPlatform_id()); |
| | | reference.setType_id(platform.getType_id()); // 直接从平台获取类型ID,更可靠 |
| | | // 添加到结果列表 |
| | | references.add(reference); |
| | | } |
| | | // 添加到总引用列表 |
| | | if (!references.isEmpty()) { |
| | | allReferences.addAll(references); |
| | | } |
| | | |
| | | //取数据库中当前关键词的当前轮次的当前问题id结果拿出来 |
| | | List<Reference> dbList = referenceService.list(new LambdaQueryWrapper<Reference>().eq(Reference::getKeyword_id, keyword.getKeyword_id()) |
| | | .eq(Reference::getNum, keyword.getNum()) |
| | | .eq(Reference::getQuestion_id, question.getQuestion_id()) |
| | | ); |
| | | |
| | | // 1. 合并两个列表 |
| | | List<Reference> combinedList = new ArrayList<>(); |
| | | combinedList.addAll(allReferences); |
| | | combinedList.addAll(dbList); |
| | | |
| | | // 2. 创建复合键的Map,用于统计完全匹配的记录 |
| | | Map<String, List<Reference>> compositeKeyMap = combinedList.stream() |
| | | .collect(Collectors.groupingBy( |
| | | ref -> ref.getTitle() + "|" + ref.getUrl() + "|" + ref.getDomain() |
| | | )); |
| | | |
| | | // 3. 处理每组重复记录 |
| | | |
| | | compositeKeyMap.forEach((key, refGroup) -> { |
| | | // 3.1 找出组内有ID的记录(优先从dbList中获取) |
| | | Optional<Reference> existingRecord = refGroup.stream() |
| | | .filter(ref -> ref.getReference_id() != null) |
| | | .findFirst(); |
| | | |
| | | // 3.2 统计该组的重复次数(总数-1) |
| | | int repetitionCount = refGroup.size() - 1; |
| | | |
| | | // 3.3 决定最终保留的记录 |
| | | Reference recordToSave = new Reference(); |
| | | if (existingRecord.isPresent()) { |
| | | // 使用已有ID的记录并更新重复次数 |
| | | recordToSave = existingRecord.get(); |
| | | recordToSave.setRepetition_num( |
| | | (recordToSave.getRepetition_num() == null ? 1 : recordToSave.getRepetition_num()) |
| | | + repetitionCount |
| | | ); |
| | | } else { |
| | | // 没有ID记录则取第一条并设置重复次数 |
| | | recordToSave = refGroup.get(0); |
| | | recordToSave.setRepetition_num(1+repetitionCount); |
| | | } |
| | | |
| | | resultList.add(recordToSave); |
| | | }); |
| | | referenceService.saveOrUpdateBatch(resultList); |
| | | } |
| | | } catch (Exception e) { |
| | | log.error(e.getMessage(), e); |
| | | System.out.println("处理问题结果失败: " + e.getMessage()); |
| | | } |
| | | } |
| | | //更新账号状态 |
| | | if ( "failed".equals(userResult.getStatus())){ |
| | | if (userResult.getError().contains("登录失败")){ |
| | | LambdaUpdateWrapper<User> userWrapper = new LambdaUpdateWrapper<>(); |
| | | userWrapper.eq(User::getUser_email, userResult.getUser_email()); |
| | | userWrapper.set(User::getStatus, "无法登录"); |
| | | userService.update(userWrapper); |
| | | |
| | | }else if (userResult.getError().contains("信息错误")){ |
| | | LambdaUpdateWrapper<User> userWrapper = new LambdaUpdateWrapper<>(); |
| | | userWrapper.eq(User::getUser_email, userResult.getUser_email()); |
| | | userWrapper.set(User::getStatus, "信息错误"); |
| | | userService.update(userWrapper); |
| | | } |
| | | } |
| | | |
| | | } |
| | | |
| | | // 4. 批量更新问题 |
| | | System.out.println(questionsToUpdate); |
| | | if (!questionsToUpdate.isEmpty()) { |
| | | questionService.updateBatchById(questionsToUpdate); |
| | | System.out.println("成功批量更新 " + questionsToUpdate.size() + " 个问题"); |
| | | } |
| | | |
| | | // 5. 批量插入引用,使用流式分批处理 |
| | | // if (!allReferences.isEmpty()) { |
| | | // int batchSize = 1000; |
| | | // IntStream.iterate(0, i -> i + batchSize) |
| | | // .limit((allReferences.size() + batchSize - 1) / batchSize) |
| | | // .forEach(i -> { |
| | | // List<Reference> batch = allReferences.subList( |
| | | // i, Math.min(i + batchSize, allReferences.size())); |
| | | // referenceService.saveBatch(batch); |
| | | // }); |
| | | // System.out.println("成功批量插入 " + allReferences.size() + " 条引用数据"); |
| | | // } |
| | | |
| | | |
| | | } catch (Exception e) { |
| | | log.error("更新问题和引用数据失败: " ,e.getMessage(), e); |
| | | // System.out.println("更新问题和引用数据失败: " + e.getMessage()); |
| | | throw new RuntimeException("更新问题和引用数据失败", e); |
| | | } |
| | | }); |
| | | } |
| | | |
| | | @GetMapping("/tasks/all") |
| | | @ApiOperation(value = "获取所有任务列表") |
| | | public Mono<TaskListResponse> getAllTasks() { |
| | | return webClient.get() |
| | | .uri(baseUrl + "/api/v1/tasks") |
| | | .accept(MediaType.APPLICATION_JSON) |
| | | .retrieve() |
| | | .bodyToMono(new ParameterizedTypeReference<TaskListResponse>() { |
| | | }) |
| | | .onErrorResume(e -> { |
| | | TaskListResponse response = new TaskListResponse(); |
| | | response.setDetail("获取任务列表失败: " + e.getMessage()); |
| | | return Mono.just(response); |
| | | |
| | | // return Mono.just(ResponseResult.error("获取任务列表失败: " + e.getMessage())); |
| | | }); |
| | | } |
| | | |
| | | @GetMapping("/health") |
| | | @ApiOperation("健康检查") |
| | | public Mono<HealthResponse> checkThirdPartyHealth() { |
| | | return webClient.get() |
| | | .uri(baseUrl + "/health") // 假设第三方健康检查接口路径为/health |
| | | .retrieve() |
| | | .bodyToMono(HealthResponse.class) |
| | | .onErrorResume(e -> Mono.just( |
| | | new HealthResponse("unhealthy", null, "", e.getMessage()))); |
| | | } |
| | | |
| | | /** |
| | | * 查询服务器资源 |
| | | */ |
| | | @GetMapping("/server/resource") |
| | | @ApiOperation(value = "查询服务器资源") |
| | | public Mono<ServerResourceResponse> getServerResource() { |
| | | return webClient.get() |
| | | .uri(baseUrl + "/api/v1/system/resources") |
| | | .retrieve() |
| | | .bodyToMono(ServerResourceResponse.class) |
| | | .onErrorResume(e -> Mono.just( |
| | | new ServerResourceResponse( e.getMessage()))); |
| | | } |
| | | /** |
| | | * 传入orderid查所有关键词id以及关键词下面的所有任务id,轮询所有任务状态,如果状态为completed,则循环调用获取结果接口,处理结果 |
| | | */ |
| | | |
| | | |
| | | } |