python – 使用thread.join确保所有线程都已连接时的Pydev PyUni
发布时间：2020-12-17 00:50:14 所属栏目：Python 来源：网络整理
导读：使用pydev进行测试时,我遇到了问题.我已经挖掘了这个问题并知道根本原因是什么.我提供了以下代码的示例,可用于重现该问题. 我主要测试Centos 6.3,python 2.7,eclipse juno,pydev 2.7.1,但问题也出现在具有类似设置的Windows 7上. 我有一个python脚本,作为服
|
使用pydev进行测试时,我遇到了问题.我已经挖掘了这个问题并知道根本原因是什么.我提供了以下代码的示例,可用于重现该问题.
我主要测试Centos 6.3,python 2.7,eclipse juno,pydev 2.7.1,但问题也出现在具有类似设置的Windows 7上. 我有一个python脚本,作为服务器不同线程的进程启动器(所有内部第三方库,所以我不能辞去系统的那一面). 为了确保所有线程都在我的process.py结束时完成,我有一段代码试图在退出之前加入所有线程. for t in threading.enumerate():
if t.getName() != 'MainThread':
t.join()
这在正常的生产代码中工作正常. 使用pydev在eclipse中运行PyUnit中的测试时会出现问题.额外的线程被添加到python导致我的测试挂起. 如果我使用Run As启动我的程序 – > Python Run,我的代码按预期运行并退出正常.如果我使用Run As启动我的程序 – > Python单元测试,测试总是挂起. 如果我看一下线程是否可用,问题就会变得清晰.使用提供的测试代码示例,我可以看到当运行测试作为python运行时,显示以下线程(如预期的那样) Thread: <bound method _MainThread.getName of <_MainThread(MainThread,started 140268135126784)>> Thread: <bound method ThreadClass.getName of <ThreadClass(Thread A,started 140268006471424)>> Thread: <bound method ThreadClass.getName of <ThreadClass(Thread B,started 140267927631616)>> 当我将测试作为单元测试运行时 Thread: <bound method _MainThread.getName of <_MainThread(MainThread,started 139904571213568)>> Thread: <bound method WriterThread.getName of <WriterThread(pydevd.Writer,started daemon 139904379361024)>> Thread: <bound method ThreadClass.getName of <ThreadClass(Thread A,started 139904130479872)>> Thread: <bound method ThreadClass.getName of <ThreadClass(Thread B,started 139904119990016)>> Thread: <bound method PyDBCommandThread.getName of <PyDBCommandThread(pydevd.CommandThread,started daemon 139904358381312)>> Thread: <bound method ReaderThread.getName of <ReaderThread(pydevd.Reader,started daemon 139904368871168)>> Thread: <bound method ServerComm.getName of <ServerComm(Thread-4,started 139904345736960)>> python添加的额外线程似乎打破了这段代码.当我的代码尝试加入ServerComm或pydev.Writer时,它会挂起. 我知道我可以尝试不按名称加入这些线程,但是这样我正在改变生产代码来处理这个问题而且我不太热衷于这个解决方案.有没有其他人遇到这个,并找到了一个很好的解决方法？任何帮助都将非常感激.以下是该问题的示例代码. 示例test_process.py import sys
import traceback
import unittest
class TestUBProcessManager(unittest.TestCase):
def setUp(self):
pass
def runTest(self):
globalsDict = {'sys':sys,'__name__':'__main__'}
haveException = False
try:
execfile('Process.py',globalsDict)
except Exception,detail:
haveException = True
traceback.print_exc()
self.assertFalse(haveException)
if __name__ == '__main__':
unittest.main()
示例Process.py import threading
import time
class ThreadClass(threading.Thread):
def __init__(self,threadName,threadRunCount,threadSleep):
threading.Thread.__init__(self)
self.name = threadName;
self.runCount = threadRunCount;
self.sleepTime = threadSleep;
def run(self):
for i in range (1,self.runCount):
print "t",self.name,"Working";
time.sleep(self.sleepTime);
class Launcher(object):
def __init__(self):
print "Init Threads";
self.threadA = ThreadClass("Thread A",3,2)
self.threadB = ThreadClass("Thread B",7,2)
def launchProcess(self):
print "Starting Threads";
self.threadA.start();
self.threadB.start();
time.sleep(2);
if __name__ == '__main__':
launcher = Launcher()
try:
launcher.launchProcess()
finally:
print "Available Threads Needed To Finish"
for t in threading.enumerate():
print "Thread: ",t.getName
print "Attempt to join threads to ensure all threads are finished"
for t in threading.enumerate():
print "About To Join : ",t.getName
if t.getName() != 'MainThread':
t.join()
print "All Done"
解决方法
只是让其他人遇到这个问题.我决定更改我的代码以检查测试开始时存在哪些线程.
if __name__ == '__main__':
initialThreads = [];
for t in threading.enumerate():
initialThreads.append(t);
launcher = Launcher()
try:
launcher.launchProcess()
finally:
print "Available Threads Needed To Finish"
for t in threading.enumerate():
print "Thread: ",t.getName
if t not in initialThreads:
t.join()
print "All Done"
（编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |