| | |
| | | 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); |
| | | List<List<UserDto>> userBatches = splitUsersIntoBatches(searchTaskRequest.getUsers(), maxConcurrentUsers,keywordId); |
| | | |
| | | // 获取 keywordId |
| | | Integer keywordId = searchTaskRequest.getKeyword_id(); |
| | | |
| | | //分割 |
| | | |
| | | |
| | |
| | | .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, "任务不存在")); |
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
| | | } else if (e.getMessage().contains("无法取消")) { |
| | | return Mono.just(ResponseResult.error(200, "任务已完成,无法取消")); |
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
| | | } |
| | | return Mono.just(ResponseResult.error(500, "取消任务失败: " + e.getMessage())); |
| | | return Mono.just(ResponseResult.error(500, e.getMessage())); |
| | | }); |
| | | } |
| | | |