如何更简单的使用线程，来进行数据采集或者图片下载

Stubborn

简单的来说，提高代码的复用性吧。可能写的不是很好，有大神的话，欢迎指点交流。

这里说两个简单的Demo。在复用的时候，只要继承TaskThreading，继而重写parse函数，作为爬虫“入口”，即可进行多线程采集

一，只开多个线程。不进行任务类型区分。

1. from queue import Queue
   
2. import threading
   
3. import requests
   
4. 
   
5. 
   
6. class ThreadPoolManagement(object):
   
7.     """线程池管理"""
   
8. 
   
9.     def __init__(self, task_thread, task_queue, number=5):
   
10.         """
    
11.         :param task_thread: 任务线程
    
12.         :param task_queue: 任务队列
    
13.         :param number: 线程池数量
    
14.         """
    
15.         self.number = number
    
16.         self.thread_list = []
    
17.         self.__create_threading__(task_thread, task_queue)
    
18. 
    
19.     def __create_threading__(self, task_thread, task_queue):
    
20.         for i in range(self.number):
    
21.             self.thread_list.append(task_thread(task_queue))
    
22. 
    
23.     def start_threading(self) -> None:
    
24.         for i in self.thread_list:
    
25.             i.start()
    
26. 
    
27.     def join_threading(self) -> None:
    
28.         for i in self.thread_list:
    
29.             i.join()
    
30. 
    
31. 
    
32. class TaskThreading(threading.Thread):
    
33.     """任务线程"""
    
34. 
    
35.     def __init__(self, task_queue: Queue):
    
36.         super().__init__()
    
37.         self.task = task_queue
    
38. 
    
39.     def run(self) -> None:
    
40.         print("子线程（-%s-）启动" % (threading.current_thread().name))
    
41.         while True:
    
42.             if self.task.empty():
    
43.                 break
    
44.             else:
    
45.                 url = self.task.get()
    
46.                 self.parse(url)
    
47.         print("子线程（-%s-）结束" % (threading.current_thread().name))
    
48. 
    
49.     def parse(self, url) -> None:
    
50.         pass
    
51. 
    
52. class TestMode(TaskThreading):
    
53. 
    
54.     def parse(self, url) -> None:
    
55.         """在这里重写你的爬虫代码"""
    
56.         response = requests.get(url=url)
    
57.         print(f"{threading.current_thread().name}正在模拟等待{url}链接响应，时长5秒")
    
58.         print(f"{threading.current_thread().name}响应成功，状态{response}")
    
59. 
    
60. def main():
    
61.     # 任何进程默认就会启动一个线程，称为主线程，主线程可以启动新的子线程
    
62.     # current_thread()返回当前线程的实例
    
63.     print("主线程（-%s-）启动" % (threading.current_thread().name))
    
64. 
    
65.     import queue
    
66.     # 建好队列，给定爬虫任务
    
67.     get_url_queue = queue.Queue()
    
68.     for page in range(5):
    
69.         get_url_queue.put(f"http://www.httpbin.org")
    
70. 
    
71.     # 初始化线程池，数量为3
    
72.     threading_pool = ThreadPoolManagement(
    
73.         task_thread=TestMode,
    
74.         task_queue=get_url_queue,
    
75.         number=3
    
76.     )
    
77. 
    
78.     # 启动线程池的线程
    
79.     threading_pool.start_threading()
    
80. 
    
81.     # 检测第一个爬虫队列任务是否为空
    
82.     while not get_url_queue.empty():
    
83.         pass
    
84. 
    
85.     # 关闭第一个爬虫线程池的线程
    
86.     threading_pool.join_threading()
    
87. 
    
88.     print("主线程（-%s-）结束" % (threading.current_thread().name))
    
89. 
    
90. 
    
91. if __name__ == '__main__':
    
92.     main()

复制代码

二，开多个线程。进行任务类型区分（生产消费者模型线程）。

1. # -*- coding: utf-8 -*-
   
2. # !/usr/bin/python3
   
3. """
   
4. Created on 2019:10-13-->16:19:36
   
5. 
   
6. @author: 1263270345@qq.com / Alex
   
7. 
   
8. """
   
9. from queue import Queue
   
10. import threading
    
11. import requests
    
12. 
    
13. stop_flag = False
    
14. data_queue = Queue()
    
15. 
    
16. class ThreadPoolManagement(object):
    
17.     """线程池管理"""
    
18. 
    
19.     def __init__(self, task_thread, task_queue, number=5):
    
20.         """
    
21.         :param task_thread: 任务线程
    
22.         :param task_queue: 任务队列
    
23.         :param number: 线程池数量
    
24.         """
    
25.         self.number = number
    
26.         self.thread_list = []
    
27.         self.__create_threading__(task_thread, task_queue)
    
28. 
    
29.     def __create_threading__(self, task_thread, task_queue):
    
30.         for i in range(self.number):
    
31.             self.thread_list.append(task_thread(task_queue))
    
32. 
    
33.     def start_threading(self) -> None:
    
34.         for i in self.thread_list:
    
35.             i.start()
    
36. 
    
37.     def join_threading(self) -> None:
    
38.         for i in self.thread_list:
    
39.             i.join()
    
40. 
    
41. 
    
42. class TaskThreading(threading.Thread):
    
43.     """任务线程"""
    
44. 
    
45.     def __init__(self, task_queue: Queue):
    
46.         super().__init__()
    
47.         self.task = task_queue
    
48. 
    
49.     def run(self) -> None:
    
50.         print("子线程（-%s-）启动" % (threading.current_thread().name))
    
51.         while True:
    
52.             if self.task.empty():
    
53.                 break
    
54.             else:
    
55.                 url = self.task.get()
    
56.                 self.parse(url)
    
57.         print("子线程（-%s-）结束" % (threading.current_thread().name))
    
58. 
    
59.     def parse(self, url) -> None:
    
60.         pass
    
61. 
    
62. 
    
63. class TestMode(TaskThreading):
    
64. 
    
65.     def parse(self, url) -> None:
    
66.         """在这里重写你的爬虫代码"""
    
67.         response = requests.get(url=url)
    
68.         print(f"{threading.current_thread().name}正在模拟等待{url}链接响应，时长5秒")
    
69.         print(f"{threading.current_thread().name}响应成功，状态{response}")
    
70.         data_queue.put(url)
    
71. 
    
72. 
    
73. class Test2Mode(TaskThreading):
    
74. 
    
75.     def run(self) -> None:
    
76.         """
    
77.         如果不进行run重写，则会因为队列里面没有来的及添加任务，
    
78.         导致线程，开启即刻关闭，所以进行堵塞
    
79.         """
    
80.         while not stop_flag:
    
81.             try:
    
82.                 url = self.task.get(False)
    
83.                 if not url:
    
84.                     continue
    
85.                 self.parse(url)
    
86.                 self.task.task_done()
    
87.             except Exception as e:
    
88.                 pass
    
89. 
    
90.     def parse(self, url) -> None:
    
91.         """在这里重写你的爬虫代码"""
    
92.         response = requests.get(url=url)
    
93.         print(f"{threading.current_thread().name}正在模拟等待{url}链接响应，时长5秒")
    
94.         print(f"{threading.current_thread().name}响应成功，状态{response}")
    
95. 
    
96. 
    
97. def main():
    
98.     # 任何进程默认就会启动一个线程，称为主线程，主线程可以启动新的子线程
    
99.     # current_thread()返回当前线程的实例
    
100.     print("主线程（-%s-）启动" % (threading.current_thread().name))
     
101. 
     
102.     import queue
     
103.     # 建好队列，给定爬虫任务
     
104.     get_url_queue = queue.Queue()
     
105.     for page in range(5):
     
106.         get_url_queue.put(f"http://www.httpbin.org")
     
107. 
     
108.     # 初始化第一个爬虫，线程池数量为3
     
109.     threading_pool = ThreadPoolManagement(
     
110.         task_thread=TestMode,
     
111.         task_queue=get_url_queue,
     
112.         number=3
     
113.     )
     
114. 
     
115.     # 初始化第二个爬虫，线程池数量为3
     
116.     threading_poolz = ThreadPoolManagement(
     
117.         task_thread=Test2Mode,
     
118.         task_queue=data_queue,
     
119.         number=3
     
120.     )
     
121. 
     
122.     # 启动两个线程池的线程
     
123.     threading_pool.start_threading()
     
124.     threading_poolz.start_threading()
     
125. 
     
126.     # 检测第一个爬虫队列任务是否为空
     
127.     while not get_url_queue.empty():
     
128.         pass
     
129. 
     
130.     # 关闭第一个爬虫线程池的线程
     
131.     threading_pool.join_threading()
     
132. 
     
133.     # 检测第二个爬虫队列任是否为空
     
134.     while not data_queue.empty():
     
135.         pass
     
136. 
     
137.     # 关闭第二个线程池的线程
     
138.     global stop_flag
     
139.     stop_flag = True
     
140.     threading_poolz.join_threading()
     
141. 
     
142.     print("主线程（-%s-）结束" % (threading.current_thread().name))
     
143. 
     
144. 
     
145. if __name__ == '__main__':
     
146.     main()
     

复制代码