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_())