关闭正在运行的python线程时出现问题

Question

我有一个字符串形式的单行加密代码和一个解码&；执行加密代码（该代码建立到网络中另一台计算机的后门连接，但在这里不是很相关），一旦加密代码运行，程序将不会继续到下一行，因为代码不会停止（以保持连接活动）

我想做的是运行代码10秒钟，然后终止它，然后重新运行它

但我不能终止代码，因为它是加密的，无法更改。我想我可以在一个单独的线程上运行它，然后在10秒后强制终止它，但问题是互联网上提到的所有解决方案都需要稍微更改加密代码

Python似乎没有办法在“n”秒后强制关闭正在运行的线程，这将非常容易地解决我的问题

以下代码演示了我的问题：

import threading, time

encrypted_code = ")'cba'(tnirp :eurT elihw"

def decrypt_and_execute(encrypted_code):
    exec(encrypted_code[::-1])

def run(encrypted_code, run_time):
    my_thread = threading.Thread(target=decrypt_and_execute, args=(encrypted_code,))
    my_thread.start()
    time.sleep(run_time)
    # CODE TO STOP 'my_thread' without changing 'encrypted_code' NEEDED HERE#

while True:
    threading.Thread(target=run, args=(encrypted_code, 10)).start()
    time.sleep(10)

上面是我尝试的解决方案（这里的加密代码是一个无限的while循环，它一直在打印abc）。 我试图在不关闭整个应用程序的情况下停止“我的线程”的任何代码都不起作用

Answer 1

也许带跟踪的线程技术对您有效，是一个您可以停止的线程子类。从https://www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/

或者使用子进程，它们已经终止。同样的帖子也很好地涵盖了这一点

编辑：由于某种原因，在所讨论的案例中，这无法终止线程。多进程模块解决了这个问题：杀死进程是在所有操作系统上停止事情的可靠方法，多进程模块有一个spawn方法，它为子进程提供了一个类似线程的API，并且可以在Windows上使用Python代码启动新进程，而无需安装Python

无论如何，对于其他情况，这种线程技术也应该起作用。实际上很少需要线程，但有时它们确实需要

# Python program using 
# traces to kill threads 

import sys 

import trace 
import threading 
import time 

class thread_with_trace(threading.Thread): 
  def __init__(self, *args, **keywords): 
    threading.Thread.__init__(self, *args, **keywords) 
    self.killed = False

  def start(self): 
    self.__run_backup = self.run 
    self.run = self.__run       
    threading.Thread.start(self) 

  def __run(self): 
    sys.settrace(self.globaltrace) 
    self.__run_backup() 
    self.run = self.__run_backup 

  def globaltrace(self, frame, event, arg): 
    if event == 'call': 
      return self.localtrace 

    else: 
      return None 

  def localtrace(self, frame, event, arg): 
    if self.killed: 
      if event == 'line': 
        raise SystemExit() 

    return self.localtrace 
  
  def kill(self): 
    self.killed = True

def func(): 
  while True: 
    print('thread running') 

t1 = thread_with_trace(target = func) 
t1.start() 

time.sleep(2) 
t1.kill() 
t1.join() 

if not t1.isAlive(): 
  print('thread killed')