src/main/java/org/example/MultiThreadCrawler.java - answer - mykniz/hw2

src/main/java/org/example/MultiThreadCrawler.java
224 строки8 KB
Шундеев Николай Николаевич
working answer with countDownLatch
12 ноя 2025, 13:01
12 ноя 2025, 13:015332dec
package org.example;

import lombok.extern.log4j.Log4j2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Log4j2
public class MultiThreadCrawler {

    private final WikiClient client = new WikiClient();
    private final int threadPoolSize;
    private final int maxDepth;

    // Атомарный счётчик запросов
    private final AtomicInteger requestCounter = new AtomicInteger(0);
    private final int maxRequestLimit;

    // Потокобезопасная очередь для BFS
    private final BlockingQueue<Node> searchQueue = new LinkedBlockingQueue<>();

    // Потокобезопасный кеш посещённых страниц: title -> depth
    private final ConcurrentMap<String, Integer> visited = new ConcurrentHashMap<>();

    // Флаг завершения
    private final AtomicBoolean found = new AtomicBoolean(false);

    // Результат
    private volatile Node resultNode = null;

    private final CountDownLatch countDownLatch = new CountDownLatch(1);

    public MultiThreadCrawler(int threadPoolSize, int maxDepth, int maxRequestLimit) {
        this.threadPoolSize = threadPoolSize;
        this.maxDepth = maxDepth;
        this.maxRequestLimit = maxRequestLimit;
    }

    public static void main(String[] args) throws Exception {
        int optimalThreads = Runtime.getRuntime().availableProcessors() * 2;
        MultiThreadCrawler crawler = new MultiThreadCrawler(optimalThreads, 8, 1000);

        long startTime = System.nanoTime();
        String result = crawler.find("Java_(programming_language)", "Cat", 30, TimeUnit.SECONDS);
        long duration = TimeUnit.SECONDS.convert(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);

        System.out.println("Took " + duration + " seconds, result: " + result);
    }


    public String find(String from, String target, long timeout, TimeUnit timeUnit) throws Exception {
        long deadline = System.nanoTime() + timeUnit.toNanos(timeout);
        ExecutorService executor = Executors.newFixedThreadPool(threadPoolSize);

        // Сброс состояния
        searchQueue.clear();
        visited.clear();
        requestCounter.set(0);
        found.set(false);
        resultNode = null;

        String normalizedFrom = normalize(from);
        String normalizedTarget = normalize(target);

        if (normalizedFrom.equals(normalizedTarget)) {
            return from;
        }

        // Инициализация
        searchQueue.offer(new Node(from, null, 0));
        visited.put(normalizedFrom, 0);

        try {
            // Запуск потоков
            for (int i = 0; i < threadPoolSize; i++) {
                executor.submit(new Worker(target, deadline));
            }

            countDownLatch.await();

            // Ожидание результата или таймаута
            while (System.nanoTime() < deadline && !found.get()) {
                Thread.sleep(10);
                if (resultNode != null || searchQueue.isEmpty()) break;
            }

            if (resultNode != null) {
                return buildPath(resultNode);
            }

            if (System.nanoTime() >= deadline) {
                throw new TimeoutException("Search timed out");
            }

            if (requestCounter.get() >= maxRequestLimit) {
                System.out.println("Stopped due to max request limit: " + maxRequestLimit);
            }

            return "not found";

        } finally {
            executor.shutdownNow();
        }
    }

    private String buildPath(Node node) {
        List<String> path = new ArrayList<>();
        while (node != null) {
            path.add(node.title);
            node = node.next;
        }
        Collections.reverse(path);
        return String.join(" > ", path);
    }

    private String normalize(String title) {
        return title.toLowerCase().trim();
    }

    private class Worker implements Runnable {
        private final String target;
        private final long deadline;

        Worker(String target, long deadline) {
            this.target = target;
            this.deadline = deadline;
        }

        @Override
        public void run() {
            log.info("Worker started");
            while (!found.get() && System.nanoTime() < deadline && !Thread.currentThread().isInterrupted()) {
                try {
                    // Таймаут на poll, чтобы не висеть вечно
                    Node node = searchQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (node == null) continue;

                    log.info("Processing node: {}", node.title);
                    // Проверка глубины
                    if (node.depth >= maxDepth) {
                        log.info("Depth limit reached");
                        continue;
                    }

                    // Получение ссылок
                    if (requestCounter.incrementAndGet() > maxRequestLimit) {
                        log.info("Request limit reached");
                        System.out.println("Thread " + Thread.currentThread().getName() + " hit request limit.");
                        return;
                    }

                    Set<String> links;
                    try {
                        links = client.getByTitle(node.title);
                        log.info("Fetched links for {}", node.title);
                    } catch (Exception e) {
                        System.err.println("Error fetching page: " + node.title + " - " + e.getMessage());
                        continue;
                    }

                    for (String link : links) {
                        String normLink = normalize(link);
                        log.info("Processing link: {}", normLink);
                        int currentDepth = node.depth + 1;

                        // Пропускаем уже посещённые (если не глубже)
                        Integer knownDepth = visited.get(normLink);
                        if (knownDepth != null && knownDepth <= currentDepth) {
                            continue;
                        }

                        // Обновляем глубину
                        if (visited.putIfAbsent(normLink, currentDepth) != null) {
                            log.info("Link already in queue: {}", normLink);
                            // Уже есть, но может быть глубже — обновим, если нужно
                            while (knownDepth == null || knownDepth > currentDepth) {
                                if (visited.replace(normLink, knownDepth, currentDepth)) break;
                                knownDepth = visited.get(normLink);
                            }
                        }

                        Node newNode = new Node(link, node, currentDepth);

                        // Проверка цели
                        if (target.equalsIgnoreCase(normLink)) {
                            if (found.compareAndSet(false, true)) {
                                log.info("Target found: {}", link);
                                resultNode = newNode;
                            }
                            countDownLatch.countDown();
                            return;
                        }

                        // Добавляем в очередь
                        searchQueue.offer(newNode);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    System.err.println("Worker error: " + e.getMessage());
                }
            }
        }
    }

    private static class Node {
        final String title;
        final Node next;
        final int depth;

        Node(String title, Node next, int depth) {
            this.title = title;
            this.next = next;
            this.depth = depth;
        }
    }

}