多线程实际应用

背景：批量上传身份证、协议照片等，要并发调用机构网关上传接口

第一步：初始化

private ExecutorService executorService;

@PostConstruct
public void init() {
    BlockingQueue queue = new LinkedBlockingQueue<>(200);
    executorService = new ThreadPoolExecutor(
            8, 12,
            120, TimeUnit.SECONDS, queue,
            (new ThreadFactoryBuilder()).setNameFormat("inst-access-thread-%d").build());
}

第二步：编写实际执行逻辑

public TaskExcutorResult executeInternal(TaskExcutorReq request, Map context) {

    List instProcessNodeList = institutionProcProcessCache.getInstitutionProcessNodeList(
            currentUnionCode, PROCESS_CODE, ProcessTypeEnum.UNIT.getCode());

    TaskExcutorResult excutorResult;
    Map requestMap = assemblyInstitutionVariable(context, dispatcher);
    List> futureList = new ArrayList();
    for (String instProcessNode : instProcessNodeList) {
        // 实现同一个接口重复调用多次且变量名称相同的情况
        String fileKeyList = FileUploadOpCodeEnum.getFileKeyListByOpCode(instProcessNode, currentUnionCode, dispatcher.getTaskType());
        String[] fileKeys = fileKeyList.split(",");
        for (String fileKey : fileKeys) {
            Map requestMapTmp = Maps.newHashMap();
            requestMapTmp.putAll(requestMap);
            requestMapTmp.put(InstGateWayVariableName.PROTOCOL_FILE_TYPE, fileKey);
            requestMapTmp.put(InstGateWayVariableName.PROTOCOL_FILE_BOS_KEY, requestMap.get(fileKey));
            InstitutionAccessThread institutionAccessThread = new InstitutionAccessThread(requestMapTmp, instProcessNode, currentUnionCode,
                    dispatcher.getCustomerId(), institutionGateWayRpcService);
            String traceKey = LogUtil.getMainThreadTraceKey();
            futureList.add(executorService.submit(() -> {
                MDC.put(MDC_TRACE_KEY, traceKey);
                return institutionAccessThread.call();
            }));
        }
    }

    if (CollectionUtils.isEmpty(futureList)) {
        excutorResult = new TaskExcutorResult(TaskBaseResp.FAILED, TaskBaseResp.FAIL_MSG);
        excutorResult.setTaskTransactionCode(transaction.getTransactionCode());
        return excutorResult;
    }

    List resultList = new ArrayList<>();

    for (Future future : futureList) {
        try {
            InstitutionAccessResult institutionAccessResult = future.get(5000, TimeUnit.MILLISECONDS);
            if (InsRisCode.FAIL.getRetCode().equals(institutionAccessResult.getAccessResult())) {
                excutorResult = new TaskExcutorResult(TaskBaseResp.FAILED, TaskBaseResp.FAIL_MSG);
                excutorResult.setNonblocking(true);
                excutorResult.setTaskTransactionCode(transaction.getTransactionCode());
                return excutorResult;
            }
            resultList.add(institutionAccessResult);
        } catch (Exception e) {
            LOG.error("请求机构发生异常", e);
        }
    }

    if (CollectionUtils.isEmpty(resultList)) {
        excutorResult = new TaskExcutorResult(TaskBaseResp.FAILED, TaskBaseResp.FAIL_MSG);
        excutorResult.setTaskTransactionCode(transaction.getTransactionCode());
        return excutorResult;
    }

    excutorResult = new TaskExcutorResult(TaskBaseResp.SUCCESS, TaskBaseResp.SUCCESS_MSG);
    excutorResult.setTaskTransactionCode(transaction.getTransactionCode());
    return excutorResult;
}

第三步：并发执行

InstitutionAccessThread 实现了 Callable接口

执行 InstitutionAccessThread类的call方法

public class InstitutionAccessThread implements Callable {
    private static final FLogger logger = FLoggerFactory.getLogger(InstitutionAccessThread.class);

    private Map requestMap;

    private String opCode;

    private String unionCode;

    private long customerId;

    private InstitutionGateWayRpcService institutionGateWayRpcService;

    //执行逻辑线程创建
    public InstitutionAccessThread(Map requestMap, String opCode, String unionCode, long customerId,
                                   InstitutionGateWayRpcService institutionGateWayRpcService) {
        this.requestMap = requestMap;
        this.opCode = opCode;
        this.unionCode = unionCode;
        this.customerId = customerId;
        this.institutionGateWayRpcService = institutionGateWayRpcService;
    }

    public InstitutionAccessResult call() throws Exception {

        InstGatewayResponse> response = institutionGateWayRpcService.dispatch(requestMap,
                opCode, unionCode, customerId);

        InstitutionAccessResult result = new InstitutionAccessResult();
        result.setOpCode(opCode);
        result.setUnionCode(unionCode);

        if (Objects.nonNull(response) && InstErrorCodeEnum.SUCCESS.getCode().equals(response.getRespCode())) {

            String risCode = (String) response.getResult().getOrDefault(InsParameterName.RIS_CODE, InsRisCode.SUCCESS.getRetCode());
            result.setAccessResult(risCode);
        } else {
            result.setAccessResult(InsRisCode.FAIL.getRetCode());
        }

        return result;
    }

}

补：返回结果类

public class InstitutionAccessResult {

    private String opCode;

    private String unionCode;

    private String accessResult;

    public String getOpCode() {
        return opCode;
    }

    public void setOpCode(String opCode) {
        this.opCode = opCode;
    }

    public String getUnionCode() {
        return unionCode;
    }

    public void setUnionCode(String unionCode) {
        this.unionCode = unionCode;
    }

    public String getAccessResult() {
        return accessResult;
    }

    public void setAccessResult(String accessResult) {
        this.accessResult = accessResult;
    }
}

复杂版：