目的

为了提高爬虫效率，实现多线程同步运行，比如一秒同时爬取50个连接。主要用到的是python自带的多线程库threading # 实现 首先引用所需要的库 

1
2
import threading
from threading import Thread

Notice

需要注意的是两个参数

  1. locks = threading.Lock()#线程锁主要用来多线程对单一资源进行写入的时候，能锁住所有线程，避免多个线程同时对一个资源进行读写操作
  2. threadmax = threading.BoundedSemaphore(50)#总线程数设置一下这个爬虫系统最大的线程数，避免资源过度消耗，或者线程无限增加导致被爬取的服务器出现宕机等情况

主要代码

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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class Duoxiancheng():
    def __init__(self):
        self.start_time = time.time()
        self.num = 0
        self.result=[]
        self.datas=[]
    def writedata(self):
        mu.acquire(True) #在多线程写入同一个资源时开启线程锁
        self.insert_mysql(self.result)#插入数据库
        self.writefile(self.result)#写入本地文件
        self.result=[]
        mu.release()#写完文件后释放线程锁，别的线程将继续工作
    def insert_mysql(self,items):
        try:
            db = pymysql.connect(host='localhost', port=3306, user='root', passwd='admin123456',
                                db='Spider', charset='utf8')
            cursor = db.cursor()
            cursor.execute('use Spider;')
            cursor.execute('CREATE TABLE IF NOT EXISTS flight('
                            'id INT PRIMARY KEY NOT NULL AUTO_INCREMENT,'
                            'airlineName VARCHAR(30),'
                            'flightNumber VARCHAR(30),'
                            'craftTypeName VARCHAR(30),'
                            'dcity VARCHAR(30),'
                            'acity VARCHAR(30),'
                            'dealtime VARCHAR(30),'
                            'arrivalDate VARCHAR(30),'
                            'price VARCHAR(30),'
                            'datetime VARCHAR(30));')
            try:
                for item in items:
                    nowtime=[str((datetime.now()).strftime('%Y-%m-%d %H:%M'))]
                    sql_query = 'INSERT INTO flight(airlineName, flightNumber, craftTypeName, dcity,acity,dealtime,arrivalDate,price,datetime) VALUES ("%s", "%s", "%s", "%s","%s", "%s", "%s", "%s", "%s")' % (
                                item[0], item[1],item[2],item[3], item[4],  item[5],item[6], item[7],nowtime,)
                    cursor.execute(sql_query)
                db.commit()
            except pymysql.MySQLError as e:
                db.rollback()
                print(e.args)
        except pymysql.MySQLError as e:
            print(e.args)
            return
    def writefile(self,data):
        这个函数用于将文件保存至本地文件，可以是excel，csv等
        ...
    def prepare_data(self):
        这里将所要需要进行多线程的数据准备好，可以放在列表中然后将数据传入thread1函数开始创建线程
        ...
        self.thread1(data)
    def task(self,*args,**kwarg):
        这里写需要爬取数据的相关代码
        ...
        threadmax1.release()#最后记得release释放线程
    def thread1(self,data):
            """多线程"""
            i=0   
            starlist=[]   
            while data:
                dflist=data.pop(0)
                if i%50==0:
                    print('\r正在运行的线程数{}\n已经爬取 {}/{} 个'.format(str(threading.active_count()),str(i),str(long)),flush=True)#可以打印正在运行的线程数，实现监控
                if dflist:
                    time.sleep(0.1)             
                    threadmax1.acquire()#从空闲的线程中取一个线程
                    data1=dflist[0]
                    data2=dflist[1]
                    data3=dflist[2]
                    t = Thread(target=self.task, args=(data1,data2,data3,))#创建一个运行task的线程，传入参数             
                    i+=1
                    starlist.append(t)
                    time.sleep(0.01)
                    t.start()  #线程开始运行 
                    self.num +=1
                    if self.ques%1000==0: #这个是用于爬取1000次以后写入一次数据
                        t2=Thread(target=self.writedata)
                        t2.start()
                    t.start()
            for z in starlist:
                z.join()#设定守护线程，主线程会等所有子线程都运行完毕后才结束，保证所有线程的数据最后都会由主线程的最后一步写入数据   
            self.writedata()
    def __del__(self):
        end_time = time.time()
        print('一共花费时间:{}(单位秒)'.format(end_time - self.start_time))

通常多线程还会涉及到代理ip池的运用

相较于免费的代理ip，付费的会更加稳定，为了避免打广告，我这里不推荐哪一个公司的产品，感兴趣可以私信我，下面提供一个较为通用的通过付费ip池构建一个项目的代理ip池的方法 

1
2
3
4
5
6
def get_ip_from():
    global ip_list
    print('获取ip')
    proxyurl='你自己申请的代理ip api'
    prox=requests.get(proxyurl).text.split('\r\n')
    ip_list=prox
此时ip_list 就存好了申请一次的ip，最便宜的那一档往往是一次几百个，有效期1到5分钟，因此需要一个线程每4分钟运行一次，更新这个ip池 
1
2
3
4
5
from threading import Timer
def time1():
    t1=4*60
    t = Timer(t1, get_ip_from)
    t.start()
同时获取ip的方法也要加上一个延时的参数，在调用一次更新ip池后再次等待4分钟进行下一次的ip获取 
1
2
3
4
5
6
7
8
def get_ip_from():
    global ip_list,task_continue
    print('获取ip')
    proxyurl='你自己申请的代理ip api'
    prox=requests.get(proxyurl).text.split('\r\n')
    ip_list=prox
    if task_continue:
        time1()
在实现网络访问的函数中加入 
1
2
3
4
5
6
prox=ipp_list[random.randint(0, len(ip_list))]
proxies = {
            'http':'http://' +prox ,
            'https': 'https://'+prox,
            }
response=requests.get('url',headers=headers,cookies=cookies,proxies=proxies)
实现通过代理ip访问,当然加入try能保证每一线程都能使用到有效ip进行获取数据 
1
2
3
4
5
6
7
8
9
10
while True:
    try:
        result=requests.get('url',headers=headers,cookies=cookies,proxies=proxies)
    except:
        prox=ip_list[random.randint(0, len(ipp_list))]
        print('更换ip')
        proxies = {
                    'http':'http://' +prox ,
                    'https': 'https://'+prox,
                    }
以上就是关于爬虫所能用到的一些多线程的使用方法了。