相关疑难解决方法(0)

我正在编写一个多线程Python应用程序,它与服务器建立了许多TCP连接.每个连接都在一个单独的线程上完成.有时线程挂了很长时间,我不想要.如何让线程在一段时间后自杀？从主线程如何确定子线程自杀？

如果可能的话,我会欣赏一段显示如何执行此操作的代码.谢谢.

更新 系统是Ubuntu 9:10

我正在使用Tkinter和线程编写应用程序.

我得到的问题是,关闭主应用程序后,一些线程仍在运行,我需要一种方法来检查根窗口是否已被销毁以避免TclError: can't invoke "wm" command.

我知道的所有方法:一旦root被销毁wminfo_exists(),state()所有返回错误.

我是python编程的新手.我正在尝试使用可停止的线程创建GUI.我从/sf/answers/22786991/借用了一些代码

class MyThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self, *args, **kwargs):
        super(MyThread, self).__init__(*args, **kwargs)
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()

    def stopped(self):
        return self._stop.isSet()

我有一个函数,它为另一个运行无限循环的类中的另一个函数创建一个线程.

class MyClass :

    def clicked_practice(self):

        self.practicethread = MyThread(target=self.infinite_loop_method)
        self.practicethread.start()

    def infinite_loop_method()
        while True :
            // Do something


    #This doesn't seem to work and I am still stuck in the loop

    def infinite_stop(self)
        if self.practicethread.isAlive():
        self.practicethread.stop()

我想创建一个方法来停止这个线程.这里发生了什么事？

我试图在python中杀死一个线程.一个例外是这样做的首选方法,作为线程的run方法通过try的优雅退出:except:pair将允许关闭资源.

我试过:有没有办法在Python中杀死一个线程？,但是指定在代码执行系统调用时不起作用(如time.sleep).有没有办法在另一个线程(或进程中,不要介意)中引发一个异常,它不起作用线程正在执行什么？

我有一个python函数,如果传递坏数据会被无限循环捕获.我想写一个单元测试来确认它优雅地处理坏参数.问题当然是,如果它没有检测到错误的参数,它将不会返回.

使用线程为这类事物编写测试是否可以接受？

import threading, unittest
import mymodule

class CallSuspectFunctionThread(threading.Thread):
    def __init__(self, func, *args):
        self.func = func
        self.args = args
        super(CallSuspectFunctionThread, self).__init__()
    def run(self):
        self.func(*self.args)

class TestNoInfiniteLoop(unittest.TestCase):
    def test_no_infinite_loop(self):
        myobj = mymodule.MyObject()
        bad_args = (-1,0)
        test_thread = CallSuspectFunctionThread(mybj.func_to_test, *bad_args)
        test_thread.daemon = True
        test_thread.start()
        # fail if func doesn't return after 8 seconds
        for i in range(32):
            test_thread.join(0.25)
            if not test_thread.is_alive():
                return
        self.fail("function doesn't return")

我看到的唯一问题是,如果测试失败,我会遇到这个额外的线程,可能会消耗cpu和内存资源,而其余的测试都会被执行.另一方面,我已经修复了代码,回归的可能性非常小,所以我不知道是否包括测试是否重要.

我正在使用一个线程从流(/ dev/tty1)读取字符串,同时处理主循环中的其他内容.我想在按CTRL-C时Thread与主程序一起终止.

   from threading import Thread

   class myReader(Thread):
      def run(self):
         with open('/dev/tty1', encoding='ascii') as myStream:
            for myString in myStream:
               print(myString)
      def quit(self):
         pass # stop reading, close stream, terminate the thread

   myReader = Reader()
   myReader.start()
   while(True):
      try:
         pass # do lots of stuff
      KeyboardInterrupt:
         myReader.quit()
         raise

通常的解决方案 - run()循环中的布尔变量 - 在这里不起作用.推荐的解决方法是什么？

我可以设置守护进程标志,但之后我将无法使用quit()方法,这可能在以后证明是有价值的(做一些清理).有任何想法吗？

我想编写一个函数，该函数将在5秒后返回：

def myfunction():
    while passed_time < 5_seconds:
        do1()
        do2()
        do3()
        .
        .
    return

我的意思是，此功能仅运行5秒钟，在5秒钟后，它应结束并继续使用其他功能：

myfunction()
otherfunction()   ----> This should start 5 seconds after myfunction() is executed.

最好的祝福

我有一个 python 程序，它实现了这样的线程：

   class Mythread(threading.Thread):
        def __init__(self, name, q):
            threading.Thread.__init__(self)
            self.name = name
            self.q = q

        def run(self):
            print "Starting %s..." % (self.name)
            while True:
                ## Get data from queue
                data = self.q.get()
                ## do_some_processing with data ###
                process_data(data)
                ## Mark Queue item as done
                self.q.task_done()
            print "Exiting %s..." % (self.name)


    def call_threaded_program():
        ##Setup the threads. Define threads,queue,locks 
        threads = []    
        q = Queue.Queue()
        thread_count = n #some number
        data_list = [] #some data list containing data

        ##Create Threads
        for …

我有一个应用程序,每隔几分钟轮询一堆服务器.为此,它为每个服务器生成一个线程进行轮询(15个服务器)并将数据写回到对象:

import requests

class ServerResults(object):
    def __init__(self):
        self.results = []

    def add_server(some_argument):
        self.results.append(some_argument)

servers = ['1.1.1.1', '1.1.1.2']
results = ServerResults()

for s in servers:
    t = CallThreads(poll_server, s, results)
    t.daemon = True
    t.start()

def poll_server(server, results):
    response = requests.get(server, timeout=10)
    results.add_server(response.status_code);

该CallThreads班是一个辅助函数来调用一个函数(在这种情况下,poll_server()带有参数(在这种情况下s和results),你可以在我的GitHub库看到源Python的实用功能.大部分能正常工作的时间,但是有时候一个线程间歇我不知道为什么,因为我在GET请求上使用了超时.无论如何,如果线程挂起,则挂起的线程会在数小时或数天内累积,然后Python崩溃:

  File "/usr/lib/python2.7/threading.py", line 495, in start
    _start_new_thread(self.__bootstrap, ())
thread.error: can't start new thread

Exception in thread Thread-575 (most likely raised during interpreter shutdown)
Exception in thread …

在此脚本中：

import threading, socket    

class send(threading.Thread):

    def run(self):
        try:
            while True:
                try:
                    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    s.connect((url,port))
                    s.send(b"Hello world!")
                    print ("Request Sent!")
                except:
                    s.close()
        except KeyboardInterrupt:
            # here i'd like to kill all threads if possible

for x in range(800):
    send().start()  

是否可以杀死除键盘中断之外的所有线程？我在网上搜索过，是的，我知道已经有人问过这个问题了，但我对Python真的很陌生，而且我对堆栈上提出的其他问题的方法不太了解。