聚沙成塔--爬虫系列（十五）（如何，成为设计师）

2017-11-14
#爬虫系列

版权声明:本文为作者原创文章，可以随意转载，但必须在明确位置标明出处！！！

tips：本基础系列旨在以爬虫带大家入门Python语言

上一篇文章讲了多线程的基本使用及需要注意的地方，在Python中如果你要使用多线程，那么你的程序一定要是I/O密集型的这样才能发挥多线程的优势，如果你的程序设计是CPU计算密集型的那么使用多进程的设计。多进程的使用方式将在下一篇文章讲到。本篇文章的重点是讲怎么去设计多线程，设计是一门技术活啊，编码只是程序员掌握的最最基本的必要条件，随着你编码的时间越来越长经验越来越丰富，你对编码的理解也越来越深了，你再也不想只当个搬砖的了，你想要成为一个NB的设计师，你想让那些搬砖的人根据你的设计去干活。呵呵，骚年想要成为一个NB的设计师你要的路还很长。作为初学者来说作者建议目前设计模式你还不需要去深入研究，要成为一个设计师不是一朝一夕的事，是设计模式的灵活运用、是常年项目经验的累积、是框架设计的反复推敲。当你有了一定的编码经验那么你可以看看《大话设计模式》，再编码的时候就可以尝试运用里面的设计模式来编写你的代码。

包工头

就像王健林说的我们先定个小目标“先挣它一个亿”呵呵，当然这对他来说确实是个小目标，对我们这些普通人来说这个目标有点大啊，我们只能先定一个小小目标，我们的目标就是成为一个包工头，让你手底下的几十号人按照你的设计去做。一定要有当将军的觉悟，这样我们才能慢慢的向高手进发。

设计思想

我们知道Pyhton里的多线程对于I/O密集型的任务有显著的提高，那么就我们的爬虫程序想想我们应该怎么去设计多线程呢？要成为一个合格的设计师首先是要去了解需求，需求了解了才能指定计划。我们之前写的爬虫逻辑是怎么样的呢？打开一个网络请求–》解析网络请求返回回来的网页元素–》写入文档或数据库。这就是之前爬虫程序的整个流程，我们现在考虑一下引入多线程后应该怎么去编写它，两种想法

  1. 每个线程都去执行所有的逻辑，从打开一个网络请求到写入文档或数据库。
  2. 把每个步骤的逻辑分开，让线程各自去赋值单一逻辑的处理，这里我们把它设计为一个线程去处理网络请求，多个线程去处理解析操作，一个线程去处理写入文档或数据库操作，这个模式可以概括为一个输入、一个输出、多个工作线程。这种模式显然比第一种好，为什么这样设计呢？输入线程只负责去请求网络数据，将请求到的数据交给工作线程去处理，工作线程处理完后的数据交接输出线程做最后的输入工作，这样每个逻辑上都是独立的，专业的人做专业的事，这里还需要重点说一下为什么设计成一个输出，如果是多个输出不加锁的情况下会照成很多问题，比如将数据写入文档，A线程还没有把数据写完CPU时间片道理，那么B现在就开始写，这就会造成最终的结构是乱序的，I/O操作本来就是比较费时的，多线程的设计对它并没有多大性能的提升，二期使用多线程设计输出那么就必须的加锁保证数据同步，这也会增大资源开销，所以设计为一个输出是合理的
    流程图

Queue(队列)

如果我们设计成一个输入、多个工作、一个输出工作模式，那么我们就会面临一个问题，我们的数据怎么传递呢？特别是对于工作线程，因为它是多个线程那么肯定会存在资源竞争，所以我们必须得保证每个工作线程拿到的数据都是唯一的。这里就需要引入一个队列的概念了，这个模块是Python自带的，你可以直接使用它，像下面这样

1
import queue

queue模块提供了两种队列，一种是先入先出队列，一种是后入先出队列它们的定义如下

1
2
3
4
# 先入先出队列
Queue(maxsize=0)
# 后入先出队列
LifoQueue(maxsize=0)

maxsize为队列大小，默认为0，表示无限制队列，就是队列可以无限大只要你的内存够用，如果maxsize大于零，则表示限制队列，当队列满了会阻塞直到有空余的位置为止，就像饭店排队吃饭一样，店满了那么你只能等吃饭的人离开空出位置后你才能去。队列的工作模式是取一个消息就少一个消息。Queue对象提供了一下方法

  • qsize(): 获取队列大小
  • empty(): 判断队列是否为空
  • full(): 判断队列是否满了，当然这个函数需要你创建队列时maxsize大于零才有效
  • put(item, block=True, timeout=None): 放一个消息到队列里，block是否阻塞，如果队列满了此参数为True时则它会一直等，等到队列可以放消息为止， timeout超时，如果队列阻塞了等待多久
  • get(block=True, timeout=None): 从队列里获取消息

实战

okay，该讲的必要条件都已经讲过了，下面我们将爬虫程序改写成上面说的到设计模式，我们将输入线程命名为AccessTread、工作线程命名为WorkThreads、输出线程命名为OutThread，修改后结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#AccessThread.py
import threading
from urllib import request
from urllib import error
class AccessThread(threading.Thread):
    """docstring for AccessThread"""
    def __init__(self, work_queue, args):
        super(AccessThread, self).__init__()
        self.args = args
        self.work_queue = work_queue
    def run(self):
        self.request_url()
    def request_url(self):
        try:
            for page in range(1, 14):
                req = request.Request(self.args['url'] + str(page), headers=self.args['headers'])
                print(self.args['url'] + str(page))
                # 打开一个请求
                response = request.urlopen(req)
                # 读取服务器返回的页面数据内容
                content = response.read().decode('utf-8')
                self.work_queue.put(content)
        except error.URLError as e:
            print(e.reason)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# WorkThreads.py
import threading
import re, os
from urllib import request
class WorkThreads(threading.Thread):
    """docstring for WorkThreads"""
    def __init__(self, work_queue, out_queue):
        super(WorkThreads, self).__init__()
        self.work_queue = work_queue
        self.out_queue = out_queue
    def run(self):
        self.deal_work()
    def deal_work(self):
        while True:
            content = self.work_queue.get()
            if content:
               
                pattern = re.compile(r'<div class="article block untagged mb15[\s\S]*?class="stats-vote".*?</div>', re.S)
                userinfos = re.findall(pattern, content)
                
                if userinfos:
                    pattern = re.compile(r'<a href="(.*?)".*?<h2>(.*?)</h2>.*?<div class="content">(.*?)</div>.*?<i class="number">(.*?)</i>', re.S)
                    picture = re.compile(r'<div class="thumb">.*?src="(.*?)"', re.S)
                    for userinfo in userinfos:
                        item = re.findall(pattern, userinfo)
                        pictures = re.findall(picture, userinfo)
                        try:
                            if item:
                                infos = []
                                userid, name, content, num = item[0]
                                # 去掉换行符，<span></span>，<br/>符号
                                userid = re.sub(r'\n|<span>|</span>|<br/>', '', userid)
                                name = re.sub(r'\n|<span>|</span>|<br/>', '', name)
                                content = re.sub(r'\n|<span>|</span>|<br/>|\x01', '', content)
                                
                                if pictures:
                                    path = './users/'
                                    if not os.path.exists(path):
                                        os.makedirs(path)
                                    request.urlretrieve('http:' + pictures[0], path + os.path.basename(pictures[0]))
                                    infos.append((userid, name, int(num), content, pictures[0]))
                                    self.out_queue.put((userid, name, int(num), content, pictures[0]))
                                else:
                                    infos.append((userid, name, int(num), content, ' '))
                                    self.out_queue.put((userid, name, int(num), content, ' '))
                                   
                        except Exception as e:
                            print(e)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# OutThread.py
import threading
from .database.MySQLImpl import MySQLImpl
from .database.SqliteImpl import SqliteImpl
from .database.SqlalchemyImpl import SqlalchemyImpl
class OutThread(threading.Thread):
    """docstring for OutThread"""
    def __init__(self, out_queue):
        super(OutThread, self).__init__()
        self.out_queue = out_queue
        self.sqlite = SqliteImpl()
    def run(self):
        self.out_work()
    def out_work(self):
        while True:
            msg = self.out_queue.get()
            #print(msg)
            if msg:
                self.sqlite.insert_record(msg)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# Scheduler.py
from .AccessThread import AccessThread
from .WorkThreads import WorkThreads
from .OutThread import OutThread
import queue
class Scheduler(object):
    def __init__(self):
        self.work_queue = queue.Queue()
        self.out_queue = queue.Queue()
        
    def handle(self):
        threads = []
        dict_info = {}
        dict_info['url'] = 'https://www.qiushibaike.com/8hr/page/'
        dict_info['headers'] = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36 MicroMessenger/6.5.2.501 NetType/WIFI WindowsWechat QBCore/3.43.691.400 QQBrowser/9.0.2524.400'}
        acc_thread = AccessThread(self.work_queue, dict_info)
        
        for _ in range(10):
            work_thread = WorkThreads(self.work_queue, self.out_queue)
            threads.append(work_thread)
        
        out_thread = OutThread(self.out_queue)
        threads.append(acc_thread)
        threads.append(out_thread)
        for t in threads:
            t.daemon = True
            t.start()
        while True:
            alive = False
            for t in threads:
                alive = alive or t.is_alive()
            if not alive:
                break

完整的代码放在我的git上，地址https://github.com/Gavinxyj/Python/tree/master/python_study/Scrapy/modules欢迎大家fork、star。

okay，本篇对多线程设计的介绍就到此结束了，读者一定要亲自上机验证才回有收获，一定不要有你觉得很简单它就真的简单了。实践是验证理论唯一的途径。

欢迎关注我的公众号:「爱做饭的老谢」，老谢一直在努力…

上一篇：聚沙成塔–爬虫系列（十四）（群架要怎么打）