注意：
该代码爬取小说不久或许会失效，有时候该网站会被封禁，代码只供参考，不同小说不同网址会有差异
神印王座II皓月当空最新章节_神印王座II皓月当空全文免费阅读-笔趣阁
一、多线程爬虫
1.1、单线程爬虫的问题
爬虫通常被认为是IO密集型的程序，因为它们主要依赖于网络请求来获取数据。然而，IO处理的速度相对较慢，可能导致爬虫的整体速度受限。具体来说，单线程爬虫在处理大量网页时效率较低，主要存在以下几个问题：
速度慢：单线程爬虫只能一个接一个地请求网页，缺乏并发能力。在面对大量网页时，这种逐个请求的方式显著增加了爬取所需的时间。
资源利用率低：在等待网络响应的过程中，CPU处于闲置状态，未能充分利用系统资源。这种低效利用不仅影响了爬虫的速度，也使得系统资源浪费。
易受限于网络延迟：网络延迟是影响爬虫效率的重要因素。单线程爬虫在面对高延迟的网络时，整体爬取时间会显著延长，从而降低用户体验。
1.2、原理
多线程爬虫的核心原理是利用多线程并行处理多个网页请求，从而提高网络请求的并发性，加速数据抓取的过程。在这种架构中，爬虫将URL队列中的链接分配给多个线程进行处理，每个线程负责发送网络请求并获取响应。
具体来说，爬虫的工作流程如下：
URL队列：爬虫首先维护一个包含待爬取URL的队列。每个线程从这个队列中取出一个URL进行处理。
并行请求：多个线程同时运行，独立地发送网络请求。在等待响应的过程中，线程不会闲置，而是可以继续处理队列中的其他URL。这种并行处理显著减少了整体的爬取时间。
结果队列：每个线程在获取到网页响应后，将结果存储到一个结果队列中。这个结构确保了数据的有序存储，并允许其他线程安全地读取这些结果。
数据写入：另一些线程则负责从结果队列中读取数据，并将其写入文件或进行进一步处理。这种分工使得爬虫的各个部分能够高效协作，最大限度地提高资源利用率。
1.3、主要组成部分
1.3.1、URL队列和结果队列
URL队列：存储待爬取的URL链接，通常使用线程安全的队列（如
queue.Queue），以便多个线程可以安全地访问和修改。结果队列：用于存放从网页中提取的结果，允许在爬取完成后统一处理或存储。
from queue import Queue
urls_queue = Queue()
out_queue = Queue()1.3.2、类包装
使用多个线程，不停的取URL队列中的url，并进行处理：
import threading
class ThreadCrawl(threading.Thread):def __init__(self, queue, out_queue):threading.Thread.__init__(self)self.queue = queueself.out_queue = out_queuedef run(self):while True:item = self.queue.get()如果队列为空，线程就会被阻塞，直到队列不为空。
处理队列中的 一条数据后，就需要通知队列已经处理完该条数据
1.3.3、函数包装
from threading import Thread
def func(args):pass
if __name__ == '__main__':info_html = Queue()t1 = Thread(target=func,args=(info_html,))1.3.4、线程池
import threading # 导入 threading 模块，用于创建和管理线程
import time # 导入 time 模块，使用 sleep 来模拟工作
import queue # 导入 queue 模块，使用线程安全的队列 class Threadingpool(): def __init__(self, max_num=10): # 初始化线程池，最大线程数量为 max_num self.queue = queue.Queue(max_num) # 创建一个最大大小为 max_num 的队列 for i in range(max_num): # 将线程类的引用放入队列中 self.queue.put(threading.Thread) def getthreading(self): # 从队列中获取一个线程 return self.queue.get() def addthreading(self): # 将新的线程类引用放回队列 self.queue.put(threading.Thread) def func(p, i): # 每个线程执行的函数 time.sleep(1) # 通过休眠1秒来模拟工作 print(i) # 打印传递给函数的索引 p.addthreading() # 线程完成后，将线程返回到池中 if __name__ == "__main__": p = Threadingpool() # 创建一个 Threadingpool 实例 for i in range(20): thread = p.getthreading() # 从池中获取一个线程 t = thread(target=func, args=(p, i)) # 创建一个新线程，目标函数和参数 t.start() # 启动线程1.4、多线程函数爬虫
import re # 导入正则表达式模块
import requests # 导入请求库，用于发送HTTP请求
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML
from threading import Thread # 导入线程模块，用于实现多线程
import time # 导入时间模块，用于控制延时 def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) def get_new_url(url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取网页内容 resp = requests.get(url, header) resp.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 titles = [] # 存储章节标题 new_urls = [] # 存储章节链接 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue # 获取章节标题并打印 print(e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0]) titles.append(e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0]) # 添加标题到列表 # 构造章节链接并添加到列表 new_urls.append('https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0]) return titles, new_urls # 返回标题和链接列表 def spider(title, new_url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取章节内容 resp = requests.get(new_url, header) resp.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 safe_title = sanitize_filename(title) # 清理标题以作为文件名 print(title) # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for i in content: f.write(i.strip() + "\n") # 写入内容并清理空白 time.sleep(2) # 每次抓取后暂停2秒，防止过于频繁的请求 if __name__ == '__main__': url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 titles, new_urls = get_new_url(url) # 获取章节标题和链接 threads = [] # 存储线程列表 # 为每个章节创建一个线程进行抓取 for title, new_url in zip(titles, new_urls): thread = Thread(target=spider, args=(title, new_url)) # 创建线程 threads.append(thread) # 添加线程到列表 thread.start() # 启动线程 # 等待所有线程完成 for thread in threads: thread.join() # 等待每个线程结束1.5、多线程类爬虫
import re # 导入正则表达式模块
import requests # 导入请求库，用于发送HTTP请求
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML
from threading import Thread # 导入线程模块
import time # 导入时间模块，用于控制延时 class NovelSpider(Thread): def __init__(self, title, new_url): super().__init__() # 调用父类的构造函数 self.title = title # 章节标题 self.new_url = new_url # 章节链接 self.headers = {'User-Agent': UserAgent().edge} # 请求头 @staticmethod def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) def run(self): # 发送GET请求获取章节内容 resp = requests.get(self.new_url, headers=self.headers) resp.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 safe_title = self.sanitize_filename(self.title) # 清理标题以作为文件名 print(f"抓取中: {safe_title}") # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for i in content: f.write(i.strip() + "\n") # 写入内容并清理空白 time.sleep(2) # 每次抓取后暂停2秒，防止过于频繁的请求 class NovelCrawler: def __init__(self, base_url): self.base_url = base_url # 基础网址 self.headers = {'User-Agent': UserAgent().edge} # 请求头 self.titles = [] # 存储章节标题 self.new_urls = [] # 存储章节链接 def get_new_url(self): # 发送GET请求获取网页内容 resp = requests.get(self.base_url, headers=self.headers) resp.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue title = e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0] # 获取章节标题 print(title) # 打印章节标题 self.titles.append(title) # 添加标题到列表 new_url = 'https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0] # 构造章节链接 self.new_urls.append(new_url) # 添加链接到列表 def run(self): self.get_new_url() # 获取章节标题和链接 threads = [] # 存储线程列表 # 为每个章节创建一个线程进行抓取 for title, new_url in zip(self.titles, self.new_urls): spider = NovelSpider(title, new_url) # 创建爬虫实例 threads.append(spider) # 添加线程到列表 spider.start() # 启动线程 # 等待所有线程完成 for thread in threads: thread.join() # 等待每个线程结束 if __name__ == '__main__': url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 crawler = NovelCrawler(url) # 创建爬虫控制器实例 crawler.run() # 运行爬虫1.6、 多线程--线程池
import re # 导入正则表达式模块
import requests # 导入请求库，用于发送HTTP请求
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML
from concurrent.futures import ThreadPoolExecutor # 导入线程池执行器
import time # 导入时间模块，用于控制延时 def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) def get_new_url(url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取网页内容 resp = requests.get(url, headers=header) resp.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 titles = [] # 存储章节标题 new_urls = [] # 存储章节链接 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue # 获取章节标题并打印 title = e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0] print(title) # 打印章节标题 titles.append(title) # 添加标题到列表 # 构造章节链接并添加到列表 new_urls.append('https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0]) return titles, new_urls # 返回标题和链接列表 def spider(title, new_url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取章节内容 resp = requests.get(new_url, headers=header) resp.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 safe_title = sanitize_filename(title) # 清理标题以作为文件名 print(f"抓取中: {safe_title}") # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for i in content: f.write(i.strip() + "\n") # 写入内容并清理空白 time.sleep(2) # 每次抓取后暂停2秒，防止过于频繁的请求 if __name__ == '__main__': url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 titles, new_urls = get_new_url(url) # 获取章节标题和链接 # 使用线程池进行抓取 with ThreadPoolExecutor(max_workers=10) as executor: # 设置最大工作线程数 # 为每个章节提交任务 for title, new_url in zip(titles, new_urls): executor.submit(spider, title, new_url) # 提交任务到线程池二、多进程爬虫
multiprocessing是python的多进程管理包，和threading.Thread 类似 multiprocessing模块
multiprocessing模块可以让程序员在给定的机器上充分的利用CPU 在multiprocessing中，通过创建Process对象生成进程，然后调用 它的start()方法
2.1、多进程函数爬虫
import re # 导入正则表达式模块
import requests # 导入请求库，用于发送HTTP请求
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML
from multiprocessing import Process, Manager # 导入进程和管理器模块
import time # 导入时间模块，用于控制延时 def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) def get_new_url(url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取网页内容 resp = requests.get(url, headers=header) resp.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 titles = [] # 存储章节标题 new_urls = [] # 存储章节链接 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue # 获取章节标题并打印 title = e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0] print(title) # 打印章节标题 titles.append(title) # 添加标题到列表 # 构造章节链接并添加到列表 new_urls.append('https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0]) return titles, new_urls # 返回标题和链接列表 def spider(title, new_url): # 设置请求头，伪装成浏览器 header = {'User-Agent': UserAgent().edge} # 发送GET请求获取章节内容 resp = requests.get(new_url, headers=header) resp.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 safe_title = sanitize_filename(title) # 清理标题以作为文件名 print(f"抓取中: {safe_title}") # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for i in content: f.write(i.strip() + "\n") # 写入内容并清理空白 time.sleep(2) # 每次抓取后暂停2秒，防止过于频繁的请求 if __name__ == '__main__': url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 titles, new_urls = get_new_url(url) # 获取章节标题和链接 processes = [] # 存储进程列表 # 为每个章节创建一个进程进行抓取 for title, new_url in zip(titles, new_urls): process = Process(target=spider, args=(title, new_url)) # 创建进程 processes.append(process) # 添加进程到列表 process.start() # 启动进程 # 等待所有进程完成 for process in processes: process.join() # 等待每个进程结束2.2、多进程类爬虫
import re # 导入正则表达式模块
import requests # 导入请求库，用于发送HTTP请求
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML
from multiprocessing import Process # 导入进程模块
import time # 导入时间模块，用于控制延时 class NovelSpider(Process): # 继承自 Process def __init__(self, title, new_url): super().__init__() # 调用父类的构造函数 self.title = title # 章节标题 self.new_url = new_url # 章节链接 self.headers = {'User-Agent': UserAgent().edge} # 请求头 @staticmethod def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) def run(self): # 发送GET请求获取章节内容 resp = requests.get(self.new_url, headers=self.headers) resp.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 safe_title = self.sanitize_filename(self.title) # 清理标题以作为文件名 print(f"抓取中: {safe_title}") # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for i in content: f.write(i.strip() + "\n") # 写入内容并清理空白 time.sleep(2) # 每次抓取后暂停2秒，防止过于频繁的请求 class NovelCrawler: def __init__(self, base_url): self.base_url = base_url # 基础网址 self.headers = {'User-Agent': UserAgent().edge} # 请求头 self.titles = [] # 存储章节标题 self.new_urls = [] # 存储章节链接 def get_new_url(self): # 发送GET请求获取网页内容 resp = requests.get(self.base_url, headers=self.headers) resp.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp.text) # 解析网页内容为HTML树结构 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue title = e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0] # 获取章节标题 print(title) # 打印章节标题 self.titles.append(title) # 添加标题到列表 new_url = 'https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0] # 构造章节链接 self.new_urls.append(new_url) # 添加链接到列表 def run(self): self.get_new_url() # 获取章节标题和链接 processes = [] # 存储进程列表 # 为每个章节创建一个进程进行抓取 for title, new_url in zip(self.titles, self.new_urls): spider = NovelSpider(title, new_url) # 创建爬虫实例 processes.append(spider) # 添加进程到列表 spider.start() # 启动进程 # 等待所有进程完成 for process in processes: process.join() # 等待每个进程结束 if __name__ == '__main__': url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 crawler = NovelCrawler(url) # 创建爬虫控制器实例 crawler.run() # 运行爬虫三、协程爬虫
网络爬虫速度效率慢，多部分在于阻塞IO这块(网络/磁盘)。在阻塞 时，CPU的中内核是可以处理别的非IO操作。因此可以考虑使用协 程来提升爬虫效率，这种操作的技术就是协程
协程一种轻量级线程，拥有自己的寄存器上下文和栈，本质是一个进程
相对于多进程，无需线程上下文切换的开销，无需原子操作锁定及同步的开销
简单的说就是让阻塞的子程序让出CPU给可以执行的子程序
一个进程包含多个线程，一个线程可以包含多个协程
多个线程相对独立，线程的切换受系统控制。 多个协程也相对独立，但是其切换由程序自己控制
pip install aiohttp
| 属性或方法 | 功能 |
|---|---|
| aiohttp.ClientSession() | 获取客户端函数 |
| session.get(url) | 发送get请求 |
| seesion.post(url) | 发送post请求 |
| resp.status | 获取响应状态码 |
| resp.url | 获取响应url地址 |
| resp.cookies | 获取响应cookie内容 |
| resp.headers | 获取响应头信息 |
| resp.read() | 获取响应bytes类型 |
| resp.text() | 获取响应文本内容 |
import re # 导入正则表达式模块
import aiohttp # 导入异步HTTP请求库
import asyncio # 导入异步库
from fake_useragent import UserAgent # 导入假用户代理库，用于伪装请求的用户代理
from lxml import etree # 导入lxml库，用于解析HTML async def sanitize_filename(title): # 替换 Windows 文件名中的无效字符 return re.sub(r'[<>:"/\\|?*]', '', title) async def fetch(session, url): # 异步获取网页内容 async with session.get(url) as response: return await response.text() async def get_new_url(url): header = {'User-Agent': UserAgent().edge} async with aiohttp.ClientSession() as session: # 创建异步会话 resp_text = await fetch(session, url) # 获取网页内容 resp_text.encoding = 'gbk' # 设置网页编码为'gbk' e = etree.HTML(resp_text) # 解析网页内容为HTML树结构 titles = [] # 存储章节标题 new_urls = [] # 存储章节链接 # 遍历指定范围内的章节 for i in range(13, 516): if i == 368: # 跳过特定章节 continue title = e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/text()")[0] # 获取章节标题 print(title) # 打印章节标题 titles.append(title) # 添加标题到列表 # 构造章节链接并添加到列表 new_urls.append('https://www.bbiquge.cc/book_61985/' + e.xpath(f"//div[@id='list']/dl/dd[{i}]/a/@href")[0]) return titles, new_urls # 返回标题和链接列表 async def spider(title, new_url): header = {'User-Agent': UserAgent().edge} async with aiohttp.ClientSession() as session: # 创建异步会话 resp_text = await fetch(session, new_url) # 异步获取章节内容 resp_text.encoding = 'gbk' # 设置编码为'gbk' e = etree.HTML(resp_text) # 解析网页内容为HTML树结构 safe_title = await sanitize_filename(title) # 清理标题以作为文件名 print(f"抓取中: {safe_title}") # 打印当前正在抓取的章节标题 content = e.xpath("//div[@id='content']/text()") # 获取章节内容 # 打开文件以写入章节内容 with open(f'./神印王座/{safe_title}.txt', 'a+', encoding='utf-8') as f: for line in content: f.write(line.strip() + "\n") # 写入内容并清理空白 async def main(): url = 'https://www.bbiquge.cc/book_61985/' # 目标网址 titles, new_urls = await get_new_url(url) # 获取章节标题和链接 # 创建任务列表并并发执行 tasks = [spider(title, new_url) for title, new_url in zip(titles, new_urls)] await asyncio.gather(*tasks) # 使用 gather() 并发 if __name__ == '__main__': asyncio.run(main()) # 运行主协程函数