- c - 在位数组中找到第一个零
- linux - Unix 显示有关匹配两种模式之一的文件的信息
- 正则表达式替换多个文件
- linux - 隐藏来自 xtrace 的命令
我有一个客户正在使用的图书馆,他们正在通过 DataRequest
具有 userid
的对象, timeout
以及其中的一些其他字段。现在我用这个 DataRequest
对象创建 URL,然后我使用 RestTemplate
进行 HTTP 调用我的服务返回一个 JSON 响应,我用它来制作 DataResponse
对象并返回此 DataResponse
反对他们。
下面是我的DataClient
客户通过传递使用的类 DataRequest
反对它。我正在使用客户在 DataRequest
中传递的超时值如果在 getSyncData
中花费太多时间,则请求超时方法。
public class DataClient implements Client {
private RestTemplate restTemplate = new RestTemplate();
// first executor
private ExecutorService service = Executors.newFixedThreadPool(15);
@Override
public DataResponse getSyncData(DataRequest key) {
DataResponse response = null;
Future<DataResponse> responseFuture = null;
try {
responseFuture = getAsyncData(key);
response = responseFuture.get(key.getTimeout(), key.getTimeoutUnit());
} catch (TimeoutException ex) {
response = new DataResponse(DataErrorEnum.CLIENT_TIMEOUT, DataStatusEnum.ERROR);
responseFuture.cancel(true);
// logging exception here
}
return response;
}
@Override
public Future<DataResponse> getAsyncData(DataRequest key) {
DataFetcherTask task = new DataFetcherTask(key, restTemplate);
Future<DataResponse> future = service.submit(task);
return future;
}
}
DataFetcherTask
类:
public class DataFetcherTask implements Callable<DataResponse> {
private DataRequest key;
private RestTemplate restTemplate;
public DataFetcherTask(DataRequest key, RestTemplate restTemplate) {
this.key = key;
this.restTemplate = restTemplate;
}
@Override
public DataResponse call() throws Exception {
// In a nutshell below is what I am doing here.
// 1. Make an url using DataRequest key.
// 2. And then execute the url RestTemplate.
// 3. Make a DataResponse object and return it.
// I am calling this whole logic in call method as LogicA
}
}
截至目前我的DataFetcherTask
类负责一个DataRequest
键如上所示..
问题陈述:-
现在我有一个小的设计更改。客户将通过DataRequest
(例如 keyA)对象到我的库,然后我将使用 DataRequest
中存在的用户 ID 对另一个服务(我在当前设计中没有这样做)进行新的 http 调用。 (keyA) 对象,它将返回用户 ID 列表,因此我将使用这些用户 ID 并制作其他一些 DataRequest
(keyB, keyC, keyD) 对象对应响应中返回的每个用户 ID。然后我会有 List<DataRequest>
将具有 keyB、keyC 和 keyD 的对象 DataRequest
目的。 List<DataRequest>
中的最大元素将是三个,仅此而已。
现在对于每个 DataRequest
List<DataRequest>
中的对象我想执行上面的 DataFetcherTask.call
并行方法然后制作List<DataResponse>
通过添加每个 DataResponse
对于每个键。所以我将对 DataFetcherTask.call
进行三个并行调用.此并行调用背后的想法是在相同的全局超时值中获取所有这些最多三个键的数据。
所以我的建议是 - DataFetcherTask
类将返回 List<DataResponse>
对象而不是 DataResponse
然后是 getSyncData
的签名和 getAsyncData
方法也会改变。所以这里是算法:
List<DataRequest>
通过调用另一个 HTTP 服务。DataRequest
进行并行调用在 List<DataRequest>
至 DataFetcherTask.call
方法并返回 List<DataResponse>
反对客户而不是DataResponse
.通过这种方式,我也可以在步骤 1 和步骤 2 上应用相同的全局超时。如果上述任何一步都需要时间,我们将在 getSyncData
中超时。方法。
DataFetcherTask
设计更改后的类:
public class DataFetcherTask implements Callable<List<DataResponse>> {
private DataRequest key;
private RestTemplate restTemplate;
// second executor here
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 = generateKeys();
CompletionService<DataResponse> comp = new ExecutorCompletionService<>(executorService);
int count = 0;
for (final DataRequest key : keys) {
comp.submit(new Callable<DataResponse>() {
@Override
public DataResponse call() throws Exception {
return performDataRequest(key);
}
});
}
List<DataResponse> responseList = new ArrayList<DataResponse>();
while (count-- > 0) {
Future<DataResponse> future = comp.take();
responseList.add(future.get());
}
return responseList;
}
// In this method I am making a HTTP call to another service
// and then I will make List<DataRequest> accordingly.
private List<DataRequest> generateKeys() {
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.
return keys;
}
private DataResponse performDataRequest(DataRequest key) {
// This will have all LogicA code here which is shown in my original design.
// everything as it is same..
}
}
现在我的问题是 -
call
另一个方法 call
方法看起来很奇怪? 我已经简化了代码,这样我的想法就很清楚了。
最佳答案
正如您的问题的评论中已经提到的,您可以使用 Java 的 ForkJoin 框架。这将为您节省 DataFetcherTask
中的额外线程池。
您只需在 DataClient
中使用 ForkJoinPool
并将您的 DataFetcherTask
转换为 RecursiveTask
(一个ForkJoinTask
的子类型)。这使您可以轻松地并行执行其他子任务。
因此,经过这些修改后,您的代码将如下所示:
DataFetcherTask
DataFetcherTask
现在是一个 RecursiveTask
,它首先生成键并为每个生成的键调用子任务。这些子任务在与父任务相同的 ForkJoinPool
中执行。
public class DataFetcherTask extends RecursiveTask<List<DataResponse>> {
private final DataRequest key;
private final RestTemplate restTemplate;
public DataFetcherTask(DataRequest key, RestTemplate restTemplate) {
this.key = key;
this.restTemplate = restTemplate;
}
@Override
protected List<DataResponse> compute() {
// Create subtasks for the key and invoke them
List<DataRequestTask> requestTasks = requestTasks(generateKeys());
invokeAll(requestTasks);
// All tasks are finished if invokeAll() returns.
List<DataResponse> responseList = new ArrayList<>(requestTasks.size());
for (DataRequestTask task : requestTasks) {
try {
responseList.add(task.get());
} catch (InterruptedException | ExecutionException e) {
// TODO - Handle exception properly
Thread.currentThread().interrupt();
return Collections.emptyList();
}
}
return responseList;
}
private List<DataRequestTask> requestTasks(List<DataRequest> keys) {
List<DataRequestTask> tasks = new ArrayList<>(keys.size());
for (DataRequest key : keys) {
tasks.add(new DataRequestTask(key));
}
return tasks;
}
// In this method I am making a HTTP call to another service
// and then I will make List<DataRequest> accordingly.
private List<DataRequest> generateKeys() {
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.
return keys;
}
/** Inner class for the subtasks. */
private static class DataRequestTask extends RecursiveTask<DataResponse> {
private final DataRequest request;
public DataRequestTask(DataRequest request) {
this.request = request;
}
@Override
protected DataResponse compute() {
return performDataRequest(this.request);
}
private DataResponse performDataRequest(DataRequest key) {
// This will have all LogicA code here which is shown in my original design.
// everything as it is same..
return new DataResponse(DataErrorEnum.OK, DataStatusEnum.OK);
}
}
}
数据客户端
DataClient
除了新的线程池外不会有太大变化:
public class DataClient implements Client {
private final RestTemplate restTemplate = new RestTemplate();
// Replace the ExecutorService with a ForkJoinPool
private final ForkJoinPool service = new ForkJoinPool(15);
@Override
public List<DataResponse> getSyncData(DataRequest key) {
List<DataResponse> responsList = null;
Future<List<DataResponse>> responseFuture = null;
try {
responseFuture = getAsyncData(key);
responsList = responseFuture.get(key.getTimeout(), key.getTimeoutUnit());
} catch (TimeoutException | ExecutionException | InterruptedException ex) {
responsList = Collections.singletonList(new DataResponse(DataErrorEnum.CLIENT_TIMEOUT, DataStatusEnum.ERROR));
responseFuture.cancel(true);
// logging exception here
}
return responsList;
}
@Override
public Future<List<DataResponse>> getAsyncData(DataRequest key) {
DataFetcherTask task = new DataFetcherTask(key, this.restTemplate);
return this.service.submit(task);
}
}
使用 Java8 后,您可以考虑将实现更改为 CompletableFuture
。然后它看起来像这样:
DataClientCF
public class DataClientCF {
private final RestTemplate restTemplate = new RestTemplate();
private final ExecutorService executor = Executors.newFixedThreadPool(15);
public List<DataResponse> getData(DataRequest initialKey) {
return CompletableFuture.supplyAsync(() -> generateKeys(initialKey), this.executor)
.thenApply(requests -> requests.stream().map(this::supplyRequestAsync).collect(Collectors.toList()))
.thenApply(responseFutures -> responseFutures.stream().map(future -> future.join()).collect(Collectors.toList()))
.exceptionally(t -> { throw new RuntimeException(t); })
.join();
}
private List<DataRequest> generateKeys(DataRequest key) {
return new ArrayList<>();
}
private CompletableFuture<DataResponse> supplyRequestAsync(DataRequest key) {
return CompletableFuture.supplyAsync(() -> new DataResponse(DataErrorEnum.OK, DataStatusEnum.OK), this.executor);
}
}
如评论中所述,Guava 的 ListenableFuture
将为 Java7 提供类似的功能,但如果没有 Lambda,它们往往会变得笨拙。
关于java - 从调用方法并行执行方法,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/34316353/
有没有办法同时运行 2 个不同的代码块。我一直在研究 R 中的并行包,它们似乎都基于在循环中运行相同的函数。我正在寻找一种同时运行不同函数的方法(循环的 1 次迭代)。例如,我想在某个数据对象上创建一
无论如何增加 Parallel.For 启动后的循环次数?示例如下: var start = 0; var end = 5; Parallel.For(start, end, i => { C
我是 Golang 的新手,正在尝试了解并发和并行。我阅读了下面提到的关于并发和并行的文章。我执行了相同的程序。但没有得到相同的(混合字母和字符)输出。首先获取所有字母,然后获取字符。似乎并发不工作,
我正在寻找同时迭代 R 中两个或多个字符向量/列表的方法,例如。有没有办法做这样的事情: foo <- c('a','c','d') bar <- c('aa','cc','dd') for(i in
我对 Raku 很陌生,我对函数式方法有疑问,尤其是 reduce。 我最初有这样的方法: sub standardab{ my $mittel = mittel(@_); my $foo =
我最近花了很多时间来学习实时音频处理的细节,我发现的大多数库/工具都是c / c++代码或脚本/图形语言的形式,并在其中编译了c / c++代码。引擎盖。 使用基于回调的API,与GUI或App中的其
我正在使用 JMeter 进行图像负载测试。我有一个图像名称数组并遍历该数组,我通过 HTTP 请求获取所有图像。 -> loop_over_image - for loop controller
我整个晚上都在困惑这个问题...... makeflags = ['--prefix=/usr','--libdir=/usr/lib'] rootdir='/tmp/project' ps = se
我正在尝试提高计算图像平均值的方法的性能。 为此,我使用了两个 For 语句来迭代所有图像,因此我尝试使用一个 Parallel For 来改进它,但结果并不相同。 我做错了吗?或者是什么导致了差异?
假设您有一个并行 for 循环实现,例如ConcRT parallel_for,将所有工作放在一个 for 循环体内总是最好的吗? 举个例子: for(size_t i = 0; i < size()
我想并行运行一部分代码。目前我正在使用 Parallel.For 如何让10、20或40个线程同时运行 我当前的代码是: Parallel.For(1, total, (ii) =>
我使用 PAY API 进行了 PayPal 自适应并行支付,其中无论用户(买家)购买什么,都假设用户购买了总计 100 美元的商品。在我的自适应并行支付中,有 2 个接收方:Receiver1 和
我正在考虑让玩家加入游戏的高效算法。由于会有大量玩家,因此算法应该是异步的(即可扩展到集群中任意数量的机器)。有细节:想象有一个无向图(每个节点都是一个玩家)。玩家之间的每条边意味着玩家可以参加同一场
我有一个全局变量 volatile i = 0; 和两个线程。每个都执行以下操作: i++; System.out.print(i); 我收到以下组合。 12、21 和 22。 我理解为什么我没有得到
我有以下称为 pgain 的方法,它调用我试图并行化的方法 dist: /***************************************************************
我有一个 ruby 脚本读取一个巨大的表(约 2000 万行),进行一些处理并将其提供给 Solr 用于索引目的。这一直是我们流程中的一大瓶颈。我打算在这里加快速度,我想实现某种并行性。我对 Ru
我正在研究 Golang 并遇到一个问题,我已经研究了几天,我似乎无法理解 go routines 的概念以及它们的使用方式。 基本上我是在尝试生成数百万条随机记录。我有生成随机数据的函数,并将创建一
我希望 for 循环使用 go 例程并行。我尝试使用 channel ,但没有用。我的主要问题是,我想在继续之前等待所有迭代完成。这就是为什么在它不起作用之前简单地编写 go 的原因。我尝试使用 ch
我正在使用 import Control.Concurrent.ParallelIO.Global main = parallel_ (map processI [1..(sdNumber runPa
我正在尝试通过 makePSOCKcluster 连接到另一台计算机: library(parallel) cl ... doTryCatch -> recvData -> makeSOCKm
我是一名优秀的程序员,十分优秀!