python - 网络爬虫的线程建议 - 使用单个列表进行调度

Question

我想在我的网络爬虫中添加多线程，但我可以看到蜘蛛计划要抓取的链接的方式可能与多线程不兼容。爬虫只会在少数新闻网站上处于事件状态，但我更愿意在同一域上打开多个线程，而不是为每个域启动一个新线程。我的网页爬取代码是通过以下函数操作的:

def crawl_links():
    links_to_crawl.append(domain[0])
    while len(links_to_crawl) > 0:
        link = links_to_crawl[0]
        if link in crawled_links or link in ignored_links:
            del links_to_crawl[0]
        else:
            print '\n', link
            try:
                html = get_html(link)
                GetLinks(html)
                SaveFile(html)
                crawled_links.append(links_to_crawl.pop(0))
            except (ValueError, urllib2.URLError, Timeout.Timeout, httplib.IncompleteRead):
                ignored_links.append(link_to_crawl.pop(0))
    print 'Spider finished!'
    print 'Ignored links:\n', ignored_links
    print 'Crawled links:\n', crawled_links
    print 'Relative links\n', relative_links

如果我对线程如何工作的理解是正确的，如果我只是在此过程中打开多个线程，它们将全部抓取相同的链接(可能多次)或者它们会有点冲突。无需详细说明，您建议如何重组调度以使其与同时运行的多个线程兼容？

我已经考虑过这个问题，我能想出的唯一解决方法是让 GetLinks() 类将链接附加到多个列表，每个线程有一个单独的列表……但是这个这似乎是一个相当笨拙的解决方法。

Answer 1

这是我用来在 Python 中运行多线程应用程序的通用方案。

该方案采用输入参数表，并为每一行并行执行一个线程。

每个线程占用一行，并为行中的每个项目顺序执行一个线程。

每个项目都包含固定数量的参数，这些参数将传递给执行的线程。

输入示例:

table = \
[
    [[12,32,34],[11,20,14],[33,67,56],[10,20,45]],
    [[21,21,67],[44,34,74],[23,12,54],[31,23,13]],
    [[31,67,56],[34,22,67],[87,74,52],[87,74,52]],
]

在这个例子中，我们将有 3 个线程并行运行，每个线程依次执行 4 个线程。

为了保持线程平衡，建议每行中的项目数相同。

线程方案:

import threading
import MyClass # This is for you to implement

def RunThreads(outFileName,errFileName):
    # Create a shared object for saving the output of different threads
    outFile = CriticalSection(outFileName)
    # Create a shared object for saving the errors of different threads
    errFile = CriticalSection(errFileName)
    # Run in parallel one thread for each row in the input table
    RunParallelThreads(outFile,errFile)

def RunParallelThreads(outFile,errFile):
    # Create all the parallel threads
    threads = [threading.Thread(target=RunSequentialThreads,args=(outFile,errFile,row)) for row in table]
    # Start all the parallel threads
    for thread in threads: thread.start()
    # Wait for all the parallel threads to complete
    for thread in threads: thread.join()

def RunSequentialThreads(outFile,errFile,row):
    myObject = MyClass()
    for item in row:
        # Create a thread with the arguments given in the current item
        thread = threading.Thread(target=myObject.Run,args=(outFile,errFile,item[0],item[1],item[2]))
        # Start the thread
        thread.start()
        # Wait for the thread to complete, but only up to 600 seconds
        thread.join(600)
        # Terminate the thread if it hasn't completed up to this point
        if thread.isAlive():
            thread._Thread__stop()
            errFile.write('Timeout on arguments: '+item[0]+' '+item[1]+' '+item[2]+'\n')

下面的类实现了一个可以在并行运行的不同线程之间安全共享的对象。它提供了一个名为 write 的单一接口(interface)方法，它允许任何线程以安全的方式更新共享对象(即，操作系统不会在此过程中切换到另一个线程)。

import codecs

class CriticalSection:
    def __init__(self,fileName):
        self.mutex = threading.Lock()
        self.fileDesc = codecs.open(fileName,mode='w',encoding='utf-8')
    def __del__(self):
        del self.mutex
        self.fileDesc.close()
    def write(self,data):
        self.mutex.acquire()
        self.fileDesc.write(data)
        self.mutex.release()

上述方案应该允许您控制应用程序中的“并行性”级别和“顺序性”级别。

例如，您可以对所有项目使用一行，并让您的应用程序以完全顺序的方式运行。

相比之下，您可以将每个项目放在单独的行中，并让您的应用程序以完全并行的方式运行。

当然，您可以选择以上任意组合...

注意:

在 MyClass 中，您需要实现方法 Run，它将获取 outFile 和 errFile 对象，以及您为每个线程定义的参数。