java - 并行调用方法时的线程安全问题

Question

我有一个图书馆，客户正在其中传递 DataRequest具有包含 userid 和其他字段的信息的对象。我们使用DataRequest对象对两个不同的 REST 服务进行 HTTP 调用，然后进行 DataResponse反对并将其退回给客户。我的库中有一个全局级别的超时，该超时应用于 HTTP 调用，如果调用超时，那么我们只需在执行 DataResponse 时向客户返回超时错误消息。对象。

给定 DataRequest对象我将对服务进行 HTTP 调用，该服务将返回一些内容，然后在此基础上我将创建 List，然后针对每个 DataRequest我将调用的对象 performDataRequest方法在相同的全局超时中并行，我在getSyncData中方法然后 make List<DataResponse>对象并返回响应。

下面是我的DataClient客户将通过 DataRequest 调用哪个类对象:

public class DataClient implements Client {

    private RestTemplate restTemplate = new RestTemplate();
    private ExecutorService service = Executors.newFixedThreadPool(15);

    @Override
    public List<DataResponse> getSyncData(DataRequest key) {
        List<DataResponse> response = new ArrayList<DataResponse>();
        Future<List<DataResponse>> responseFuture = null;

        try {
            responseFuture = getAsyncData(key);
            response = responseFuture.get(key.getTimeout(), key.getTimeoutUnit());
        } catch (TimeoutException ex) {
            response.add(new DataResponse(DataErrorEnum.CLIENT_TIMEOUT, DataStatusEnum.ERROR));
            responseFuture.cancel(true); // terminating the tasks that have got timed out
            // logging exception here               
        }

        return response;
    }   

    @Override
    public Future<List<DataResponse>> getAsyncData(DataRequest key) {
        DataFetcherTask task = new DataFetcherTask(key, restTemplate);
        Future<List<DataResponse>> future = service.submit(task);

        return future;
    }
}

下面是我的DataFetcherTask完成所有工作的类:

public class DataFetcherTask implements Callable<List<DataResponse>> {

    private DataRequest key;
    private RestTemplate restTemplate;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);

    public DataFetcherTask(DataRequest key, RestTemplate restTemplate) {
        this.key = key;
        this.restTemplate = restTemplate;
    }

    @Override
    public List<DataResponse> call() throws Exception {
        List<DataRequest> keys = performKeyRequest();
        List<Future<DataResponse>> responseFutureList = new ArrayList<Future<DataResponse>>();

        for (final DataRequest key : keys) {
            responseFutureList.add(executorService.submit(new Callable<DataResponse>() {
                @Override
                public DataResponse call() throws Exception {
                    return performDataRequest(key);
                }
            }));
        }

        List<DataResponse> responseList = new ArrayList<DataResponse>();
        for (Future<DataResponse> future : responseFutureList) {
            responseList.add(future.get());
        }

        return responseList;
    }

    private List<DataRequest> performKeyRequest() {
        List<DataRequest> keys = new ArrayList<>();
        // use key object which is passed in contructor to make HTTP call to another service
        // and then make List of DataRequest object and return keys.
        // max size of keys list will be three.
        return keys;
    }       

    private DataResponse performDataRequest(DataRequest key) {
        Mappings mappings = ShardMapping.getMappings(key.getType());
        List<String> hostnames = mappings.getAllHostnames(key);

        for (String hostname : hostnames) {
            if (DataUtils.isEmpty(hostname) || ShardMapping.isBlockHost(hostname)) {
                continue;
            }
            try {
                String url = generateUrl(hostname);
                URI uri = URI.create(url);
                ResponseEntity<String> response = restTemplate.exchange(uri, HttpMethod.GET, key.getEntity(), String.class);

                ShardMapping.unblockHost(hostname);
                if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
                    return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
                            DataStatusEnum.SUCCESS);
                } else {
                    return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);
                }
            } catch (HttpClientErrorException | HttpServerErrorException ex) {
                HttpStatusCodeException httpException = ex;
                DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
                String errorMessage = httpException.getResponseBodyAsString();
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
                // logging exception here                   
            } catch (RestClientException ex) {
                ShardMapping.blockHost(hostname);
                // logging exception here                                       
            }
        }

        return new DataResponse(DataErrorEnum.SERVICE_UNAVAILABLE, DataStatusEnum.ERROR);       
    }
}

问题陈述:-

• 我的代码线程安全吗？我调用的方式performDataRequest方法与调用方法并行？
• 其次，拥有 call 感觉很奇怪另一个方法call做这个工作？为此，我还有两名执行者，其中一名在 DataClient 内。具有 15 个线程的类和 DataFetcherTask 中的其他线程有 10 个线程的类。不确定这是否是正确的方法？还有更好的办法吗？

Answer 1

Is my code thread safe with the way I am calling performDataRequest method in parallel from call method?

大部分，但不完全。一个线程是否可以在另一个线程调用 ShardMapping.getMapping() 时修改 ShardMapping？例如，ShardMapping.unblockHost()是否修改了ShardMapping？如果是这样，如果两个线程尝试同时调用 ShardMapping.unblockHost()，您就完蛋了。这有道理吗？

修复方法是让 PerformDataRequest() 仅执行 HTTP 请求，而不执行 ShardMapping 逻辑。像这样的事情:

private DataResponse performDataRequest(URI uri, DataRequest key) {
       try {
           ResponseEntity<String> response = restTemplate.exchange(uri, HttpMethod.GET, key.getEntity(), String.class);
           if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
                return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
                        DataStatusEnum.SUCCESS);
            } else {
                return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);
            }
        } catch (HttpClientErrorException | HttpServerErrorException ex) {
            HttpStatusCodeException httpException = ex;
            DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
            String errorMessage = httpException.getResponseBodyAsString();
            return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            // logging exception here
         } catch (RestClientException ex) {
            return null;
            // logging exception here                                       
         }  
}                 

然后将 ShardMapping 代码移到 future 之外，在 for (final DataRequest key :keys) { 循环中。

And secondly it feels very weird to have call method inside another call to do this job? And for this I am having two executors as well, one inside DataClient class with 15 threads and other in DataFetcherTask class with 10 threads. Not sure whether this is the right way to do? Is there any better way?

这有点愚蠢，而且不是最好的方法。现在，您的设置如下所示:

                          +----------------- performDataRequest()
                          |
         max 3 sec        |       
getAsyncData --- DataFetcherTask ----------- performDataRequest()
                          |
                          |
                          +----------------- performDataRequest()

相反，为什么不在 performDataRequest() future 上设置 3 秒超时，然后正常调用 DataFetcherTask.call() 呢？

                                  max 3 sec
                          +----------------- performDataRequest()
                          |
                          |       max 3 sec
                 DataFetcherTask ----------- performDataRequest()
                          |
                          |       max 3 sec
                          +----------------- performDataRequest()