我要评论引言
在现代软件开发中，处理大量文件或数据时，提高处理效率和并发性是非常重要的。
python 的 asyncio 库提供了一种强大的方式来实现异步编程，从而提高程序的并发处理能力。
本文将面向 python 初级程序员，介绍如何使用 asyncio 和 logging 模块来实现一个异步批处理文件的并发处理系统。
代码实现
1. 日志配置
首先，我们需要配置日志系统，以便在处理文件时记录日志信息。
日志配置包括设置日志格式和输出位置。
import logging
import os
# 获取当前文件的绝对路径
current_file = os.path.abspath(__file__)
# 配置日志格式
log_format = '%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s'
logging.basicconfig(format=log_format, level=logging.info)
# 创建一个文件处理器，并将日志输出到文件
file_handler = logging.filehandler('app.log')
file_handler.setformatter(logging.formatter(log_format))
logging.getlogger().addhandler(file_handler)2. 异步批处理类
接下来，我们定义一个 asyncbatchprocessor 类，用于处理批量文件。
该类使用 asyncio.semaphore 来控制并发任务的数量。
import asyncio
import random
default_max_concurrent_tasks = 2 # 最大并发任务数
max_retries = 3 # 最大重试次数
class asyncbatchprocessor:
def __init__(self, max_concurrent: int = default_max_concurrent_tasks):
self.max_concurrent = max_concurrent
self.semaphore = asyncio.semaphore(max_concurrent)
async def process_single_file(
self,
input_file: str,
retry_count: int = 0
) -> none:
"""处理单个文件的异步方法"""
async with self.semaphore: # 使用信号量控制并发
try:
logging.info(f"processing file: {input_file}")
# 模拟文件处理过程
await asyncio.sleep(random.uniform(0.5, 2.0))
logging.info(f"successfully processed {input_file}")
except exception as e:
logging.error(f"error processing {input_file} of attempt {retry_count}: {str(e)}")
if retry_count < max_retries:
logging.info(f"retrying {input_file} (attempt {retry_count + 1})")
await asyncio.sleep(1)
await self.process_single_file(input_file, retry_count + 1)
else:
logging.error(f"failed to process {input_file} after {max_retries} attempts")
async def process_batch(
self,
file_list: list
) -> none:
total_files = len(file_list)
logging.info(f"found {total_files} files to process")
# 创建工作队列
queue = asyncio.queue()
# 将所有文件放入队列
for file_path in file_list:
await queue.put(file_path)
# 创建工作协程
async def worker(worker_id: int):
while true:
try:
# 非阻塞方式获取任务
input_file_path = await queue.get()
logging.info(f"worker {worker_id} processing: {input_file_path}")
try:
await self.process_single_file(input_file_path)
except exception as e:
logging.error(f"error processing {input_file_path}: {str(e)}")
finally:
queue.task_done()
except asyncio.queueempty:
# 队列为空，工作结束
break
except exception as e:
logging.error(f"worker {worker_id} encountered error: {str(e)}")
break
# 创建工作任务
workers = []
for i in range(self.max_concurrent):
worker_task = asyncio.create_task(worker(i))
workers.append(worker_task)
# 等待队列处理完成
await queue.join()
# 取消所有仍在运行的工作任务
for w in workers:
w.cancel()
# 等待所有工作任务完成
await asyncio.gather(*workers, return_exceptions=true)3. 异步批处理入口函数
最后，我们定义一个异步批处理入口函数 batch_detect，用于启动批处理任务。
async def batch_detect(
file_list: list,
max_concurrent: int = default_max_concurrent_tasks
):
"""异步批处理入口函数"""
processor = asyncbatchprocessor(max_concurrent)
await processor.process_batch(file_list)
# 示例调用
file_list = ["file1.pdf", "file2.pdf", "file3.pdf", "file4.pdf"]
asyncio.run(batch_detect(file_list))代码解释
1.日志配置：
- 使用
logging模块记录日志信息，包括时间、日志级别、文件路径和行号、以及日志消息。 - 日志输出到文件
app.log中，便于后续查看和分析。
2.异步批处理类 asyncbatchprocessor：
__init__方法初始化最大并发任务数和信号量。process_single_file方法处理单个文件，使用信号量控制并发，模拟文件处理过程，并在失败时重试。process_batch方法处理批量文件，创建工作队列和协程，控制并发任务的执行。
3.异步批处理入口函数 batch_detect：
- 创建
asyncbatchprocessor实例，并调用process_batch方法启动批处理任务。
总结
通过使用 asyncio 和 logging 模块，我们实现了一个高效的异步批处理文件系统。
该系统能够并发处理大量文件，并在处理失败时自动重试，直到达到最大重试次数。
日志系统帮助我们记录每个文件的处理过程，便于后续的调试和分析。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持代码网。
发表评论