正确使用Python中编写的数据收集应用程序中的多线程和队列模块

Question

我正在从几个设备收集数据，由于测试持续时间很长，我想使用Python的threading和Queue模块。我已经写了一个简短的脚本来弄清楚如何使用它们，这很明显，我不明白让这个工作起作用的细微差别。

这里是我的脚本：

import ue9 
import LJ_Util 
import DAQ_Util 
import threading 
import Queue 

from datetime import datetime 
from time import sleep 

queue = Queue.Queue() 
now = datetime.now().isoformat() 

def DAQThread(ue9ipAddr): 
    print '\nExecuting in DAQThread at IP Address: %s' % ue9ipAddr 
    a = ue9.UE9(ethernet=True, ipAddress=ue9ipAddr) 
    SN = (a.commConfig()).get('SerialNumber') 
    count = 5 
    while count > 0: 
     reading = a.feedback() 
     dataReturn = (SN, now, reading) 
     queue.put(dataReturn) 
     count -= 1 
     print count 
     sleep(5) 


def listenThread(counter): 
    while queue.empty() != True: 
     try: 
      outcome = queue.get() 
      print outcome 
      counter -=1 
      print counter 
     except: 
      return 'queue.get() command loop failing.' 


print "\nOpening device..." 
ipAdd = '192.168.1.127' 
feedbackThread = threading.Thread(target=DAQThread, args=(ipAdd,)) 
feedbackThread.start() 

print "\nListening for data..." 
queryThread = threading.Thread(target=listenThread, args = (10,)) 
queryThread.start() 

print queue.get() 
print(threading.activeCount()) 
print "\nDone" 

下面是执行这个脚本的输出结果：

$ python threading-and-queue-test.py 

Opening device... 

Executing in DAQThread at IP Address: 192.168.1.127 
Listening for data... 

4 
(278956853, '2012-09-03T20:02:47.656024', {'AIN4': -0.012, 'AIN5': -0.012, 'CIODir': 0, 'AIN7': -0.012, 'EIODir': 0, 'AIN1': -0.012, 'AIN2': -0.012, 'AIN3': -0.012, 'MIOState': 7, 'AIN8': -0.012, 'AIN6': -0.012, 'AIN9': -0.012, 'CIOState': 15, 'AIN0': -0.012, 'Counter0': 0, 'Counter1': 0, 'EIOState': 255, 'TimerC': 0, 'TimerB': 0, 'TimerA': 0, 'MIODir': 0, 'FIODir': 32, 'AIN14': -0.012, 'AIN15': -0.012, 'AIN12': -0.012, 'AIN13': -0.012, 'AIN10': -0.012, 'AIN11': -0.012, 'FIOState': 255}) 
2 

Done 
3 
2 
1 
0 
$ 

很明显线程活动的时间安排被“关闭”，但我不知道如何解决它，因为我以前从未使用这些模块进行编程，也没有使用过的线程。任何意见或建议都会受到欢迎。 在此先感谢！

Answer 1

正如在评论中提到的，一个问题是在您的聆听线程中。一旦从队列中“抓取”了一个条目，队列中就不再有任何条目，并且由于您每隔5秒只添加一个条目，因此您的侦听线程将清空队列，因此

while queue.empty() != True将评估False，退出循环

例如：

>>> import Queue 
>>> q=Queue.Queue() 
>>> q.put(1) 
>>> q.empty() 
False 
>>> q.get() 
1 
>>> q.empty() 
True 
>>> q.empty()!=True 
False 

一个办法来解决这个问题是使用另一个队列为停止或取消队列，以便修改你听线程你可以做这样的事情：

stopQue=Queue.Queue() 

def listenThread(counter): 
    while True: 
     if queue.empty()!=True: 
      outcome = queue.get() 
      print outcome 
      counter -=1 
      print counter 
     if stopQue.empty()!=True: 
      break 
    print 'Exiting Listening Thread' 

这样，如果你把任何东西放在stopQue，即stopQue.put(1)，它应该退出。

完整的示例基于您的初始代码。我切出不具有与队列和线程做代码：

import threading 
import Queue 
from time import sleep 

dataQue = Queue.Queue() 
stopQue = Queue.Queue() 

def DAQThread(ue9ipAddr): 
    print 'Executing in DAQThread\n' 
    count = 5 
    while count > 0: 
     dataQue.put('data: %s' % count) 
     count -= 1 
     sleep(5) 
    stopQue.put(1) 
    print 'Exiting DAQThread\n' 


def listenThread(counter): 
    while True: 
     if dataQue.empty() != True: 
      outcome = dataQue.get() 
      print outcome 
      counter -=1 
     if stopQue.empty() != True: 
      break 
    print 'Exiting Listening Thread' 


print "Opening device..." 
ipAdd = '192.168.1.127' 
feedbackThread = threading.Thread(target=DAQThread, args=(ipAdd,)) 
feedbackThread.setDaemon(True) 
feedbackThread.start() 

print "Listening for data..." 
queryThread = threading.Thread(target=listenThread, args = (10,)) 
queryThread.setDaemon(True) 
queryThread.start() 

print "Done" 

产生输出：

>>> 
Opening device... 
Executing in DAQThread 
Listening for data... 

Donedata: 5  #Notice how the script is done however the threads are still running 

>>> data: 4 
data: 3 
data: 2 
data: 1 
Exiting DAQThread 
Exiting Listening Thread