在任务仍在运行时返回结果

Posted 2023-02-27

【中文标题】在任务仍在运行时返回结果

【英文标题】：Returning results while a task is still running

【发布时间】：2021-10-07 00:08:20

【问题描述】：

我有一个方法需要在调用后立即返回结果，甚至这个方法中的所有任务都没有完成。

在这种情况下：

用户提交存入资金的地址 我想保存他们的地址并向他们发送存款地址 我想启动一个任务，开始检查用户是否将资金存入存款地址

第 2 步是我要返回的内容，第 3 步是我想要在返回第 3 步后继续在后台运行的内容。

我不确定如何安排这些事件。以下是我的方法，不确定如何处理addressRegistrationService.watchAndTransact(userAddresses); 任务。这可能需要一整天的时间才能完成，并且用户需要返回存款地址才能存入资金。

@PostMapping("/registeraddress")
public ResponseEntity<Mono<UserAddressesDTO>> addUserAddress(@RequestBody UserAddresses userAddresses) throws Exception 
    log.info("Registering  with addresses ", userAddresses.getAccountId(), userAddresses.getAddresses());

    //Checking if we can use the user provided addresses and account id
    if (addressRegistrationService.isRegistrationValid(userAddresses)) 
        throw new InvalidAddressException("Address is invalid");
    

    log.info("Deposit Address is ", generateJobcoinAddress.generateJobcoinAddress());

    //Generate the deposit address and add it the UserAddress
    String depositAddress = generateJobcoinAddress.generateJobcoinAddress();

    //Add input and deposit address to the object
    UserAddressesDTO userAddressesDTO = new UserAddressesDTO(userAddresses.getAccountId(), depositAddress, userAddresses.getAddresses());

    //Request the Jobcoin Service to start watching for the transaction
    //Once Jobcoin Service detects it will post the transaction to the house account
    //then to the user addresses - > we will be notified separately once this is complete
    addressRegistrationService.watchAndTransact(userAddresses);

    //Store addresses in database, calls the data-service to store these details
    return ResponseEntity.ok().body(addressRegistrationService.saveAddressDB(userAddressesDTO));

【参考方案1】：

您可以使用CompletableFuture。所以你的代码看起来像......

CompletableFuture.runAsync(() -> addressRegistrationService.watchAndTransact(userAddresses));

它将在不同的线程中运行您的 watchAndTransact 方法。主线程不会等待结果。所以基本上它会在后台运行。

注意：但万一失败，您将不知道发生了什么。因此，您可以添加一些自定义指标和日志。这可以是服务本身的一部分。

CompletableFuture.runAsync(() -> 
    try 
        addressRegistrationService.watchAndTransact(userAddresses);
        // Notify user
     catch (RuntimeException e) 
        // custom metrics implementation here
        // log error here
    
);

【讨论】：

请注意，默认情况下runAsync 使用 Fork-Join 池，该池具有有限的线程数，不适用于 IO 操作。

@MartinTarjányi 那你有什么建议呢？或者如何配置runAsync以便我可以将其用于IO

取决于 watchAndTransact 方法内部发生的事情。如果这是一种反应式方法，那么 subscribe 可能就足够了。否则 runAsync 有一个重载版本，它接受一个执行器，您可以在其中传递自定义线程池。