线程和进程大部分人估计都知道，但协程就不一定了。
一、进程
进程是操作系统分配资源和调度的基本单位，一个程序开始运行时，操作系统会给他分配一块独立的内存空间并分配一个PCB作为唯一标识。初始化内存空间后进程进入就绪态，PCB插入就绪队列。轮到该进程时，操作系统会给进程分配CPU时间片，让进程进入运行态。时间片用完后，重新返回就绪队列，等待下一次分配。如果运行途中，进程遇到了阻塞事件，就会让出CPU给其他就绪进程，自己则进入阻塞队列，待阻塞结束后，重新返回就绪队列。
特征：
- 动态性：进程是程序的一次执行过程，有生命期。
- 并发性：多个进程实体同存于内存中，能并发执行。
- 独立性：进程是资源分配的基本单位，拥有独立的内存空间和系统资源。
- 异步性：进程以各自独立、不可预知的速度向前推进。
- 结构特性：每个进程由程序段、数据段和一个进程控制块（PCB）三部分组成。
使用：
进程用的比较少，线程协程用的多，因为进程之间的切换需要的资源太多了，比较慢，而且因为内存独立而不好通信。
今天试试这个网站泰坦陨落2steam版新手常见问题解决方法汇总 新手入门指南_逗游网一个游戏攻略，我们要尝试获取每一页红框中的内容，总共9页。
先来看看数据在不在页面源代码中，使用 Ctrl+U进入页面源代码，再Ctrl+F查找文字内容。发现页面源代码里有，这表明内容非脚本生成的，少了一大截麻烦。
老样子，通过抓包找到请求，就知道了url和请求方法，根据p不同的取值（1~9）即可切换不同的页面。现在我们可以开始写代码了。
初始代码
先来个无并行的，只记录请求部分时间。最后结果存在word文档里面。之后只展示控制台输出，word输出没什么差别。
import time
from io import BytesIO
import requests
from bs4 import BeautifulSoup
from docx import Document
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
from docx.shared import Inches, RGBColor
def out_word(bs_datas, out_path): # 将bs_datas内容保存，out_path为保存文件名
# 创建Word文档
doc = Document()
# 遍历HTML内容
for bs_data in bs_datas:
if bs_data is None:
continue
for section in bs_data.find_all('p'):
section_all = section.find_all() # 获取部分中所有元素
section_all.insert(0, section) # 列表第一个插入
section_text = section_all[-1].string # 最后一个应该是无嵌套的纯文本
if section_text is None or section_text.strip() == "": # 空的看看是不是图片
if section_all[-1].name == "img":
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run()
# 下载图片
img_url = section_all[-1]['src']
img_response = requests.get(img_url) # 下载图片
img_stream = BytesIO(img_response.content)
# 将图片添加到Word文档中
run.add_picture(img_stream, Inches(5)) # 调整图片宽度
else:
continue
else: # 有文本，写下来
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run('\t' + section_text.strip()) # 获取标签文本,前面空格
for part in section_all: # 遍历每个部分，给段落添加属性
if part.name == 'strong': # 粗体
run.bold = True # 设置为粗体
elif 'align' in part.attrs: # 有对齐方式，这估计只有图片有个居中对齐，不过都写上吧
align = part['align'].upper()
if align == 'LEFT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif align == 'RIGHT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.RIGHT
elif align == 'CENTER':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
elif align == 'JUSTIFY':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.JUSTIFY
else:
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif part.name == 'span' and 'style' in part.attrs: # 有颜色
style = part['style']
if style.startswith('color:'):
color_str = style.split(':')[1]
# 将颜色字符串转换为 RGB 分量
r, g, b = int(color_str[1:3], 16), int(color_str[3:5], 16), int(color_str[5:7], 16)
# 创建一个 RGBColor 对象
color = RGBColor(r, g, b)
run.font.color.rgb = color # 上色
# 保存Word文档
doc.save(out_path + '.docx')
def get_text(bs_datas, url, i): # 获取p=i页内容（bs4）存在bs_datas[i-1]
params = {
"p": str(i)
}
with requests.get(url=url, headers=headers, params=params) as resp:
resp.encoding = "utf-8" # 当页面乱码改这里
bs = BeautifulSoup(resp.text, "html.parser")
data = bs.find("div", class_="CH396071PsfiiY01QjM3f")
bs_datas[i - 1] = data
print(f"页面{i}结束")
url = "https://www.doyo.cn/article/396071"
headers = {
# 用户代理，某些网站验证用户代理，微微改一下，如果提示要验证码之类的，使用它
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 "
"Safari/537.36 Edg/126.0.0.0",
}
bs_datas = [None] * 9
now_time = time.time() # 记录访问时间
for i in range(1, 10):
get_text(bs_datas, url, i)
# get_text(bs_datas, url, 1)
print("用时:", time.time() - now_time)
print(bs_datas)
out_word(bs_datas, 'output')
进程改装代码
平常使用时，我们可以通过multiprocessing.Pool创建进程池或者multiprocessing.Process创建进程。先来两段代码展示一下进程池和进程的使用
进程池,进程池像是工具箱，会自动分配和回收进程。
import time
from multiprocessing import Pool
import os
def task(m,n):
for i in range(m,n):
print(f"进程： {os.getpid()} 输出：{i}")
# time.sleep(0) # 睡一下，让出cpu（为了更好展示并发性）
if __name__ == "__main__":
ranges = [(10,100)]*10
with Pool(processes=3) as pool: # 使用3个进程
pool.starmap(task, ranges) # 有10个任务
进程，注意你启动进程后，只是把它放到了就绪队列，具体什么时候运行要看什么时候轮到它。
import time
from multiprocessing import Process
import os
def task(m,n):
for i in range(m,n):
print(f"进程： {os.getpid()} 输出：{i}")
# time.sleep(0) # 睡一下，让出cpu（为了更好展示并发性）
if __name__ == "__main__":
processes = []
for i in range(3):
p = Process(target=task, args=(100,1000))
processes.append(p)
for p in processes: # 启动所有进程
p.start()
for p in processes: # join等待进程结束后才会继续运行。
p.join()
print("全部进程结束")
两种方式都会出现下面这种现象，进程a和进程有几率交叉输出，这就是并发的表现
下面是用进程改装的代码，（图片请求591了，可能是刷太多次不让看了？所以加了个200判断，不影响。）切实能快一点。
import time
from io import BytesIO
from multiprocessing import Pool
import requests
from bs4 import BeautifulSoup
from docx import Document
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
from docx.shared import Inches, RGBColor
def out_word(bs_datas, out_path): # 将bs_datas内容保存，out_path为保存文件名
# 创建Word文档
doc = Document()
# 遍历HTML内容
for bs_data in bs_datas:
if bs_data is None:
continue
for section in bs_data.find_all('p'):
section_all = section.find_all() # 获取部分中所有元素
section_all.insert(0, section) # 列表第一个插入
section_text = section_all[-1].string # 最后一个应该是无嵌套的纯文本
if section_text is None or section_text.strip() == "": # 空的看看是不是图片
if section_all[-1].name == "img":
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run()
# 下载图片
img_url = section_all[-1]['src']
img_response = requests.get(img_url)
if not img_response ==200:
continue# 下载图片
img_stream = BytesIO(img_response.content)
# 将图片添加到Word文档中
run.add_picture(img_stream, Inches(5)) # 调整图片宽度
else:
continue
else: # 有文本，写下来
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run('\t' + section_text.strip()) # 获取标签文本,前面空格
for part in section_all: # 遍历每个部分，给段落添加属性
if part.name == 'strong': # 粗体
run.bold = True # 设置为粗体
elif 'align' in part.attrs: # 有对齐方式，这估计只有图片有个居中对齐，不过都写上吧
align = part['align'].upper()
if align == 'LEFT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif align == 'RIGHT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.RIGHT
elif align == 'CENTER':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
elif align == 'JUSTIFY':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.JUSTIFY
else:
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif part.name == 'span' and 'style' in part.attrs: # 有颜色
style = part['style']
if style.startswith('color:'):
color_str = style.split(':')[1]
# 将颜色字符串转换为 RGB 分量
r, g, b = int(color_str[1:3], 16), int(color_str[3:5], 16), int(color_str[5:7], 16)
# 创建一个 RGBColor 对象
color = RGBColor(r, g, b)
run.font.color.rgb = color # 上色
# 保存Word文档
doc.save(out_path + '.docx')
def get_text(bs_datas, headers, url, i): # 获取p=i页内容（bs4）存在bs_datas[i-1]
params = {
"p": str(i)
}
with requests.get(url=url, headers=headers, params=params) as resp:
resp.encoding = "utf-8" # 当页面乱码改这里
bs = BeautifulSoup(resp.text, "html.parser")
data = bs.find("div", class_="CH396071PsfiiY01QjM3f")
print(f"页面{i}结束")
return str(data),i
if __name__ == "__main__":
url = "https://www.doyo.cn/article/396071"
headers = {
# 用户代理，某些网站验证用户代理，微微改一下，如果提示要验证码之类的，使用它
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 "
"Safari/537.36 Edg/126.0.0.0",
}
bs_datas = [None] * 9
data_chunks = [(bs_datas,headers, url, i) for i in range(1,10)]
now_time = time.time() # 记录访问时间
with Pool(processes=4) as pool: # 使用4个进程
datas = pool.starmap(get_text, data_chunks)
print("用时:", time.time() - now_time)
for data in datas:
bs_datas[data[1]-1]=BeautifulSoup(data[0], "html.parser")
print(bs_datas)
out_word(bs_datas, 'output')
二、线程
线程是cpu调度的基本单位，一个进程能有很多个线程（至少一个），进程是线程的容器。他的状态、特性、使用都和进程相似，有时候也成为轻量级进程。
与进程相比，线程的资源分配、调度、切换的花销都更少。而线程因为没有自己独立的内存空间，在通信上更灵活。
特征：
- 轻量级：相对于进程而言，线程是轻量级的执行单元，它只拥有一点必不可少的资源，如程序计数器、一组寄存器和栈。
- 共享资源：线程属于同一进程，它们共享进程的内存空间和资源，这使得线程之间的通信更加方便。
- 独立执行流：每个线程都有自己的执行路径，线程在执行过程中独立运行，互不干扰。
- 上下文切换快：线程间的上下文切换相对较快，因为线程共享了大部分上下文信息。
使用：
在python中，进程和线程的创建、使用的代码十分相似，这里只展示concurrent库线程池的使用：
import threading
from concurrent.futures import ThreadPoolExecutor
import time
def task(m, n):
for i in range(m, n):
print(f"线程： {threading.get_ident()} 输出：{i}")
# time.sleep(0) # 睡一下，让出cpu（为了更好展示并发性）
if __name__ == "__main__":
with ThreadPoolExecutor(3) as t: # 使用3个线程
for i in range(10):
t.submit(task,m=10,n=100)
print("end")
使用线程修改代码： 线程因为切换消耗少，自然速度能更快一些。
import time
from io import BytesIO
from concurrent.futures import ThreadPoolExecutor
import requests
from bs4 import BeautifulSoup
from docx import Document
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
from docx.shared import Inches, RGBColor
def out_word(bs_datas, out_path): # 将bs_datas内容保存，out_path为保存文件名
# 创建Word文档
doc = Document()
# 遍历HTML内容
for bs_data in bs_datas:
if bs_data is None:
continue
for section in bs_data.find_all('p'):
section_all = section.find_all() # 获取部分中所有元素
section_all.insert(0, section) # 列表第一个插入
section_text = section_all[-1].string # 最后一个应该是无嵌套的纯文本
if section_text is None or section_text.strip() == "": # 空的看看是不是图片
if section_all[-1].name == "img":
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run()
# 下载图片
img_url = section_all[-1]['src']
img_response = requests.get(img_url)
if not img_response ==200:
continue# 下载图片
img_stream = BytesIO(img_response.content)
# 将图片添加到Word文档中
run.add_picture(img_stream, Inches(5)) # 调整图片宽度
else:
continue
else: # 有文本，写下来
paragraph = doc.add_paragraph() # 添加一个新的段落
run = paragraph.add_run('\t' + section_text.strip()) # 获取标签文本,前面空格
for part in section_all: # 遍历每个部分，给段落添加属性
if part.name == 'strong': # 粗体
run.bold = True # 设置为粗体
elif 'align' in part.attrs: # 有对齐方式，这估计只有图片有个居中对齐，不过都写上吧
align = part['align'].upper()
if align == 'LEFT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif align == 'RIGHT':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.RIGHT
elif align == 'CENTER':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
elif align == 'JUSTIFY':
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.JUSTIFY
else:
paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.LEFT
elif part.name == 'span' and 'style' in part.attrs: # 有颜色
style = part['style']
if style.startswith('color:'):
color_str = style.split(':')[1]
# 将颜色字符串转换为 RGB 分量
r, g, b = int(color_str[1:3], 16), int(color_str[3:5], 16), int(color_str[5:7], 16)
# 创建一个 RGBColor 对象
color = RGBColor(r, g, b)
run.font.color.rgb = color # 上色
# 保存Word文档
doc.save(out_path + '.docx')
def get_text(bs_datas, headers, url, i): # 获取p=i页内容（bs4）存在bs_datas[i-1]
params = {
"p": str(i)
}
with requests.get(url=url, headers=headers, params=params) as resp:
resp.encoding = "utf-8" # 当页面乱码改这里
bs = BeautifulSoup(resp.text, "html.parser")
data = bs.find("div", class_="CH396071PsfiiY01QjM3f")
bs_datas[i - 1] = data
print(f"页面{i}结束")
if __name__ == "__main__":
url = "https://www.doyo.cn/article/396071"
headers = {
# 用户代理，某些网站验证用户代理，微微改一下，如果提示要验证码之类的，使用它
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 "
"Safari/537.36 Edg/126.0.0.0",
}
bs_datas = [None] * 9
data_chunks = [(bs_datas,headers, url, i) for i in range(1,10)]
now_time = time.time() # 记录访问时间
with ThreadPoolExecutor(4) as t: # 使用4个线程
for i in range(1,10):
t.submit(get_text,bs_datas=bs_datas,headers=headers,url=url,i=i)
print("用时:", time.time() - now_time)
print(bs_datas)
out_word(bs_datas, 'output')
三、协程
协程比线程更小，它完全是在程序中切换代码的执行，而不需要操作系统的参与，具体来说，它可以在程序阻塞时去执行其他的代码段，而不需要白白让出cpu，不必切换线程也意味着没有切换消耗。下面来一段程序来表现协程：
import asyncio
import time
def f1s():
print("f1s-in")
time.sleep(1)
print("f1s-out")
def f3s():
print("f3s-in")
time.sleep(3)
print("f3s-out")
def f5s():
print("f5s-in")
time.sleep(5)
print("f5s-out")
async def af1s():
print("f1s-in")
# time.sleep(1) # time.sleep是同步操作，会终止异步
await asyncio.sleep(1) # 挂起代码，异步操作
print("f1s-out")
async def af3s():
print("f3s-in")
# time.sleep(3)
await asyncio.sleep(3)
print("f3s-out")
async def af5s():
print("f5s-in")
# time.sleep(5)
await asyncio.sleep(5)
print("f5s-out")
async def main():
now_time = time.time()
tasks = [asyncio.create_task(af1s()),
asyncio.create_task(af3s()),
asyncio.create_task(af5s())]
await asyncio.wait(tasks) # 一次启动多个任务
print("启动协程：", time.time() - now_time)
if __name__ == '__main__':
now_time = time.time()
f1s()
f3s()
f5s()
print("正常：", time.time() - now_time)
asyncio.run(main())
可以看出协程能够在阻塞时执行其他函数，节约寿命。在爬虫中，请求服务器的过程中有大量等待操作，协程能尽可能利用这段等待时间。
如果要在爬虫中使用协程，我们需要将会产生阻塞的函数换成协程异步函数，比如文件读写用aiofiles库，网络请求用aiohttp库。
下面是用协程的代码，只需0.25s更快了。（没加后面保存部分，之前也一直没计算保存部分时间。）
import asyncio
import time
import aiohttp
from bs4 import BeautifulSoup
async def get_text(bs_datas, headers, url, i): # 获取p=i页内容（bs4）存在bs_datas[i-1]
params = {
"p": str(i)
}
async with aiohttp.ClientSession() as session: # aiohttp.ClientSession()相当于requests
async with session.get(url=url, headers=headers, params=params)as resp:
bs = BeautifulSoup(await resp.text(), "html.parser") # 图片换.content.red()
data = bs.find("div", class_="CH396071PsfiiY01QjM3f")
bs_datas[i - 1] = data
print(f"页面{i}结束")
async def main(bs_datas, headers, url):
now_time = time.time() # 记录访问时间
tasks = []
for i in range(1, 10):
tasks.append(asyncio.create_task(get_text(bs_datas=bs_datas, headers=headers, url=url, i=i)))
await asyncio.wait(tasks) # 一次启动多个任务
print("启动协程：", time.time() - now_time)
if __name__ == "__main__":
url = "https://www.doyo.cn/article/396071"
headers = {
# 用户代理，某些网站验证用户代理，微微改一下，如果提示要验证码之类的，使用它
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 "
"Safari/537.36 Edg/126.0.0.0",
}
bs_datas = [None] * 9
# asyncio.run(main(bs_datas, headers, url))
loop = asyncio.get_event_loop()
loop.run_until_complete(main(bs_datas, headers, url))
print(bs_datas)
在大量请求下，协程能够节省大量时间，请求数目越多越明显
后续：五、抓取图片、视频
会有大量请求，正是协程展示的好时机
