工作节点是分布式爬虫的核心部分，负责执行实际的爬取和下载任务。以下是工作节点的实现步骤和示例代码。

1. 工作节点的职责

• 接收任务 ：从消息队列中获取任务（如URL）。
• 爬取网页 ：下载网页内容并提取PDF链接。
• 下载PDF ：下载PDF文件并保存到指定位置。
• 返回结果 ：将任务结果返回给主节点或存储到数据库。

2. 实现步骤

1. 初始化工作节点 ：

   • 连接到消息队列（如RabbitMQ、Kafka）。
   • 准备好存储PDF文件的目录。

2. 接收任务 ：

   • 从消息队列中获取URL任务。

3. 爬取网页 ：

   • 下载网页内容，解析并提取PDF链接。

4. 下载PDF ：

   • 下载PDF文件并保存到本地或分布式存储。

5. 返回结果 ：

   • 将下载结果返回给主节点或存储到数据库。

3. 示例代码（Python + Celery）

# 初始化Celery应用 app = Celery('pdf_crawler', broker='pyamqp://guest@localhost//')

@app.task def download_pdf(url, save_dir): # 爬取网页内容 response = requests.get(url) soup = BeautifulSoup(response.content, 'html.parser')

提取PDF链接

pdf_links = [] for link in soup.find_all(‘a’, href=True): if link[‘href’].endswith(‘.pdf’): pdf_links.append(link[‘href’])

下载PDF文件

downloaded_files = [] for pdf_link in pdf_links: pdf_url = pdf_link if pdf_link.startswith(‘http’) else f”{url}/{pdf_link}” pdf_response = requests.get(pdf_url) pdf_filename = os.path.join(save_dir, os.path.basename(pdf_url)) with open(pdf_filename, ‘wb’) as f: f.write(pdf_response.content) downloaded_files.append(pdf_filename)

return downloaded_files

# 工作节点启动代码 if **name** == "**main** ": # 启动Celery worker app.worker_main() ```

### 4\. **运行工作节点**

  * 在命令行中运行以下命令启动工作节点： `bash celery -A worker_node worker --loglevel=info`
  * 其中`worker_node`是你的Python脚本文件名（不含`.py`后缀）。

### 5\. **工作节点的扩展**

  * **并发处理** ：可以通过增加工作节点数量或配置Celery的并发数来提高处理能力。
  * **错误处理** ：增加重试机制和错误处理逻辑，确保任务失败后能重新尝试。
  * **日志记录** ：记录任务执行情况，便于监控和调试。

### 6\. **总结**

  * **任务接收** ：工作节点从消息队列中获取任务。
  * **网页爬取** ：下载网页并提取PDF链接。
  * **PDF下载** ：下载PDF文件并保存。
  * **结果返回** ：将结果返回给主节点或存储到数据库。

通过这些步骤，你可以实现一个高效的工作节点，用于分布式爬取和下载PDF文件。