Java ExecutorService Runnable 不更新值

Question

我正在使用 Java 下载 HTML 其 URL 存储在数据库中的网站的内容。我也想将他们的 HTML 放入数据库。为此，我正在使用 Jsoup：

public String downloadHTML(String byLink) {
        String htmlInPage = "";
        try {
            Document doc = Jsoup.connect(byLink).get();
            htmlInPage = doc.html();
        } catch (org.jsoup.UnsupportedMimeTypeException e) {
            // process this and some other exceptions
        } 
        return htmlInPage;
    }

我想同时下载网站并使用此功能：

public void downloadURL(int websiteId, String url, 
                        String categoryName, ExecutorService executorService) {
   executorService.submit((Runnable) () -> {
       String htmlInPage = downloadHTML(url);
       System.out.println("Category: " + categoryName + " " + websiteId + " " + url);
       String insertQuery = 
              "INSERT INTO html_data (website_id, html_contents)  VALUES (?,?)";
       dbUtils.query(insertQuery, websiteId, htmlInPage);   
   });
}

dbUtils 是我的 class 基于 Apache Commons DbUtils。详情在这里：http://pastebin.com/iAKXchbQ
我以这样的方式使用上面提到的所有内容：（List<Object[]> 详细信息也在 pastebin 上进行了解释）

public static void main(String[] args) {
        DbUtils dbUtils = new DbUtils("host", "db", "driver", "user", "pass");
        List<String> categoriesList = 
                     Arrays.asList("weapons", "planes", "cooking", "manga");
        String sql = "SELECT lw.id, lw.website_url, category_name " +
                "FROM list_of_websites AS lw JOIN list_of_categories AS lc " +
                "ON lw.category_id = lc.id " +
                "where category_name = ? ";

        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (String category : categoriesList) {
            List<Object[]> sitesInCategory = dbUtils.select(sql, category );
            for (Object[] entry : sitesInCategory) {
                int websiteId = (int) entry[0];
                String url = (String) entry[1];
                String categoryName = (String) entry[2];
                downloadURL(websiteId, url, categoryName, executorService);
            }
        }
        executorService.shutdown();
}

我不确定这个解决方案是否正确，但它确实有效。现在我想修改代码以保存 HTML 不是来自我数据库中的所有网站，而是仅保存它们在每个类别中的固定数量。
例如，从 "weapons" 类别下载并保存 50 个网站的 HTML，从 "planes" 类别下载并保存 50 个，等等。我认为没有必要为此目的使用 sql : 如果我们 select 每个类别 50 个站点，这并不意味着我们将它们全部保存，因为可能存在语法错误和连接问题。
我已经尝试创建单独的 class 实现 Runnable 字段：counter 和 maxWebsitesPerCategory，但这些变量没有更新。另一个想法是创建字段 Map<String,Integer> sitesInCategory 而不是 counter，将每个类别作为键放在那里并递增其值直到达到 maxWebsitesPerCategory，但它也不起作用。请帮我！
P.S：对于与我实现并发下载相关的任何建议，我也将不胜感激（我之前没有在 Java 中使用过并发，这是我的第一次尝试）

Answer 1

这个怎么样？

for (String category : categoriesList) {
        dbUtils.select(sql, category).stream()
            .limit(50)
            .forEach(entry -> {
                int websiteId = (int) entry[0];
                String url = (String) entry[1];
                String categoryName = (String) entry[2];
                downloadURL(websiteId, url, categoryName, executorService);
            });
    }

sitesInCategory 已替换为最多 50 个元素的流，那么您的代码在每个条目上都是运行。

编辑

关于评论。我已经继续并进行了一些重组，您可以modify/implement我建议的方法的内容。

public void werk(Queue<Object[]> q, ExecutorService executorService) {
    executorService.submit(() -> {
        try {
            Object[] o = q.remove();
            try {
                String html = downloadHTML(o); // this takes one of your object arrays and returns the text of an html page

                insertIntoDB(html); // this is the code in the latter half of your downloadURL method
            }catch (/*narrow exception type indicating download failure*/Exception e) {
                werk(q, executorService);
            }
        }catch (NoSuchElementException e) {}
    });
}

^^^ 这种方法完成了大部分工作。

for (String category : categoriesList) {
    Queue<Object[]> q = new ConcurrentLinkedQueue<>(dbUtils.select(sql, category));
    IntStream.range(0, 50).forEach(i -> werk(q, executorService));
}

^^^ 这是主

中的 for 循环

现在每个类别尝试下载 50 个页面，如果下载一个页面失败，它会继续尝试下载另一个页面。这样，您将下载 50 个页面或已尝试下载该类别中的所有页面。

Java ExecutorService Runnable 不更新值

Java ExecutorService Runnable doesn't update value

java

concurrency

runnable

executorservice

java.util.concurrent

编辑