java对接Dify的工作流API(实战篇)

@RestController@RequestMapping("/workflow")public class WorkFlowController {    @Autowired    private WorkFlowService workFlowService;
    /**     * 上传文件     *     * @param file     * @return     * @throws IOException     */    @PostMapping("/upload")    public WorkFlowFileVo upload(@RequestParam("file") MultipartFile file) throws IOException {        return workFlowService.upload(file);    }
    /**     * 执行工作流     *     * @param workFlowRunDto     * @return     */    @PostMapping("/runWorkFlow")    public SseEmitter runWorkFlow(@RequestBody WorkFlowRunDto workFlowRunDto) {        return workFlowService.runWorkFlow(workFlowRunDto);    }
    /**     * 获取工作流执行情况     *     * @param workflowRunId     * @return     */    @GetMapping("/workFlowInfo")    public WorkFlowExeVo workFlowRunInfo(String workflowRunId) {        return workFlowService.workFlowRunInfo(workflowRunId);    }
}

public interface WorkFlowService {    /**     * 上传文件     *     * @param file     * @return     */    public WorkFlowFileVo upload(@RequestParam("file") MultipartFile file) throws IOException;
    /**     * 执行工作流     *     * @param workFlowRunDto     * @return     */    public SseEmitter runWorkFlow(@RequestBody WorkFlowRunDto workFlowRunDto);
    /**     * 获取工作流执行情况     *     * @param workflowRunId     * @return     */    public WorkFlowExeVo workFlowRunInfo(String workflowRunId);
}

     /**     * 上传文件     *     * @param file     * @return     * @throws IOException     */    @Override    public WorkFlowFileVo upload(MultipartFile file) throws IOException {        // 设置请求头        HttpHeaders headers = new HttpHeaders();        headers.setContentType(MediaType.MULTIPART_FORM_DATA);        headers.set("Authorization", difyConfig.getSaveDataAuthorization());        // 创建请求实体        MultiValueMap<String, Object> body = new LinkedMultiValueMap<>();        // 处理文件内容        body.add("file", new ByteArrayResource(file.getBytes()) {            @Override            public String getFilename() {                // 返回原始文件名                return file.getOriginalFilename();            }        });
        HttpEntity<MultiValueMap<String, Object>> requestEntity = new HttpEntity<>(body, headers);        String uploadUrl = difyConfig.getSaveDataUrl() + "/files/upload";
        // 发送请求        ResponseEntity<String> response = restTemplate.exchange(uploadUrl, HttpMethod.POST, requestEntity, String.class);        log.info("上传文件的response: {}", response);
        WorkFlowFileVo workFlowFileVo = JSON.parseObject(response.getBody(), WorkFlowFileVo.class);        // 构建工作流请求体        WorkFlowRunDto workFlowRunDto = buildWorkFlowRunDto(workFlowFileVo.getId());        // 执行工作流        this.runWorkFlow(workFlowRunDto);    }

     /**     * 执行工作流     *     * @param workFlowRunDto     * @return     */    @Override    public SseEmitter runWorkFlow(WorkFlowRunDto workFlowRunDto) {        // 设置超时时间 300s        SseEmitter emitter = new SseEmitter(300_000L);
        // 使用异步线程处理流式响应        ExecutorService executor = Executors.newSingleThreadExecutor();        executor.execute(() -> {            try {                String runUrl = difyConfig.getSaveDataUrl() + "/workflows/run";                log.info("runUrl: {}", runUrl);
                // 创建请求头                HttpHeaders headers = new HttpHeaders();                headers.set("Authorization", difyConfig.getSaveDataAuthorization());                headers.setContentType(MediaType.APPLICATION_JSON);                // 设置接收流                headers.set(HttpHeaders.ACCEPT, MediaType.TEXT_EVENT_STREAM_VALUE);                HttpEntity<WorkFlowRunDto> requestEntity = new HttpEntity<>(workFlowRunDto, headers);
                // 使用restTemplate的execute方法处理流式响应                restTemplate.execute(                        runUrl,                        HttpMethod.POST,                        request -> {                            // 关键修复：手动设置Content-Type和请求头                            request.getHeaders().setContentType(MediaType.APPLICATION_JSON);                            request.getHeaders().addAll(requestEntity.getHeaders());
                            // 写入请求体                            if (requestEntity.getBody() != null) {                                new ObjectMapper().writeValue(request.getBody(), requestEntity.getBody());                            }                        },                        response -> {                            try (BufferedReader reader = new BufferedReader(new InputStreamReader(response.getBody()))) {                                // 标记是否已处理workflow_run_id                                boolean workflowRunIdProcessed = false;                                String line;                                while ((line = reader.readLine()) != null) {                                    if (line.startsWith("event: ping")) {                                        // 跳过心跳事件                                        continue;                                    }                                    emitter.send(line);                                    log.info("line: {}", line);                                    if (!workflowRunIdProcessed) {                                        try {                                            workflowRunIdProcessed = processLine(line);                                        } catch (Exception e) {                                            log.error("记录id出现异常: {}", e.getMessage());                                            emitter.completeWithError(e);                                        }                                    }                                }                            }                            emitter.complete();                            return null;                        }                );
            } catch (Exception e) {                log.error("处理过程中发生错误: {}", e.getMessage());                emitter.completeWithError(e);            } finally {                log.info("流式输出结束...");            }        });        // 关闭线程池        executor.shutdown();        log.info("流式输出完成...");        return emitter;    }

    @Override    public WorkFlowExeVo workFlowRunInfo(String workflowRunId) {        log.info("获取到的工作流id: {}", workflowRunId);
        // 设置请求头        HttpHeaders headers = new HttpHeaders();        headers.set("Authorization", difyConfig.getWorkFlowAuthorization());
        String workFlowInfoUrl = difyConfig.getWorkFlowUrl() + "/workflows/run/" + workflowRunId;
        HttpEntity<String> requestEntity = new HttpEntity<>(headers);
        // 发送请求        ResponseEntity<String> response = restTemplate.exchange(workFlowInfoUrl, HttpMethod.GET, requestEntity, String.class);        log.info("response: {}", response);
        return JSON.parseObject(response.getBody(), WorkFlowExeVo.class);    }