代码之家 › 专栏 › 技术社区 › Prateek Narendra

Java JSONArray与迭代的并行化

parallel-processing multithreading json java

Prateek Narendra · 技术社区 · 6 年前

我有一段代码,我想并行化,因为它们是独立的操作-

        List<StockQuote> topGainers = new ArrayList<StockQuote>();
        JSONObject jsonObject = (JSONObject)new JSONParser().parse(
                new InputStreamReader(response.getEntity().getContent(), "UTF-8"));
        JSONArray dataArray = (JSONArray) jsonObject.get("data");

        //Parallelize the for loop
        for(int iter=0;iter<dataArray.size();iter++) {
            JSONObject temp = (JSONObject) dataArray.get(iter);
            System.out.println(temp.get("symbol"));
            //getQuote function has network calls. 
            //Serial implementation makes it take too much time
            topGainers.add(this.getQuote((String)temp.get("symbol")));
        }

        return topGainers;

如何将这段代码并行化?ArrayList及其Add操作是否线程安全?

我试过这个-

        int size = dataArray.size();
        ForkJoinPool forkJoinPool = new ForkJoinPool(size);
        forkJoinPool.submit(() ->
        IntStream.range(1, size).parallel().filter( (IntPredicate)i -> { 
        JSONObject temp = (JSONObject) dataArray.get(i);
            System.out.println(temp.get("symbol"));
            try {
                System.out.println(temp.get("symbol"));
                topGainers.add(this.getQuote((String)temp.get("symbol")));
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        return true;}));

我得到的topGainers数组为空

1 回复 | 直到 6 年前

Amit Bera 6 年前

ArrayList 不是 Synchronized 收集所以 ArrayList#add 不是线程安全的。使用 CopyOnWriteArrayList 相反,它是线程安全的。Java-8为我们提供了一些优秀的特性,其中之一是 parallelStream 整个系列。所以你可以利用这一点。但是,如果任务给您带来更多好处,您可以随时创建自己的线程池并执行任务。

下面的示例可能会帮助您

    CopyOnWriteArrayList<StockQuote> topGainers = new CopyOnWriteArrayList<>();
    dataArray.parallelStream().forEach(e->{
        JSONObject temp = (JSONObject) dataArray.get(iter);
        System.out.println(temp.get("symbol"));
        //getQuote function has network calls. 
        //Serial implementation makes it take too much time
        topGainers.add(this.getQuote((String)temp.get("symbol")));
    });

您的云也使用 Collections#synchronizedList 。

评论中问题1的答案: dataArray.parallelStream().forEach 将并行迭代列表。

从评论中回答问题2:

   dataArray.parallelStream().forEach(e->{
        JSONObject temp = (JSONObject) dataArray.get(iter);
        System.out.println(temp.get("symbol"));
        //getQuote function has network calls. 
        //Serial implementation makes it take too much time
        if (some condition here) {
           throw new YourException("Your exception messge");
         }
        topGainers.add(this.getQuote((String)temp.get("symbol")));
    });