我在使用 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()
这在正常的生产代码中工作正常。
在 Eclipse 中使用 pydev 在 PyUnit 中运行测试时会出现此问题。额外的线程被添加到 python 中，导致我的测试挂起。
如果我使用 Run As -> Python Run 启动我的程序，我的代码将按预期运行并正常退出。如果我使用 Run As -> Python unit-test 启动我的程序，测试总是挂起。
如果我查看可用的线程，问题就会变得清晰。使用提供的测试代码示例，我可以看到当只是将测试作为 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()
示例流程.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"