pyqt5线程与弹窗示例

import datetime import json import sys import time from PyQt5.QtCore import QThread, pyqtSignal, QMutex, QWaitCondition from PyQt5.QtWidgets import QMainWindow, QApplication, QMessageBox from thread_demo import Ui_MainWindow class Window(QMainWindow, Ui_MainWindow): def __init__(self): super(QMainWindow, self).__init__() self.setup_ui() # 渲染画布 self.product_thread = ProductThread(self) # 开启生产者线程 self.consumer_thread = ConsumerThread(self) # 开启消费者线程 self.connect_signals() # 绑定触发事件 self.producer_data_list = [] # 生产者数据显示列表 self.consumer_data_list = [] # 消费者数据显示列表 self.status = 'init' # 初始状态 self.title.setReadOnly(True) # 标题设置成只读 self.set_lamp_color(self.status) def setup_ui(self): self.setupUi(self) # 渲染pyqt5界面 # 设置只读，背景色为灰色 style = 'background: #D3D3D3' self.producer_data.setStyleSheet(style) self.producer_data.setReadOnly(True) self.consumer_data.setStyleSheet(style) self.consumer_data.setReadOnly(True) def connect_signals(self): # 绑定触发事件 self.btn_start.clicked.connect(self.btn_start_clicked) # 开始按钮点击事件 self.btn_pause.clicked.connect(self.btn_pause_clicked) # 暂停按钮点击事件 self.product_thread._signal_product.connect(self.product_threading_slot) # 生产者线程回调函数 self.consumer_thread._signal_consumer.connect(self.consumer_threading_slot) # 消费者线程回调函数 def btn_start_clicked(self): # 开始按钮 if self.status == '0': return self.status = '0' self.lamp.setText(' 运行') self.set_lamp_color(self.status) # 设置指示灯为绿色 if self.product_thread.is_pause: # 重新启动生产者线程 self.product_thread.cond.wakeAll() else: self.product_thread.start() self.product_thread.is_pause = False if self.consumer_thread.is_pause: # 重新启动消费者线程 self.consumer_thread.cond.wakeAll() else: self.consumer_thread.start() self.consumer_thread.is_pause = False def btn_pause_clicked(self): # 暂停按钮 if self.status != '0': return # 先弹出窗口确认 select = QMessageBox.warning(self, "暂停线程程序", "确定要暂停程序吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes) if select == QMessageBox.Yes: self.status = '1' self.lamp.setText(' 暂停') self.set_lamp_color(self.status) self.product_thread.is_pause = True # 生产者线程进入暂停状态 self.consumer_thread.is_pause = True # 消费者线程进入暂停状态 def set_lamp_color(self, status): # 设置指示灯颜色 初始为灰色， 运行为绿色，暂停为红色 color = {'init': '#0B610B', '0': '#9ACD32', '1': '#FF4000'}[status] style = """min-width: 44px; min-height: 44px; max-width:44px; max-height: 44px; border-radius: 22px; border:1px solid black; background:{}; font-size:14px; color:white """.format(color) self.lamp.setStyleSheet(style) def product_threading_slot(self, data): # 生产者回调函数 data = json.loads(data) if 'product_data' in data: self.producer_data_list.append(data['product_data']) self.producer_data.clear() # 先清空显示列表 for d in self.producer_data_list: self.producer_data.append(d) # 把生产者数据显示出来 def consumer_threading_slot(self, data): # 消费者回调函数 data = json.loads(data) if 'consumer_data' in data: self.consumer_data_list.append(data['consumer_data']) if len(self.consumer_data_list) > 13: self.consumer_data_list.pop(0) # 消费者数据显示列表满13条就出栈 self.consumer_data.clear() # 先清空显示列表 for d in self.consumer_data_list: self.consumer_data.append(d) # 把消费者数据显示出来 # 生产者线程 class ProductThread(QThread): _signal_product = pyqtSignal(str) def __init__(self, parent=None): ''' QWaitCondition()用于多线程同步，一个线程调用QWaitCondition.wait()阻塞等待， 直到另外一个线程调用QWaitCondition.wake()唤醒才继续往下执行 QMutex()：是锁对象 :param parent: ''' super(ProductThread, self).__init__(parent) self.window = parent self.cond = QWaitCondition() self.qmut = QMutex() # 线程锁 self.is_pause = False # 信号量 def run(self): count = 1 while 1: self.qmut.lock() if self.is_pause: self.cond.wait(self.qmut) # 线程挂起 self.qmut.unlock() now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 发送信号给槽函数，需要传送的是字符串 self._signal_product.emit(json.dumps({'product_data': '{} 数据{}'.format(now, count)})) count += 1 time.sleep(0.8) # 消费者线程 class ConsumerThread(QThread): _signal_consumer = pyqtSignal(str) def __init__(self, parent=None): ''' QWaitCondition()用于多线程同步，一个线程调用QWaitCondition.wait()阻塞等待， 直到另外一个线程调用QWaitCondition.wake()唤醒才继续往下执行 QMutex()：是锁对象 :param parent: ''' super(ConsumerThread, self).__init__(parent) self.window = parent self.cond = QWaitCondition() self.qmut = QMutex() # 线程锁 self.is_pause = False # 信号量 def run(self): while 1: self.qmut.lock() if self.is_pause: self.cond.wait(self.qmut) # 线程挂起 self.qmut.unlock() if self.window.producer_data_list: # 发送信号给槽函数，需要传送的是字符串 self._signal_consumer.emit(json.dumps({'consumer_data': '消费了{}'.format(self.window.producer_data_list.pop(0))})) time.sleep(0.1) else: time.sleep(1) if __name__ == '__main__': app = QApplication(sys.argv) mywindow = Window() mywindow.show() sys.exit(app.exec_())