我正在尝试使用 matplotlib 以有意义的方式绘制多线程代码的线程。 我希望每个线程都用一种颜色可视化。这样,绘图将清楚地显示哪个线程执行了哪些任务等。 所以为了清楚起见,我想说(下图)黄色条是线程 1 执行的进程,红色条是线程 2 执行的进程,蓝色条是线程 3 执行的进程。
这似乎很难,我能想到的最好的是在下面(见图和代码)。 在示例中,我们有 3 个线程和 12 个任务(每个任务的持续时间在某个点上是可变的)。线程 1 以黄色开始,线程 2 以红色开始,线程 3 以蓝色开始。我很乐意在整个图表中像这样对它们进行颜色编码。然而,我们看到的是线程 1 执行任务 0、7 和 10,但它会将颜色从黄色切换为红色再切换为红色。线程 2 相同:它执行任务 2、5、8 和 11,但将颜色从红色切换为蓝色,再从蓝色切换为蓝色。线程 3 相同。 所以颜色实际上以 3 的循环出现,并且与线程无关。正如我之前所说,我想让它们依赖于线程号,以使多线程图更有意义(因为目前还没有)。
有人知道如何做到这一点吗?
import threading
import multiprocessing
import math
import numpy as np
import time
import matplotlib.pyplot as plt
import glob
from PIL import Image
import random
from random import sample
import string
from concurrent.futures import ThreadPoolExecutor
cpu_workers = 3
nSim = 12
def generate_bar_colors(cpu_workers):
colors = ['red', 'gold', 'royalblue']
return colors
def visualize_runtimes(results, title):
colors = generate_bar_colors(cpu_workers)
plt.rcParams["font.family"] = "Times New Roman"
plt.rcParams['axes.axisbelow'] = True
start,stop = np.array(results).T
plt.barh(range(len(start)),stop-start,left=start, color=colors)
plt.grid(axis='x', color= 'lightgrey')
plt.title("Tasks", rotation='horizontal', fontsize=12, horizontalalignment="left", x=0)
plt.xlabel("Seconds", fontsize=12, horizontalalignment='right', x=1.0)
def multithreading(func, args, workers):
begin_time=time.time()
with ThreadPoolExecutor(max_workers = workers) as executor:
res = executor.map(func, args, [begin_time for i in range (len(args))])
return list(res)
def simulation(i, base):
start = time.time() - base
print(str(threading.current_thread().getName()) + ': '+ str(i))
time.sleep(math.cos(i)+i*0.1+1)
stop = time.time() - base
return start, stop
if __name__ == '__main__':
visualize_runtimes(multithreading(simulation, i, cpu_workers), "Multi-threading")
plt.savefig('foo.png', bbox_inches='tight')
plt.show()
最佳答案
实现此目的的一种方法(参见下面的代码)。现在每个线程都分配了颜色,很明显计算是多线程的(在这种情况下,3 个线程正在执行 12 个任务)。
import threading
import math
from matplotlib.lines import Line2D
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor
import matplotlib.pyplot as plt
from matplotlib.ticker import AutoMinorLocator, MultipleLocator, FuncFormatter
import time
import numpy as np
cpu_workers = 3
nSim = 12
i = range(nSim)
#Multi-threading function ----------------------------------------------------------------------------------------------
def multithreading(func, args, workers):
with ThreadPoolExecutor(max_workers = workers) as executor:
responses = executor.map(func, args)
return list(responses)
#List of unique arguments in preserved order as in my_list -------------------------------------------------------------
def pres_uniq_list(my_list): #returns a unique list in preserved order
seen = set()
result = []
for e in my_list:
if e not in seen:
result.append(e)
seen.add(e)
return result
#Get netto simulation start- and end-times as well as duration ---------------------------------------------------------
def sep_list_elements(list_of_lists, proc_start_t):
start_values = [inner_list[0] for inner_list in list_of_lists]
start_values = np.array(start_values) - proc_start_t
end_values = [inner_list[1] for inner_list in list_of_lists]
end_values = np.array(end_values) - proc_start_t
return start_values, end_values
#Match colors with threads (one color per thread) ----------------------------------------------------------------------
def thread_colors(list_of_lists):
thread_ids = [inner_list[2] for inner_list in list_of_lists]
color_guide = ['red', 'royalblue', 'gold', 'darkgray', 'forestgreen', 'orangered', 'lightpink', 'teal']
lookup = dict(zip(pres_uniq_list(thread_ids), color_guide))
colors = [lookup[number] for number in thread_ids]
return colors
#Graph legend to match with bars ---------------------------------------------------------------------------------------
def thread_legend(list_of_lists):
thread_list = [inner_list[3] for inner_list in list_of_lists]
color_guide = ['red', 'royalblue', 'gold', 'darkgray', 'forestgreen', 'orangered', 'lightpink', 'teal']
lookup = dict(zip(pres_uniq_list(thread_list), color_guide))
colors = [lookup[number] for number in thread_list]
thread_legend = ([Line2D([0], [0], color=c, alpha=0.4, lw=4) for c in pres_uniq_list(colors)])
names = [str('Thread-')+str(i) for i in range(cpu_workers)]
return thread_legend, names
#Graph definition using MatPlotLib -------------------------------------------------------------------------------------
def graph_settings(start_t, end_t, title, colors, legend):
plt.rcParams["font.family"] = "Times New Roman"
plt.rcParams["font.size"] = 10
plt.rcParams['axes.axisbelow'] = True
plt.rcParams['axes.edgecolor'] = 'black'
plt.rcParams['axes.linewidth'] = 0.8
plt.rcParams['xtick.color'] = 'black'
plt.rcParams['ytick.color'] = 'black'
#plt.rcParams['font.weight']= 'heavy'
fig = plt.figure(figsize=((12.0/2.54), (7.42/2.54)), facecolor='w', edgecolor='black') #set (12,7.42) gulden snede but figsize is in inches
fig, ax = plt.subplots()
#ax.xaxis.set_major_locator(MultipleLocator(1.000))
#ax.xaxis.set_minor_locator(AutoMinorLocator(4))
#ax.yaxis.set_major_locator(MultipleLocator(1.000))
#ax.yaxis.set_minor_locator(AutoMinorLocator(4))
ax.set_xlim(-0.2, end_t[nSim-1])
ax.set_ylim(-1, nSim)
ax.spines['top'].set_visible(False) #to set color: ax.spines['top'].set_color('green')
ax.spines['right'].set_visible(False)
ax.spines['left'].set_smart_bounds(True)
ax.spines['bottom'].set_smart_bounds(True)
#ax.tick_params(which='major')
#ax.tick_params(which='major', length=5)
#ax.tick_params(which='minor', labelsize=10)
#ax.tick_params(which='minor', length=2.5, labelsize=10, labelcolor='0.25')
ax.margins(0.02) #margins in %, default is 5%
ax.barh(range(len(start_t)), end_t-start_t, left=start_t, color=colors, alpha=0.4, edgecolor="none") #alpha adds vagueness to the bars
ax.barh(range(len(start_t)), end_t-end_t+0.00001, left=end_t, color=colors, alpha=1, edgecolor="none") #alpha adds vagueness to the bars
#w = pres_uniq_list(p.get_height() for p in ax.patches)
#w_fl = float(i) for i in w]
#ax.plot(end_t, range(nSim), linewidth=0, marker="o", markersize=w[0], color="grey")
#print(end_t)
#plt.grid(axis='x', color= 'lightgrey')
#plt.ylabel("Tasks", fontsize=12, horizontalalignment='left', y=1.02, rotation='horizontal')
ax.set_title(title, fontsize=14, fontweight="bold", horizontalalignment='center', y=1.04) #set title instead of y-label, bold does not work
ax.annotate("Tasks", fontsize=12, xy=(0, 1), xytext=(0, 10), xycoords="axes fraction", textcoords="offset points", ha="right", ) #set ha="left" to have it above the axis
ax.annotate(("Tasks: " + str(nSim) + " - Threads: " + str(cpu_workers)), fontsize=10, fontstyle='italic', xy=(0.5, 1), xytext=(0, 4), xycoords="axes fraction", textcoords="offset points", ha="center", color='gray' ) #set ha="left" to have it above the axis
ax.set_xlabel("Time [s]", fontsize=12, horizontalalignment='right', x=1.0)
#ax.set_suptitle(title, fontsize=16, fontweight='bold', y=1.005)
thread_legend = reversed(legend[0])
names = reversed(legend[1])
leg = ax.legend(thread_legend, names, fancybox=True, loc='lower right', fontsize=10, edgecolor=None)
#plt.legend(frameon=False)
leg.get_frame().set_facecolor('none')
leg.get_frame().set_edgecolor('none')
plt.savefig('P1_5test.pdf', bbox_inches='tight') #dpi=300 when png , bbox_inches='tight'
plt.show()
#plt.figure(figsize=(12, 10))
#plt.savefig('foo1.pdf', , dpi = 300)
return None
#Run code for if __name__ == '__main__' that links all functions -------------------------------------------------------
def plot_runtimes(workers, func, args, title):
proc_start_t = time.time() # we track time from here
runtimes = multithreading(func, args, workers) #collect task runtimes from function
#net_runtimes = net_start_t(proc_start_t, runtimes) #extract net_runtimes from runtimes
start_t, end_t = sep_list_elements(runtimes, proc_start_t) #seperate start_t and end_t
colors = thread_colors(runtimes) #select thread colors
legend = thread_legend(runtimes) #define thread legend
graph_settings(start_t, end_t, title, colors, legend) #settings for the horizontal bar charts
#Tasks to perform ------------------------------------------------------------------------------------------------------
def simulation(i):
#i = range(nSim)
start_t = time.time()
#print(str(i) + ': start')
a = str(threading.current_thread().ident)
b = str(threading.current_thread().getName())
print(str(threading.current_thread().getName()) + ': '+ str(i))
#print(str(i) + ': finish')
time.sleep(math.cos(i)+i*0.1+1)
end_t = time.time()
return [start_t, end_t, a, b]
#return a
#Definition ------------------------------------------------------------------------------------------------------------
if __name__ == '__main__':
plot_runtimes(workers = cpu_workers,
func = simulation,
args = i,
title = "Multi-threading")
关于python - 如何在多线程任务/时间图中可视化线程?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58369194/