python语言使用threading.Value解密程序代码

importosimportrarfileimportthreadingfromqueueimportQueueimporttime# 指定路径rarfile.UNRAR_TOOL=r"UnRAR.exe"# 创建一个线程安全的队列用于存储密码password_queue=Queue()# 创建一个线程安全的标志用于指示是否已找到密码password_found=threading.Event()# 存储找到的密码found_password=None# 创建一个锁用于线程安全的输出print_lock=threading.Lock()# ========== 新增：线程安全的全局计数器（'i'表示int类型） ==========global_password_count=threading.Value('i',0)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_passwordwhilenotpassword_found.is_set():password=password_queue.get()ifpassword_found.is_set():password_queue.task_done()break# ========== 关键修改：安全更新线程安全计数器 ==========withglobal_password_count.get_lock():# 手动加锁（可选，Value底层已保证原子性，加锁更严谨）global_password_count.value+=1current_global_count=global_password_count.value# 尝试解密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}")# main函数与方案1完全一致，无需修改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.value}")# 注意这里取值用 .valueif__name__=="__main__":main()