python语言使用全局变量+线程锁解密程序代码

importosimportrarfileimportthreadingfromqueueimportQueueimporttime# 指定路径rarfile.UNRAR_TOOL=r"UnRAR.exe"# 创建一个线程安全的队列用于存储密码password_queue=Queue()# 创建一个线程安全的标志用于指示是否已找到密码password_found=threading.Event()# 存储找到的密码found_password=None# 创建一个锁用于线程安全的输出print_lock=threading.Lock()# ========== 新增：全局计数器 + 计数器锁 ==========global_password_count=0count_lock=threading.Lock()# 保证计数器修改的线程安全deftry_decrypt(file_path,password):"""尝试使用给定密码解压 RAR 文件"""try:withrarfile.RarFile(file_path)asrf:rf.extractall(pwd=password.encode())returnTrueexceptrarfile.BadRarFile:returnFalseexceptExceptionase:withprint_lock:print(f"Error with password{password}:{e}")returnFalsedefworker(file_path):"""工作线程函数，从队列中获取密码并尝试解密"""globalfound_password,global_password_count# 引用全局计数器whilenotpassword_found.is_set():# 从队列中获取密码，如果队列为空则阻塞password=password_queue.get()# 如果已经找到密码，退出循环ifpassword_found.is_set():password_queue.task_done()break# ========== 关键修改：更新全局计数器（加锁保证线程安全） ==========current_global_count=0withcount_lock:global_password_count+=1current_global_count=global_password_count# 获取当前全局计数（避免后续被其他线程修改）# 尝试解密iftry_decrypt(file_path,password):# 设置标志为已找到密码password_found.set()found_password=passwordwithprint_lock:print(f"\n✅ 找到密码:{password}（全局第{current_global_count}个密码）")# 标记任务已完成password_queue.task_done()# 输出当前尝试的密码（使用全局计数，加锁保证打印有序）ifnotpassword_found.is_set():withprint_lock:print(f"🔄 尝试第{current_global_count}个密码:{password}")defmain():# RAR 文件路径file_path=r"password.rar"# 密码字典文件路径，你需要将其替换为实际的密码字典文件路径dict_path=r"wordlist1.txt"# 线程数量num_threads=10# 检查文件是否存在ifnotos.path.exists(file_path):print(f"RAR 文件不存在:{file_path}")returnifnotos.path.exists(dict_path):print(f"密码字典文件不存在:{dict_path}")return# 读取密码字典try:withopen(dict_path,"r",encoding="utf8")asf:passwords=[line.strip()forlineinf.readlines()]ifnotpasswords:print("密码字典为空")returnexceptExceptionase:print(f"读取密码字典时出错:{e}")return# 将密码加入队列forpasswordinpasswords:password_queue.put(password)print(f"开始多线程破解 RAR 文件，使用{num_threads}个线程")print(f"RAR 文件:{file_path}")print(f"密码字典:{dict_path}")print(f"总密码数:{len(passwords)}")print("-"*50)start_time=time.time()# 创建并启动工作线程threads=[]for_inrange(num_threads):t=threading.Thread(target=worker,args=(file_path,))t.daemon=Truet.start()threads.append(t)# 等待队列中的所有任务完成或找到密码password_queue.join()elapsed_time=time.time()-start_time# 输出结果ifpassword_found.is_set():print(f"🎉 破解成功！密码为:{found_password}")else:print("😔 未能找到有效密码")print(f"耗时:{elapsed_time:.2f}秒")print(f"全局总尝试密码数:{global_password_count}")if__name__=="__main__":main()