python - 当多个 worker 完成未知大小的工作时如何得到通知？

Question

我正在尝试使用 django-rq 在 Python 中创建网站爬虫.到目前为止，我的 worker 看起来像这样:

1. 从队列中获取下一页。
2. 在数据库中创建一个页面记录。设置 status=1。
3. 下载页面内容和流程。最多可能需要一分钟左右的时间。
4. 对于页面中的每个链接
   1. 检查链接是否已在数据库中注册。
   2. 如果没有，创建一个新的页面记录。设置 status=0 并将链接添加到队列。
5. for循环结束后，检查status=0的页数是否为0，如果是，则作业完成。

status=1 表示页面已处理。 status=0 表示页面尚未处理。

现在，该算法仅适用于单个工作人员。然而，当有更多的 worker 时它不会，因为工作程序的结束有时会比它应该的更早触发。

实现此 worker 的正确方法是什么？

Answer 1

1。调整您当前的系统

• 如果您在处理完页面之后才将页面状态设置为 1，那么您的工作人员不应提前宣布“工作完成”。
• 您的第 2 步仅对于您开始抓取的第一个页面是必需的。

所以你的系统会是这样的:

start job:
1. Create a page record in the database. Set status=0. Add page to queue.

worker:
1. Get the next page from the queue.
2. Download the page contents and process. Might take up to a minute or so.
3. For each link in the page
    1. Check if the link is already registered in the database.
    2. If not, create a new page record. Set status=0 and add the link to the queue.
4. After the for loop ends, set status=1 for this page.
5. Check whether the count of pages with status=0 is 0. If yes, the job is done.

存在的问题是，如果在前一个网络抓取作业完成之前开始后续网络抓取作业，您将只会在最后一个网络抓取作业结束时获得“作业完成”。您也许可以将作业 ID 添加到数据库页面记录中，并将“完成的作业”重新定义为 count(status=0 and job-id=x) = 0

2。利用 RQ 的作业类

来自RQ docs :

When jobs get enqueued, the queue.enqueue() method returns a Job instance. ... it has a convenience result accessor property, that will return None when the job is not yet finished, or a non-None value when the job has finished (assuming the job has a return value in the first place, of course).

您可以对两种不同类型的作业进行排队，一种是“获取网页”，另一种用于管理抓取过程。

管理工作将启动并跟踪所有“获取网页”工作。它会知道什么时候“工作完成”，因为它的所有子工作都已完成。

您不一定需要向数据库写入任何内容来管理抓取过程。

您需要运行 2 个以上的工作程序，以便可以同时处理 crawl 和 fetch，也许在不同的队列上。

def something_web_facing():
    ...
    queue.enqueue(crawl, 'http://url.com/start_point.html')
    ...

def crawl(start_url):
    fetch_jobs = []
    seen_urls = set()

    seen_urls.add(start_url)
    fetch_jobs.append( queue.enqueue(fetch, start_url) )

    while len(fetch_jobs) > 0:

        # loop over a copy of fetch_jobs
        for job in list(fetch_jobs):

            # has this job completed yet?
            if job.result:

                # a fetch job returns a list of the next urls to crawl
                for url in job.result:

                    # fetch this url if we haven't seen it before
                    if url not in seen_urls:
                        seen_urls.add(url)
                        fetch_jobs.append( queue.enqueue(fetch, url) )

                fetch_jobs.remove(job)

        time.sleep(1)

    return "Job done!"

def fetch(url):
    """Get web page from url, return a list of links to follow next"""

    html_page = download_web_page(url)
    links_to_follow = find_links_to_follow(html_page)
    return links_to_follow

3。使用别人的网络爬虫代码

Scrapy

您可以排队使用 scrapy 的作业。 Run scrapy from a script