我要评论测试准备
在 pyqt 中使用多进程比多线程复杂，因为多进程之间没有共享内存，不能用qt的信号槽通信。但采用python的multiprocessing和queue，其实也没复杂到拿去。
为了体现出多进程工作的特点，接下来实现如下功能，即4个进程分别生成数据，分别对应图中的一条线。
本测试与pyqt+matplotlib多线程绘图采用了相同的ui，由于实现异步的方式不同，故而import的模块稍有差异
import numpy as np import multiprocessing as mp from pyqt6.qtcore import qtimer import sys import numpy as np from pyqt6 import uic from pyqt6.qtwidgets import qapplication, qmainwindow, qhboxlayout, qwidget from matplotlib.backends.backend_qtagg import navigationtoolbar2qt as navigationtoolbar
测试函数
为pyqt的多进程绘图软件编写如下测试函数，三个输入参数，分别是任务id，多进程消息队列以及绘图点数。
def heavy_compute(task_id, q, n):
xs, ys = [], []
freq = (task_id + 1) * 0.5
for i in range(n):
arr = np.random.rand(50000)
res = np.sum(np.sin(arr * freq) * np.cos(arr))
xs.append(i)
ys.append(float(res))
if i % 100 == 0:
q.put((task_id, xs.copy(), ys.copy()))
q.put(("done", task_id))
主窗口
主窗口代码如下。和基于qthread与pyqtsignal的多线程方案相比，主要区别在于采用了mp.queue()作为进程间数据传输的工具。由于缺少pyqtsignal这样便捷的事件触发绑定工具，主窗口在绘图时采用了timer轮询的方案，即不断查询queue是否返回了数据，如果返回，就画图，否则继续循环。
class mainwindow(qmainwindow):
def __init__(self):
super().__init__()
uic.loadui("thtest.ui", self)
self.toolbar_container = self.findchild(qwidget, "toolbar_container")
layout = self.toolbar_container.layout()
if layout is none:
layout = qhboxlayout(self.toolbar_container)
layout.setcontentsmargins(0, 0, 0, 0)
layout.setspacing(0)
self.toolbar = navigationtoolbar(self.mpl_canvas, self)
layout.addwidget(self.toolbar)
self.num_tasks = 4 # 要同时运行的进程数
self.steps = 1000 # 每个进程的计算总步数
self.colors = ['#e6194b', '#3cb44b', '#4363d8', '#f58231'] # 4种颜色
self.isrunning = false
self.processes = []
self.queue = mp.queue()
self.finished_count = 0
self.timer = qtimer()
self.timer.timeout.connect(self._poll_queue)
self.ax = self.mpl_canvas.axes
self.ax.clear()
self.lines = {}
for i in range(self.num_tasks):
line, = self.ax.plot([], [], color=self.colors[i], linewidth=2, label=f'process {i}')
self.lines[i] = line
self.ax.legend(loc='upper left')
self.ax.set_title("multiprocessing pool: real-time multi-curve")
self.mpl_canvas.draw()
self.pbdraw.clicked.connect(self._toggle_task)
def _toggle_task(self):
if not self.isrunning:
self._start_pool()
else:
self._stop_pool()
def _start_pool(self):
self.isrunning = false
self.finished_count = 0
self.pbdraw.settext("停止/等待完成")
self.processes = []
for i in range(self.num_tasks):
p = mp.process(target=heavy_compute, args=(i, self.queue, self.steps), daemon=true)
p.start()
self.processes.append(p)
self.isrunning = true
self.timer.start(50)
def _stop_pool(self):
self.pbdraw.settext("正在等待子进程结束...")
self.pbdraw.setenabled(false)
def _poll_queue(self):
has_update = false
while true:
try:
msg = self.queue.get_nowait()
except exception:
break # 队列空了，退出本次轮询
if msg[0] == "done":
self.finished_count += 1
if self.finished_count >= self.num_tasks:
self._on_all_finished()
else:
# 正常的数据更新
task_id, x, y = msg
# 根据 task_id 找到对应的线条并更新数据
self.lines[task_id].set_data(x, y)
has_update = true
# 只要有新数据，就刷新一次画布 (避免频繁刷新)
if has_update:
self.ax.relim()
self.ax.autoscale_view()
self.mpl_canvas.draw_idle()
def _on_all_finished(self):
self.timer.stop()
for p in self.processes:
p.join() # 等待所有子进程结束
self.isrunning = false
self.pbdraw.setenabled(true)
self.pbdraw.settext("启动计算")
print(">>> 所有进程池任务已执行完毕！")
到此这篇关于pyqt+matplotlib实现多进程图像刷新的文章就介绍到这了,更多相关pyqt matplotlib多进程图像刷新内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论