多线程foreach hashmap循环

Question

我存储的数据借助JSoup检索并提交给我自己的HTTP API。

问题：如何迭代我的HashMap与多线程，没有每个线程对待我的HashMap相同的值，就像它是目前的情况。

其实：

Thread1: a 
Thread2: a 
Thread3: a 
Thread4: a 

Thread1: b 
Thread2: b 
Thread3: b 
Thread4: b 

我想是这样的：

Thread1 : a 
Thread2 : b 
Thread3 : c 
Thread4 : d 

package ygg.org; 

import java.io.IOException; 
import java.net.URLEncoder; 
import java.util.ArrayList; 
import java.util.HashMap; 
import java.util.Hashtable; 
import java.util.Set; 
import java.util.concurrent.ConcurrentHashMap; 

import org.jsoup.Jsoup; 
import org.jsoup.nodes.Document; 
import org.jsoup.nodes.Element; 
import org.jsoup.select.Elements; 

public class Filmstreaming1 { 

    final static int NB_PAGE = 2; 
    final static int THREADS = 1; 
    static ConcurrentHashMap<String, String> movies_list = new ConcurrentHashMap<>(); 
    static int count = 0; 

    static void Initialize() { 

     System.out.println("----------------------------------"); 
     System.out.println("Homer is starting..."); 
     System.out.println("------------------------------"); 

     for (int i = 1 ; i <= NB_PAGE ; i++) { 
      try { 

       Document page = Jsoup.connect("http://xxxxxxx.com/page/" + i + "/") 
            .userAgent("Mozilla") 
            .timeout(3000) 
            .post(); 
       Elements movies = page.getElementsByClass("margin-b40").get(0).getElementsByClass("short-link").select("a"); 
       for (Element movie : movies) { 
        String href = movie.attr("href"); 
        String movie_title = movie.text().replaceAll("\\(.*\\)", ""); 
        boolean isMovieExists = movies_list.containsKey(href); 
        if (isMovieExists == false) { 
         movies_list.put(href, movie_title); 
         System.out.println("Ajout du film " + movie_title); 
        }       
       } 
       System.out.println("Total récupérés " + movies_list.size() + " page : " + i); 
      } catch(IOException ioe) { 
       System.out.println("Exception: " + ioe); 
      } 
     } 
     try { 
      for (int i = 0; i <= THREADS; i++) { 
       Thread api = new ThreadApi(); 
       api.start(); 
      } 
     } catch(Exception e) { 
      System.out.println("Exception: " + e.getMessage()); 
     } 
    } 
} 

class ThreadApi extends Thread { 

    public void run() { 
     while(true) { 

      Filmstreaming1.movies_list.forEach((key, value) -> { 

       try { 

        String code = key.substring(key.indexOf("com/") + 4, key.indexOf("-")); 

        Document page = Jsoup.connect("http://xxxxxxx.com/" + code + "--.html") 
               .userAgent("Mozilla") 
               .timeout(3000) 
               .post(); 

        String director = page.getElementsByClass("finfo-text").get(5).text().toString(); 

        Document page1 = Jsoup.connect("http://xxxxxxx.com/play.php?newsid=" + code + "&vt=ol&sr=3") 
               .referrer("http://xxxxxxx.com/" + code + "--.html") 
               .userAgent("Mozilla") 
               .timeout(3000) 
               .post(); 


        String link = page1.getElementsByTag("iframe").first().attr("src").toString(); 

        String encoded_title = URLEncoder.encode((String) value, "UTF-8"); 
        String encoded_director = URLEncoder.encode((String) director, "UTF-8"); 

        String url = "http://xxxxxxx.com/api/movie?movie=" + encoded_title + "&director=" + encoded_director; 

        // On affiche l'url 
        System.out.println(url); 

        Document api = Jsoup.connect(url) 
               .userAgent("Mozilla") 
               .timeout(3000) 
               .get(); 

        String response = api.text(); 
        System.out.println(response); 
        if (response == "-1") { 
         System.out.println("Erreur");  

        } else { 

         url = "http://xxxxxxx.com/api/video?link=" + link + "&ref=" + response + "&version=vf";    

         Document submit = Jsoup.connect(url) 
                .userAgent("Mozilla") 
                .timeout(3000) 
                .get(); 

         response = submit.text(); 

         Filmstreaming1.movies_list.remove(key); 
         System.out.println(response); 
        } 
       } catch(Exception e) { 
         System.out.println("Exception " + e.getMessage()); 
       } 
      }); 
     } 
    } 
} 

Answer 1

当你的Map已经是一个ConcurrentHashMap，你可以只使用ConcurrentHashMap.forEach - 这使得配置paralleslismThreshold其如果超过阈值，可能会自动并行执行调用。

的文档有以下要说的阈值参数的影响：

这些批量操作接受parallelismThreshold说法。如果当前地图大小估计小于给定阈值，则方法顺序进行。使用Long.MAX_VALUE的值可以抑制所有的并行性。使用1的值通过划分为足够的子任务来充分利用用于所有并行计算的ForkJoinPool.commonPool（），从而获得最大的并行度。通常情况下，您最初会选择这些极端值中的一个，然后衡量使用中间开销与吞吐量之间的值的性能。

所以没必要创建自己的Thread甚至Runnable - 实施，充当BiConsumer<? super K,? super V>会做的任何方法引用或lambda。