【Python系列】并发编程在数据处理中的应用

news2024/12/23 9:09:48

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
导航
檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等
常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等
数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等
懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

- 并发编程的基本概念
- ThreadPoolExecutor 的工作原理
- 示例代码解析
- 并发编程的优势
- 注意事项
- 结语

在现代软件开发中，数据处理是一个常见且关键的任务。随着数据量的不断增长，传统的单线程处理方式已经无法满足日益增长的处理速度需求。为了提高效率，开发者们开始使用并发编程技术，以实现多任务同时执行。Python 语言提供了多种并发执行任务的方法，其中ThreadPoolExecutor是concurrent.futures模块中一个非常实用的工具，它允许开发者轻松地创建线程池来并发执行任务。
在这里插入图片描述

并发编程的基本概念

并发编程是指在计算机程序中，多个任务或线程同时执行，以提高程序的执行效率。这种编程模式可以充分利用多核处理器的优势，实现资源的最大化利用。在 Python 中，有多种实现并发的方法，包括多线程、多进程、异步编程等。

ThreadPoolExecutor 的工作原理

ThreadPoolExecutor是 Python 标准库concurrent.futures模块中的一部分，它提供了一个简单的高层 API 来创建线程池。线程池中的线程可以并发执行多个任务，这些任务可以是函数调用或任何可调用对象。通过使用线程池，可以避免为每个任务创建和销毁线程的开销，从而提高程序的效率。

# 并发请求
from concurrent.futures import ThreadPoolExecutor
import concurrent.futures
futures = []
with ThreadPoolExecutor(max_workers=10) as executor:
    for chunk in combined_chunk_document_list:
        chunk_doc = Document(
            page_content=chunk.page_content.encode("utf-8"), metadata=chunk.metadata
        )
        futures.append(
            executor.submit(llm_transformer.convert_to_graph_documents, [chunk_doc])
        )

    for i, future in enumerate(concurrent.futures.as_completed(futures)):
        graph_document = future.result()
        graph_document_list.append(graph_document[0])

示例代码解析

在提供的代码示例中，我们可以看到ThreadPoolExecutor是如何被用来并发处理文档转换任务的。以下是对示例代码的详细解析：

导入必要的模块：首先，代码导入了concurrent.futures模块中的ThreadPoolExecutor和as_completed函数。
创建线程池：使用with ThreadPoolExecutor(max_workers=10)语句创建了一个最大容纳 10 个工作线程的线程池。with语句确保线程池在使用完毕后能够正确关闭。
任务分发：在with语句的代码块中，遍历combined_chunk_document_list列表，对每个文档块创建一个Document对象，并将其作为参数提交给llm_transformer.convert_to_graph_documents函数进行处理。提交的任务被添加到futures列表中。
任务执行与结果收集：使用concurrent.futures.as_completed(futures)函数遍历futures列表，等待每个任务完成。一旦任务完成，通过调用future.result()获取任务结果，并将结果添加到graph_document_list列表中。