pyqt5实现线程与弹窗功能
效果图：
示例下载
点我下载
https://download.csdn.net/download/lm_is_dc/87982279
简介
Pyqt5线程使用 QThread, pyqtSignal, QMutex, QWaitCondition来实现，涉及到线程，锁，信号量，线程挂起，线程唤醒。
本文实现流程如下：
1、先设计UI；
2、把UI转py;
3、编写主函数
4、主函数流程：
4.1、创建QT窗口类、创建生产者类、消费者类
4.2、渲染画布、开启生产者线程、消费者线程
4.3、绑定触发事件，按钮点击、线程回调
4.4、标题设置成只读、设置指示灯颜色
4.5、点击开始按钮启动线程
4.6、点击暂停按钮，先弹出窗口，选择YES则挂起线程
1、UI设计
使用pycharm创建一个pyqt5项目，命名为QTThread_demo,界面如下：
2、ui转py
使用pyuic5把ui转成py文件，如图：
转成的thread_demo.py代码如下：
# -*- coding: utf-8 -*-
# Form implementation generated from reading ui file 'thread_demo.ui'
#
# Created by: PyQt5 UI code generator 5.15.4
#
# WARNING: Any manual changes made to this file will be lost when pyuic5 is
# run again. Do not edit this file unless you know what you are doing.
from PyQt5 import QtCore, QtGui, QtWidgets
class Ui_MainWindow(object):
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(800, 600)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.title = QtWidgets.QTextEdit(self.centralwidget)
self.title.setGeometry(QtCore.QRect(210, 0, 351, 51))
self.title.setObjectName("title")
self.producer_data = QtWidgets.QTextEdit(self.centralwidget)
self.producer_data.setGeometry(QtCore.QRect(30, 160, 341, 251))
self.producer_data.setObjectName("producer_data")
self.label = QtWidgets.QLabel(self.centralwidget)
self.label.setGeometry(QtCore.QRect(30, 140, 72, 15))
self.label.setObjectName("label")
self.label_2 = QtWidgets.QLabel(self.centralwidget)
self.label_2.setGeometry(QtCore.QRect(400, 140, 72, 15))
self.label_2.setObjectName("label_2")
self.consumer_data = QtWidgets.QTextEdit(self.centralwidget)
self.consumer_data.setGeometry(QtCore.QRect(400, 160, 341, 251))
self.consumer_data.setObjectName("consumer_data")
self.btn_start = QtWidgets.QPushButton(self.centralwidget)
self.btn_start.setGeometry(QtCore.QRect(30, 80, 93, 28))
self.btn_start.setObjectName("btn_start")
self.btn_pause = QtWidgets.QPushButton(self.centralwidget)
self.btn_pause.setGeometry(QtCore.QRect(180, 80, 93, 28))
self.btn_pause.setObjectName("btn_pause")
self.lamp = QtWidgets.QLabel(self.centralwidget)
self.lamp.setGeometry(QtCore.QRect(400, 70, 72, 31))
self.lamp.setObjectName("lamp")
MainWindow.setCentralWidget(self.centralwidget)
self.menubar = QtWidgets.QMenuBar(MainWindow)
self.menubar.setGeometry(QtCore.QRect(0, 0, 800, 27))
self.menubar.setObjectName("menubar")
MainWindow.setMenuBar(self.menubar)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.title.setHtml(_translate("MainWindow", "<!DOCTYPE HTML PUBLIC \"-//W3C//DTD HTML 4.0//EN\" \"http://www.w3.org/TR/REC-html40/strict.dtd\">\n"
"<html><head><meta name=\"qrichtext\" content=\"1\" /><style type=\"text/css\">\n"
"p, li { white-space: pre-wrap; }\n"
"</style></head><body style=\" font-family:\'SimSun\'; font-size:9pt; font-weight:400; font-style:normal;\">\n"
"<p align=\"center\" style=\" margin-top:0px; margin-bottom:0px; margin-left:0px; margin-right:0px; -qt-block-indent:0; text-indent:0px;\"><span style=\" font-size:18pt; font-weight:600; color:#0055ff;\">pyqt5线程与弹窗</span></p></body></html>"))
self.label.setText(_translate("MainWindow", "生产者"))
self.label_2.setText(_translate("MainWindow", "消费者"))
self.btn_start.setText(_translate("MainWindow", "开始"))
self.btn_pause.setText(_translate("MainWindow", "暂停"))
self.lamp.setText(_translate("MainWindow", "指示灯"))
3、主程序
主程序main.py如下:
# !/usr/bin/python
# -*- coding: utf-8 -*-
"""
@contact: 微信 1257309054
@file: main.py
@time: 2023/7/1 23:55
@author: LDC
"""
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_())
4、运行效果图
运行：
暂停：