Многопоточный анализатор веб-сайтов и проблемы с памятью

Question

Привет. Я кодирую синтаксический анализатор веб-сайта, который должен быть быстрым и, следовательно, многопоточным. Внешние библиотеки, которые я использую: HTTP-клиент apache, Jsoup (для анализа HTML) и GPars (для потоков, управляемых сообщениями). Сейчас я покажу некоторую концепцию, которую я пытаюсь реализовать.

static StaticDispatchActor<String> httpActor;
    public static void main(String[] args) throws ClientProtocolException, IOException {
        int numThreads = 25;
        try{
            numThreads = Integer.parseInt(args[0]);
        } catch (Exception e){
            System.out.println("Number of threads defaulted to "+numThreads);
        }
        final int numberOfThreads = numThreads;
        final ExecutorService threadpool = Executors.newCachedThreadPool();
        final Async async = Async.newInstance().use(threadpool);
        AtomicInteger jobCount = new AtomicInteger(0);
//.....
// This is a parser itself which parses usernames out of every page.
        Actor jsoupUser = new StaticDispatchActor<String>(){ // actor to parse users
            HashSet<String> users = new HashSet<>(); // found users
            public void onMessage(String html){ // takes html -> adds parsed users 2 set
                users.addAll(Jsoup.parse(html)
                    .select("a[href*=/profile/]").stream() // select links
                    .map(e -> e.text()) // extract usernames
                    .filter(s -> s.length() > 0) // empty lines -> out
                    .collect(Collectors.toSet()));
                System.out.print("Users: "+users.size()+", Jobs: "+jobCount.get()+"\r");

            }
        }.start();
// This actor shall extract new links to parse out of every page
        Actor jsoupLinker = new StaticDispatchActor<String>(){ // link extractor
            HashSet<String> usedLinks = new HashSet<>(); // already found links
            public synchronized void removeBack(String url){

            @Override
            public void onMessage(String html) {
                Set<String> links =  Jsoup.parse(html).select("a[href]").stream().parallel()
                .map(e -> e.attr("href").replace("#comments", "")// here also some replacements...
                )
                .filter(s -> (!usedLinks.contains(s) && /* other filters */ )
                .collect(Collectors.toSet());
                links.forEach(url -> httpActor.send(url)); // send to process new URLs
            }
        }.start(); // start actor
// this actor is the processor of new links and where the error comes in:
httpActor = new StaticDispatchActor<String>(){ // process responses async
            public void onMessage(String url) {
                try{
                while(jobCount.get()>numberOfThreads); // wait for running threads to be less than wanted value; without this number of running jobs goes out of any control
                async.execute(Request.Get(defaultWebSiteUrl+url), new FutureCallback<Content>(){ @Override // do request and process async
                    public void completed(Content c) {
                        jobCount.decrementAndGet();
                        try{
                        String s = c.asString();
                        jsoupUser.send(s);
                        jsoupLinker.send(s);
                        } catch (OutOfMemoryError e1){
                            System.out.println("out of my memory, "); // This is the thrown error the question is about - [1]
                        }
                    }

            @Override public void failed(Exception e) {
                        jobCount.decrementAndGet();
                        try {
                            throw e;
                        } catch (ConnectException e4){ // if the request is timed out resend it
                            httpActor.send(url);
                            System.out.println("resent\r\n");
                        } catch (HttpResponseException e0){

                        } catch (Exception e1) { // for all other exceptions
                            e1.printStackTrace();
                        }
                    }
            @Override public void cancelled() {
                jobCount.decrementAndGet(); // never done actually
            }
        });
            jobCount.incrementAndGet();
        } catch (IllegalArgumentException e3){
            System.out.println("some illigal shit");
        }
    }
};
httpActor.start();

Теперь проблема в том, что, хотя я ограничил количество запущенных заданий, мой код почему-то выходит из памяти (найдите [1] в коде, чтобы узнать, где). Может быть, у вас есть идеи о том, как ее решить. Или есть некоторая демонстрация для подобной задачи, потому что я заполняю очень неправильно обо всем проекте приложения и возможно я должен изменить это вообще? Спасибо.

Answer 1

Итак, используя подсказку BiziClop, я смог понять ошибку. Если кому-то это интересно, я, как вы можете видеть выше, отправлял HTML-код, полученный с сервера, в виде строки двум различным субъектам, а затем, внутри этих участников, анализировал их. Это было причиной всех ошибок нехватки памяти, поскольку эти HTML-страницы довольно большие, особенно учитывая, сколько из них ожидает обработки в очереди сообщений. Решение, которое я использовал, - просто проанализировать документ и выбрать необходимые элементы и передать их список соответствующим акторам для дальнейшей обработки.

Document doc = Jsoup.parse(c.asString());
jsoupUser.send(doc.select("a[href*=/profile/]"));
jsoupLinker.send(doc.select("a[href]"));

Тем не менее, если кому-то есть что сказать о том, как улучшить алгоритм, я буду очень признателен.