7

我正在尝试编写一个多线程网络爬虫。

我的主要入口类具有以下代码：

ExecutorService exec = Executors.newFixedThreadPool(numberOfCrawlers);
while(true){
    URL url = frontier.get();
    if(url == null)
         return;
exec.execute(new URLCrawler(this, url));
}

URLCrawler 获取指定的 URL，解析 HTML，从中提取链接，并将不可见的链接安排回前沿。

边界是未抓取的 URL 队列。问题是如何编写 get() 方法。如果队列为空，它应该等到任何 URLCrawler 完成，然后重试。仅当队列为空且当前没有活动的 URLCrawler 时，它才应返回 null。

我的第一个想法是使用 AtomicInteger 来计算当前工作 URLCrawler 的数量，并使用辅助对象进行 notifyAll()/wait() 调用。每个爬虫在开始时增加当前工作的 URLCrawler 的数量，在退出时减少它，并通知对象它已经完成。

但我读到 notify()/notifyAll() 和 wait() 是一些不推荐使用的线程通信方法。

我应该在这种工​​作模式中使用什么？它类似于M个生产者和N个消费者，问题是如何处理生产者的消耗。

4

6 回答 6

3

我不确定我是否理解你的设计，但这可能是一份工作Semaphore

于 2010-08-04T05:50:53.897 回答
3

一种选择是使“边界”成为阻塞队列，因此任何试图从中“获取”的线程都会阻塞。只要任何其他 URLCrawler 将对象放入该队列，任何其他线程都会自动收到通知（对象出队）

于 2010-08-04T05:52:00.977 回答
2

我认为在这种情况下使用等待/通知是合理的。想不出任何直接的方法来使用 juc
在一个类中，让我们调用 Coordinator：

private final int numOfCrawlers;
private int waiting;

public boolean shouldTryAgain(){
    synchronized(this){
        waiting++;
        if(waiting>=numOfCrawlers){
            //Everybody is waiting, terminate
            return false;
        }else{
            wait();//spurious wake up is okay
            //waked up for whatever reason. Try again
            waiting--;
            return true;
        }
    }

public void hasEnqueued(){
    synchronized(this){
        notifyAll();
    }
}

然后，

ExecutorService exec = Executors.newFixedThreadPool(numberOfCrawlers);
while(true){
    URL url = frontier.get();
    if(url == null){
        if(!coordinator.shouldTryAgain()){
            //all threads are waiting. No possibility of new jobs.
            return;
        }else{
            //Possible that there are other jobs. Try again
            continue;
        }
    }
    exec.execute(new URLCrawler(this, url));
}//while(true)
于 2010-08-04T06:49:27.157 回答
2

我认为您的用例的基本构建块是“锁存器”，类似于 CountDownLatch，但与 CountDownLatch 不同，它也允许增加计数。

这种闩锁的接口可能是

public interface Latch {
    public void countDown();
    public void countUp();
    public void await() throws InterruptedException;
    public int getCount();
}

计数的合法值为 0 及以上。await() 方法会让你阻塞直到计数降到零。

如果你有这样的闩锁，你的用例可以很容易地描述。我还怀疑可以在此解决方案中消除队列（边界）（执行器无论如何都提供了一个，因此有点多余）。我会将您的主要例程重写为

ExecutorService executor = Executors.newFixedThreadPool(numberOfCrawlers);
Latch latch = ...; // instantiate a latch
URL[] initialUrls = ...;
for (URL url: initialUrls) {
    executor.execute(new URLCrawler(this, url, latch));
}
// now wait for all crawling tasks to finish
latch.await();

您的 URLCrawler 将以这种方式使用闩锁：

public class URLCrawler implements Runnable {
    private final Latch latch;

    public URLCrawler(..., Latch l) {
        ...
        latch = l;
        latch.countUp(); // increment the count as early as possible
    }

    public void run() {
        try {
            List<URL> secondaryUrls = crawl();
            for (URL url: secondaryUrls) {
                // submit new tasks directly
                executor.execute(new URLCrawler(..., latch));
            }
        } finally {
            // as a last step, decrement the count
            latch.countDown();
        }
    }
}

至于闩锁实现，可能有多种实现，从基于 wait() 和 notifyAll() 的实现，一种使用 Lock 和 Condition 的实现，到使用 AbstractQueuedSynchronizer 的实现。我认为所有这些实现都非常简单。请注意，wait()-notifyAll() 版本和 Lock-Condition 版本将基于互斥，而 AQS 版本将使用 CAS（比较和交换），因此在某些情况下可能会更好地扩展。

于 2010-08-05T00:32:38.490 回答
2

这个问题有点老了，但我想我找到了一些简单有效的解决方案：

像下面这样扩展 ThreadPoolExecutor 类。新功能是保持活动任务计数（不幸的是，提供getActiveCount()的是不可靠的）。如果taskCount.get() == 0并且没有更多排队的任务，则意味着没有什么可做并且执行器关闭。你有你的退出标准。此外，如果您创建了执行程序，但未能提交任何任务，它不会阻塞：

public class CrawlingThreadPoolExecutor extends ThreadPoolExecutor {

    private final AtomicInteger taskCount = new AtomicInteger();

    public CrawlingThreadPoolExecutor() {
        super(8, 8, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {

        super.beforeExecute(t, r);
        taskCount.incrementAndGet();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {

        super.afterExecute(r, t);
        taskCount.decrementAndGet();
        if (getQueue().isEmpty() && taskCount.get() == 0) {
            shutdown();
        }
    }
}

您还需要做的另一件事是以Runnable保持参考Executor您正在使用的方式实现您的，以便能够提交新任务。这是一个模拟：

public class MockFetcher implements Runnable {

    private final String url;
    private final Executor e;

    public MockFetcher(final Executor e, final String url) {
        this.e = e;
        this.url = url;
    }

    @Override
    public void run() {
        final List<String> newUrls = new ArrayList<>();
        // Parse doc and build url list, and then:
        for (final String newUrl : newUrls) {
            e.execute(new MockFetcher(this.e, newUrl));
        }
    }
}
于 2012-12-07T09:13:25.863 回答
0

我想推荐一个 AdaptiveExecuter。根据特征值，您可以选择序列化或并行化线程以执行。在下面的示例中，PUID 是我想用来做出该决定的字符串/对象。您可以更改逻辑以适合您的代码。代码的某些部分被注释以允许进一步的实验。

类 AdaptiveExecutor 实现 Executor { final Queue tasks = new LinkedBlockingQueue(); 可运行活动；//ExecutorService threadExecutor=Executors.newCachedThreadPool(); static ExecutorService threadExecutor=Executors.newFixedThreadPool(4);

AdaptiveExecutor() {
    System.out.println("Initial Queue Size=" + tasks.size());
}

public void execute(final Runnable r) {
    /* if immediate start is needed do either of below two
    new Thread(r).start();

    try {
        threadExecutor.execute(r);
    } catch(RejectedExecutionException rEE ) {
        System.out.println("Thread Rejected " + new Thread(r).getName());
    }

    */


    tasks.offer(r); // otherwise, queue them up
    scheduleNext(new Thread(r)); // and kick next thread either serial or parallel.
    /*
    tasks.offer(new Runnable() {
        public void run() {
            try {
                r.run();
            } finally {
                scheduleNext();
            }
        }
    });
    */
    if ((active == null)&& !tasks.isEmpty()) {
        active = tasks.poll();
        try {
            threadExecutor.submit(active);
        } catch (RejectedExecutionException rEE) {
            System.out.println("Thread Rejected " + new Thread(r).getName());
        }
    }

    /*
    if ((active == null)&& !tasks.isEmpty()) {
        scheduleNext();
    } else tasks.offer(r);
    */
    //tasks.offer(r);

    //System.out.println("Queue Size=" + tasks.size());

}

private void serialize(Thread th) {
    try {
        Thread activeThread = new Thread(active);

        th.wait(200);
        threadExecutor.submit(th);
    } catch (InterruptedException iEx) {

    }
    /*
    active=tasks.poll();
    System.out.println("active thread is " +  active.toString() );
    threadExecutor.execute(active);
    */
}

private void parallalize() {
    if(null!=active)
        threadExecutor.submit(active);
}

protected void scheduleNext(Thread r) {
    //System.out.println("scheduleNext called") ;
    if(false==compareKeys(r,new Thread(active)))
        parallalize();
    else serialize(r);
}

private boolean compareKeys(Thread r, Thread active) {
    // TODO: obtain names of threads. If they contain same PUID, serialize them.
    if(null==active)
        return true; // first thread should be serialized
    else return false;  //rest all go parallel, unless logic controlls it
}

}

于 2011-02-28T21:35:28.583 回答