我有大量文件要处理。我写了一个脚本来获取、排序和绘制我想要的数据。到目前为止，一切都很好。我已经对其进行了测试，它给出了预期的结果。
然后我想用多线程来做到这一点。我查看了互联网上的文档和示例，在我的程序中使用一个线程工作正常。但是当我使用更多时，在某些时候我会遇到随机的 matplotlib 错误，我怀疑那里存在一些冲突，即使我使用了一个带有绘图名称的函数，我也看不出问题出在哪里。
这是整个脚本，如果您需要更多评论，我会添加它们。谢谢。
#!/usr/bin/python
import matplotlib
matplotlib.use('GTKAgg')
import numpy as np
from scipy.interpolate import griddata
import matplotlib.pyplot as plt
import matplotlib.colors as mcl
from matplotlib import rc #for latex
import time as tm
import sys
import threading
import Queue #queue in 3.2 and Queue in 2.7 !
import pdb #the debugger
rc('text', usetex=True)#for latex
map=0 #initialize the map index. It will be use to index the array like this: array[map,[x,y]]
time=np.zeros(1) #an array to store the time
middle_h=np.zeros((0,3)) #x phi c
#for the middle of the box
current_file=open("single_void_cyl_periodic_phi_c_middle_h_out",'r')
for line in current_file:
if line.startswith('# === time'):
map+=1
np.append(time,[float(line.strip('# === time '))])
elif line.startswith('#'):
pass
else:
v=np.fromstring(line,dtype=float,sep=' ')
middle_h=np.vstack( (middle_h,v[[1,3,4]]) )
current_file.close()
middle_h=middle_h.reshape((map,-1,3)) #3d array: map, x, phi,c
#####
def load_and_plot(): #will load a map file, and plot it along with the corresponding profile loaded before
while not exit_flag:
print("fecthing work ...")
#try:
if not tasks_queue.empty():
map_index=tasks_queue.get()
print("----> working on map: %s" %map_index)
x,y,zp=np.loadtxt("single_void_cyl_growth_periodic_post_map_"+str(map_index),unpack=True, usecols=[1, 2,3])
for i,el in enumerate(zp):
if el<0.:
zp[i]=0.
xv=np.unique(x)
yv=np.unique(y)
X,Y= np.meshgrid(xv,yv)
Z = griddata((x, y), zp, (X, Y),method='nearest')
figure=plt.figure(num=map_index,figsize=(14, 8))
ax1=plt.subplot2grid((2,2),(0,0))
ax1.plot(middle_h[map_index,:,0],middle_h[map_index,:,1],'*b')
ax1.grid(True)
ax1.axis([-15, 15, 0, 1])
ax1.set_title('Profiles')
ax1.set_ylabel(r'$\phi$')
ax1.set_xlabel('x')
ax2=plt.subplot2grid((2,2),(1,0))
ax2.plot(middle_h[map_index,:,0],middle_h[map_index,:,2],'*r')
ax2.grid(True)
ax2.axis([-15, 15, 0, 1])
ax2.set_ylabel('c')
ax2.set_xlabel('x')
ax3=plt.subplot2grid((2,2),(0,1),rowspan=2,aspect='equal')
sub_contour=ax3.contourf(X,Y,Z,np.linspace(0,1,11),vmin=0.)
figure.colorbar(sub_contour,ax=ax3)
figure.savefig('single_void_cyl_'+str(map_index)+'.png')
plt.close(map_index)
tasks_queue.task_done()
else:
print("nothing left to do, other threads finishing,sleeping 2 seconds...")
tm.sleep(2)
# except:
# print("failed this time: %s" %map_index+". Sleeping 2 seconds")
# tm.sleep(2)
#####
exit_flag=0
nb_threads=2
tasks_queue=Queue.Queue()
threads_list=[]
jobs=list(range(map)) #each job is composed of a map
print("inserting jobs in the queue...")
for job in jobs:
tasks_queue.put(job)
print("done")
#launch the threads
for i in range(nb_threads):
working_bee=threading.Thread(target=load_and_plot)
working_bee.daemon=True
print("starting thread "+str(i)+' ...')
threads_list.append(working_bee)
working_bee.start()
#wait for all tasks to be treated
tasks_queue.join()
#flip the flag, so the threads know it's time to stop
exit_flag=1
for t in threads_list:
print("waiting for threads %s to stop..."%t)
t.join()
print("all threads stopped")
请您参考如下方法:
按照 David 的建议，我在 multiprocessing 中做了。我用 8 个处理器获得了 5 倍的速度。我相信剩下的就是在我的脚本开始时对单进程工作的处理。 编辑:然而，有时脚本在最后一张 map 上“挂起”，即使它生成了正确的 map ，并出现以下错误:
文件“single_void_cyl_plot_mprocess.py”，第 90 行，在 tasks_queue.join() 中
文件“/usr/local/epd-7.0-2-rh5-x86_64/lib/python2.7/multiprocessing/queues.py”，第316行，加入self._cond.wait()
文件“/usr/local/epd-7.0-2-rh5-x86_64/lib/python2.7/multiprocessing/synchronize.py”，第220行，等待self._wait_semaphore.acquire(True, timeout)
import numpy as np
from scipy.interpolate import griddata
import matplotlib.pyplot as plt
from matplotlib import rc #for latex
from multiprocessing import Process, JoinableQueue
import pdb #the debugger
rc('text', usetex=True)#for latex
map=0 #initialize the map index. It will be use to index the array like this: array[map,x,y,...]
time=np.zeros(1) #an array to store the time
middle_h=np.zeros((0,3)) #x phi c
#for the middle of the box
current_file=open("single_void_cyl_periodic_phi_c_middle_h_out",'r')
for line in current_file.readlines():
if line.startswith('# === time'):
map+=1
np.append(time,[float(line.strip('# === time '))])
elif line.startswith('#'):
pass
else:
v=np.fromstring(line,dtype=float,sep=' ')
middle_h=np.vstack( (middle_h,v[[1,3,4]]) )
current_file.close()
middle_h=middle_h.reshape((map,-1,3)) #3d array: map, x, phi,c
#######
def load_and_plot(): #will load a map file, and plot it along with the corresponding profile loaded before
while tasks_queue.empty()==False:
print("fecthing work ...")
try:
map_index=tasks_queue.get() #get some work to do from the queue
print("----> working on map: %s" %map_index)
x,y,zp=np.loadtxt("single_void_cyl_growth_periodic_post_map_"+str(map_index),\
unpack=True, usecols=[1, 2,3])
for i,el in enumerate(zp):
if el<0.:
zp[i]=0.
xv=np.unique(x)
yv=np.unique(y)
X,Y= np.meshgrid(xv,yv)
Z = griddata((x, y), zp, (X, Y),method='nearest')
figure=plt.figure(num=map_index,figsize=(14, 8))
ax1=plt.subplot2grid((2,2),(0,0))
ax1.plot(middle_h[map_index,:,0],middle_h[map_index,:,1],'*b')
ax1.grid(True)
ax1.axis([-15, 15, 0, 1])
ax1.set_title('Profiles')
ax1.set_ylabel(r'$\phi$')
ax1.set_xlabel('x')
ax2=plt.subplot2grid((2,2),(1,0))
ax2.plot(middle_h[map_index,:,0],middle_h[map_index,:,2],'*r')
ax2.grid(True)
ax2.axis([-15, 15, 0, 1])
ax2.set_ylabel('c')
ax2.set_xlabel('x')
ax3=plt.subplot2grid((2,2), (0,1),rowspan=2,aspect='equal')
sub_contour=ax3.contourf(X,Y,Z,np.linspace(0,1,11),vmin=0.)
figure.colorbar(sub_contour,ax=ax3)
figure.savefig('single_void_cyl_'+str(map_index)+'.png')
plt.close(map_index)
tasks_queue.task_done() #work for this item finished
except:
print("failed this time: %s" %map_index)
#######
nb_proc=8 #number of processes
tasks_queue=JoinableQueue() #a queue to pile up the work to do
jobs=list(range(map)) #each job is composed of a map
print("inserting jobs in the queue...")
for job in jobs:
tasks_queue.put(job)
print("done")
#launch the processes
for i in range(nb_proc):
current_process=Process(target=load_and_plot)
current_process.start()
#wait for all tasks to be treated
tasks_queue.join()
