记录学习python的日常 分布式爬虫

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了记录学习python的日常 分布式爬虫，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含3822字，纯文字阅读大概需要6分钟。

内容图文

首先创建分布式管理：管理内得到的队列 注册到网络上 然后暴露网络

url管理进程 针对每一个url进行分析 操作 url_manager_proc()

判断url类中是否有新的链接 有的话塞入url_q队列中

判断在conn_q队列中有没有新的内容 有的话 塞入url类中

result_solve_proc 主要是从爬虫调度器中出来的数据 爬虫调度器每一条url中爬出来的信息 依次塞入 result_q队列

有关新url 的 放入 conn_q队列

数据则放入数据储存队列 store_q

store_proc 主要是处理store_q 队列中的内容 依次存储数据库或本地

from multiprocessing.managers import BaseManager
import time

from multiprocessing import Process, Queue

from DataOutputNew import DataOutputNew
from URLManagerNew import UrlManagerNew

# 首先创建分布式管理：管理内得到的队列 注册到网络上 然后暴露网络

# url管理进程  针对每一个url进行分析 操作  url_manager_proc()
# 判断url类中是否有新的链接 有的话塞入url_q队列中
# 判断在conn_q队列中有没有新的内容 有的话 塞入url类中

# result_solve_proc  主要是从爬虫调度器中出来的数据  爬虫调度器每一条url中爬出来的信息 依次塞入 result_q队列
# 有关新url 的 放入 conn_q队列
# 数据则放入数据储存队列 store_q

# store_proc  主要是处理store_q 队列中的内容 依次存储数据库或本地

class NodeManage(object):
    # 创造分布式管理
    def star_Manager(self,url_q,result_q):
        # 把创建的两个队列注册在网络上，利用register方法，callable参数关联了Queue对象，
        # 将Queue对象在网络中暴露
        BaseManager.register('get_task_queue', callable=lambda: url_q)
        BaseManager.register('get_result_queue', callable=lambda: result_q)
        manager = BaseManager(address=('',8001),authkey=b'ceshi')
        return manager

    # URL管理进程将从conn_
    # q队列获取到的新URL提交给URL管理器，经过去重之后，取出url放到url_queue 给爬虫调度
    # conn_q 新的url
    def url_manager_proc(self,url_q,conn_q,root_url):
        url_manager = UrlManagerNew()
        url_manager.add_new_url(root_url)
        while True:
            while(url_manager.has_new_url()):
                #从URL管理器获取新的url
                new_url = url_manager.get_new_url()
                #将新的URL发给工作节点
                url_q.put(new_url)
                print ('old_url=',url_manager.old_url_size())
                #加一个判断条件，当爬去2000个链接后就关闭,并保存进度
                if(url_manager.old_url_size()>100):
                    #通知爬行节点工作结束
                    url_q.put('end')
                    print ('控制节点发起结束通知!')
                    #关闭管理节点，同时存储set状态
                    url_manager.save_progress('new_urls.txt',url_manager.new_urls)
                    url_manager.save_progress('old_urls.txt',url_manager.old_urls)
                    return
            #将从result_solve_proc获取到的urls添加到URL管理器之间
            try:
                if not conn_q.empty():
                    urls = conn_q.get()
                    url_manager.add_new_urls(urls)
            except :
                time.sleep(0.1)#延时休息

    # 数据提取进程从result_queue
    # 队列读取返回的数据，并将数据中的URL添加到conn_q.
    # 队列交给URL管理进程, 将数据中的文章标题和摘要添加到store_q队列交给数据存储进程。
    # store_q存储进程队列
    def result_solve_proc(self,result_q,conn_q,store_q):
        while(True):
            try:
                if not result_q.empty():
                    content = result_q.get(True)
                    if content['new_urls']=='end':
                        #结果分析进程接受通知然后结束
                        print ('结果分析进程接受通知然后结束!')
                        store_q.put('end')
                        return
                    conn_q.put(content['new_urls'])#url为set类型
                    store_q.put(content['data'])#解析出来的数据为dict类型
                else:
                    time.sleep(0.1)#延时休息
            except :
                time.sleep(0.1)#延时休息


    def store_proc(self,store_q):
        output = DataOutputNew()
        while True:
            if not store_q.empty():
                data = store_q.get()
                if data=='end':
                    print ('存储进程接受通知然后结束!')
                    output.ouput_end(output.filepath)
                    return
                output.store_data(data)
            else:
                time.sleep(0.1)
        pass


if __name__=='__main__':
    #初始化4个队列

    url_q = Queue()
    result_q = Queue()
    store_q = Queue()
    conn_q = Queue()
    # 创建分布式应用
    node = NodeManage()
    manager = node.star_Manager(url_q,result_q)
    #创建URL管理进程、 数据提取进程和数据存储进程
    url_manager_proc = Process(target=node.url_manager_proc, args=(url_q,conn_q,'http://baike.baidu.com/item/网络爬虫',))
    result_solve_proc = Process(target=node.result_solve_proc, args=(result_q,conn_q,store_q,))
    store_proc = Process(target=node.store_proc, args=(store_q,))
    #启动3个进程和分布式管理器
    url_manager_proc.start()
    result_solve_proc.start()
    store_proc.start()
    manager.get_server().serve_forever()

内容总结

以上是互联网集市为您收集整理的记录学习python的日常 分布式爬虫全部内容，希望文章能够帮你解决记录学习python的日常 分布式爬虫所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---