在 Spring Boot 中调用 API 的多线程方法

Question

我需要在 Springboot 中实现多线程，同时使用 POST 方法调用 API。我根据 SELECT 查询从 oracle 数据库中提取记录，然后使用 RowMapper 逐条检查每条记录。在下一步中，我只是调用一个方法将这些记录发送到 API，以 postmapping 的形式发送这些记录并取回记录。

因为选择查询一次可以返回 10,20 或 100 条记录。逐个调用每条记录并不理想。我在想是否可以一次发送这些多条记录。我不知道如何实现它，因为我是 Springboot 新手并且仍在学习它。

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

import com.demo.payengine.api.ProviderUpdateAPI;
import com.demo.payengine.config.DataSourceDbConfig;
import com.demo.payengine.pojo.ProvRecordMappingPOJO;

public class ProvRecordProcessing {
    
    //declare variables
    private static String tn_id                     ;
    private static String pr_id                     ; 
    private static String pr_entity                 ;
    private static String pr_cl_eft_ind             ;    
    private static String pr_edi_dest_ind           ; 
    private static String pr_ra_dest_ind            ; 
    private static String br_id                     ; 
    private static String br_account_no             ; 
    private static String br_account_name           ; 
    private static String br_account_number_qual    ;

    @Autowired
    DataSourceDbConfig dbConfig;
    
    @Autowired
    ProviderUpdateAPI provUpdate;
    
    DataSource dataSource;
    
    
    public void dbProvRecordProcessing() {
        
        dataSource = dbConfig.dataSource();
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        
        
        String sqlQuery ="tn_id, pr_id, pr_entity, pr_cl_eft_ind, pr_edi_dest_ind, pr_ra_dest_ind, \r\n" + 
                "br_id, br_account_no, br_account_name, br_account_number_qual\r\n" + 
                "from mc_pr_prv pr \r\n" + 
                "inner join sg_prb_enrll_fl enroll\r\n" + 
                "on pr.tn_id = enroll.pr_payee_id\r\n" + 
                "inner join mc_mbr_bank_rel br\r\n" + 
                "on br.br_ck = pr.br_ck\r\n" + 
                "where pr.pr_entity ='G'";
        
        
        RowMapper<ProvRecordMappingPOJO> rowMapper = new RowMapper<ProvRecordMappingPOJO>(){

            @Override
            public ProvRecordMappingPOJO mapRow(ResultSet rs, int rowNum) throws SQLException {
                tn_id                   = rs.getString("tn_id")                     ;
                pr_id                   = rs.getString("pr_id")                     ;
                pr_entity               = rs.getString("pr_entity")                 ;
                pr_cl_eft_ind           = rs.getString("pr_cl_eft_ind")             ;
                pr_edi_dest_ind         = rs.getString("pr_edi_dest_ind ")          ;
                br_id                   = rs.getString("br_id")                     ;
                br_account_no           = rs.getString("br_account_no")             ;
                br_account_name         = rs.getString("br_account_name")           ;
                br_account_number_qual  = rs.getString("br_account_number_qual")    ;
                
                
                return new ProvRecordMappingPOJO(tn_id, pr_id, pr_entity, pr_cl_eft_ind, pr_edi_dest_ind, 
                        pr_ra_dest_ind, br_id, br_account_no, br_account_name, br_account_number_qual);
                }
            };
        
        //intialize sql query
        List<ProvRecordMappingPOJO> provRecords = jdbcTemplate.query(sqlQuery, rowMapper);
        
        for(ProvRecordMappingPOJO record : provRecords) {
            
            System.out.println("Processing provider record.." + record);
            
            //This method will call API <--- how to make it multi threaded
            provUpdate.provUpdateAPI(tn_id, pr_id, pr_entity, pr_cl_eft_ind, pr_edi_dest_ind, pr_ra_dest_ind, 
                    br_id, br_account_no, br_account_name, br_account_number_qual); 
            
        }
        
    }
    
}

Answer 1

创建一个类ProvUpdateExecutor，它接受一个ProvUpdateResult 对象。为了便于阅读，我用ProvUpdateResult 对象封装了下面方法中的所有方法参数

provUpdate.provUpdateAPI(tn_id, pr_id, pr_entity, pr_cl_eft_ind, pr_edi_dest_ind, pr_ra_dest_ind, 
                    br_id, br_account_no, br_account_name, br_account_number_qual);

ProvUpdateExecutor 类的职责是以多线程方式执行 API 调用

public class ProvUpdateExecutor {

  private ExecutorService executorService = Executors.newFixedThreadPool(noThreads);
  private ProviderUpdateAPI providerUpdateAPI;

  public ProvUpdateExecutor(ProviderUpdateAPI providerUpdateAPI) {
    this.providerUpdateAPI = providerUpdateAPI;
  }

  public void execute(List<ProvUpdateResult> provUpdateResults) {
    List<Future<?>> futures = new ArrayList<>();
    for (ProvUpdateResult provUpdateResult : provUpdateResults) {
      futures.add(executorService.submit(() -> providerUpdateAPI.provUpdateAPI(provUpdateResult)));
    }

    for (Future<?> future : futures) {
      try {
        future.get();
      } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
      }
    }
  }
}

noThreads 是可配置的（通过配置参数传递）。请注意，创建线程池会产生成本，因此您可能希望将此类创建为singleton abject。

在你的方法调用provUpdateExecutor.execute(provUpdateResults)外循环

Answer 2

更好的选择可能是 CompletableFuture API，创建 API 调用的 CompletableFutureAPI 列表，然后将它们连接在一起，一旦所有这些都完成提取结果。

检查这两个网页：

1. 这将为您的需要提供一个实际用例 -> https://theboreddev.com/combining-multiple-api-calls-with-completablefuture/
2. 这将更好地理解并行运行和连接多个 completableFuture -> https://www.baeldung.com/java-completablefuture#Multiple