类星体纤维在线程启动后返回空结果

Posted 2023-03-28

【中文标题】类星体纤维在线程启动后返回空结果

【英文标题】：quasar fiber returning empty results after the thread is started

【发布时间】：2021-02-18 00:04:44

【问题描述】：

我正在我的 Spring Boot 应用程序上本地测试我的 POST 端点。我有一个方法可以生成一个光纤线程来运行一组调用端点 A 的指令，并且我的 POST 端点返回 A 返回的结果。但是，当我的 POST 请求完成时，邮递员中显示的结果为空。 我的代码如下

@RequestMapping("/prediction")
    public CustomResponse prediction(@RequestBody CustomRequest input, HttpServletRequest request) 
        return predictionClass.prediction(input);
    

public CustomResponse prediction(CustomRequest input) 
        CustomResponse customResponse = new customResponse();
        new Fiber<CustomResponse>(new SuspendableRunnable() 

            public void  run() throws SuspendExecution, InterruptedException 
                
                List<CustomRequest> inputs = new ArrayList<>();

                // A for loop is here to duplicate CustomRequest input parameter received and populate the inputs list
                
                List<CustomResponse> customResponses = inputs.stream()
                        .map(req -> processPrediction(req)).collect(Collectors.toList());

                for (CustomResponse x : customResponses) 
                    if (inputs.size() > 1) 
                        for (String outputKey : x.getOutputVars().keySet()) 
                            customResponse.getOutputVars().put(x.getModelName() + "_" + outputKey, x.getOutputVars().get(outputKey));
                        
                     else 
                        // Else statement will be run because the input is only size 1
                        customResponse.getOutputVars().putAll(x.getOutputVars());
                    
                
                System.out.println(customResponse.getOutputVars().size());
            
        ).start();
        return customResponse;

    

    public CustomResponse processPrediction(CustomRequest input) 
        CustomResponse res = new CustomResponse();

        RestTemplate gzipRestTemplate = new RestTemplateBuilder()
                .additionalInterceptors(new GzipHttpRequestInterceptor())
                .build();

        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity<Map<String, Object>> entity = new HttpEntity<>(input, headers);

        ResponseEntity<Map> responseEntity = gzipRestTemplate.postForEntity("an-endpoint-url", entity, Map.class);

        Map<String, Object> outputs = (Map<String, Object>) responseEntity.getBody();

        res.getOutputVars().putAll(outputs);

        return res;

    

在这个测试中，我的输入只有大小 1，当我使用 Postman 触发 POST 请求时，System.out.println(customResponse.getOutputVars().size()); 返回 16，但在 Postman 上它显示我的 outputVars 为空。

有趣的是，我决定做如下 2 个实验。

实验 1

public CustomResponse prediction() 
        CustomResponse customResponse = new CustomResponse ();
        new Fiber<Void>(new SuspendableRunnable() 

            public void  run() throws SuspendExecution, InterruptedException 
                customResponse .setModelName("name");
                Map<String, Object> test = new HashMap<>();
                test.put("pcd4Score", "hello");
                customResponse .getOutputVars().put("message", "hello");
            
        ).start();
        return customResponse ;

    

Postman 返回 customResponse，其中包含消息和 hello

实验 2

此实验与实验 1 相同，但使用 Thread.sleep(1000);我在想 thread.sleep 可以代表 processPrediction 我在我的原始代码中

public CustomResponse prediction() 
        CustomResponse customResponse = new CustomResponse ();
        new Fiber<Void>(new SuspendableRunnable() 

            public void  run() throws SuspendExecution, InterruptedException 
                customResponse .setModelName("name");
                Map<String, Object> test = new HashMap<>();
                test.put("pcd4Score", "hello");
                customResponse .getOutputVars().put("message", "hello");
            
        ).start();
        return customResponse ;
    

这次 customResponse 是空的，在我的 Spring Boot 应用程序终端中，错误是

[quasar] ERROR: while transforming the-path-to-my-class-for-prediction-method$1: Unable to instrument the-path-to-my-class-for-prediction-method$1#run()V because of blocking call to java/lang/Thread#sleep(J)V

感觉实验 1 是成功的，因为指令不是 cpu 密集型的，我知道我可以用一种单独的方法启动纤程的方式对其进行编码，然后只调用 prediction，因为看起来就像邮递员在空的CustomResponse中返回，然后只有run()里面的指令开始运行，我只是想了解Fiber的行为。我在谷歌搜索我的情况时遇到了麻烦（我的谷歌关键字是休息端点在光纤线程启动后不返回结果）因此我在 *** 上问这个问题。我对 java 中的整个多线程主题也很陌生。

【参考方案1】：

我通过在customResponse 像这样返回之前添加光纤连接解决了这个问题。但是只为 .join() 尝试并捕获似乎不是很优雅，有没有更优雅的方法来重做整个方法？

public CustomResponse prediction(CustomRequest input) 
        CustomResponse customResponse = new customResponse();
        Fiber fiber = new Fiber<CustomResponse>(new SuspendableRunnable() 

            public void  run() throws SuspendExecution, InterruptedException 
                
                List<CustomRequest> inputs = new ArrayList<>();

                // A for loop is here to duplicate CustomRequest input parameter received and populate the inputs list
                
                List<CustomResponse> customResponses = inputs.stream()
                        .map(req -> processPrediction(req)).collect(Collectors.toList());

                for (CustomResponse x : customResponses) 
                    if (inputs.size() > 1) 
                        for (String outputKey : x.getOutputVars().keySet()) 
                            customResponse.getOutputVars().put(x.getModelName() + "_" + outputKey, x.getOutputVars().get(outputKey));
                        
                     else 
                        // Else statement will be run because the input is only size 1
                        customResponse.getOutputVars().putAll(x.getOutputVars());
                    
                
                System.out.println(customResponse.getOutputVars().size());
            
        ).start();
        try 
          fiber.join();
         catch (Exception e) 
          e.printStackTrace();
        

        return customResponse;