如何判断Python生成tar.gz压缩文件的进程是否正在进行
判断Python生成tar.gz进程是否在运行的几种方法
嘿,这个需求我之前做批量压缩的时候也遇到过,分享几个实用的方案给你:
1. 临时文件标记法(最可靠)
这是我最常用的方法——在压缩过程中先写入一个临时文件,完成后再重命名为最终的.tar.gz文件。这样你只需要检查临时文件是否存在,就能直观判断进程是否在运行。
举个Python代码示例:
import tarfile import os def compress_to_targz(source_dir, output_path): # 定义临时文件路径,建议和最终文件同目录 tmp_file = f"{output_path}.tmp" try: # 写入临时文件 with tarfile.open(tmp_file, "w:gz") as tar: tar.add(source_dir, arcname=os.path.basename(source_dir)) # 压缩完成后,重命名为最终文件 os.rename(tmp_file, output_path) print("压缩完成!") except Exception as e: # 出错时清理临时文件,避免残留 if os.path.exists(tmp_file): os.remove(tmp_file) print(f"压缩出错: {str(e)}") raise e
- 当
{output_path}.tmp存在时,说明压缩正在进行; - 如果临时文件消失且最终
.tar.gz文件存在,说明压缩完成; - 如果临时文件消失但最终文件不存在,大概率是压缩过程出错了。
2. 跟踪子进程状态(用系统tar命令时适用)
如果你是通过subprocess调用系统的tar命令来完成压缩,那可以直接跟踪子进程的运行状态:
import subprocess import time # 启动压缩子进程 compress_proc = subprocess.Popen( ["tar", "-czf", "my_archive.tar.gz", "./source_files"], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # 轮询检查进程状态 while compress_proc.poll() is None: print("压缩进程正在运行...") time.sleep(3) # 每隔3秒检查一次 # 进程结束后判断结果 if compress_proc.returncode == 0: print("压缩成功完成!") else: _, err = compress_proc.communicate() print(f"压缩失败,错误信息: {err.decode('utf-8')}")
proc.poll()返回None表示进程仍在运行;- 返回
0表示压缩成功,非0值则表示出错。
3. 监控输出文件大小变化(备选方案)
如果不想用临时文件,可以通过监控输出文件的大小变化来判断——压缩过程中文件大小会持续增长,直到停止变化(说明压缩完成或卡住)。不过这个方法可靠性稍差,适合简单场景:
import os import time def is_compression_running(output_path, check_window=5): if not os.path.exists(output_path): return False # 文件还没生成,可能还没开始 # 记录初始大小 initial_size = os.path.getsize(output_path) time.sleep(check_window) # 再次获取大小 current_size = os.path.getsize(output_path) # 如果大小增长了,说明还在压缩 return current_size > initial_size # 使用示例 while is_compression_running("my_archive.tar.gz"): print("压缩中...") time.sleep(2) print("压缩完成或已停止")
一些额外建议
- 临时文件名最好加上唯一标识(比如进程ID
os.getpid()),避免多进程同时压缩时出现冲突; - 不管用哪种方法,都要做好异常处理,防止进程意外退出后留下无效文件或临时文件;
- 如果是长时间运行的压缩任务,可以结合日志输出,更直观地跟踪进度。
内容的提问来源于stack exchange,提问作者Lobo
相关产品推荐
相关产品推荐

