【100天精通python】Day38：GUI界面编程_PyQt 从入门到实战（中）_数据库操作与多线程编程

from PyQt6.QtSql import QSqlQuery
query = QSqlQuery()
query.exec("SELECT * FROM employees")
while query.next():
    name = query.value("name")
    print("Employee name:", name)

from PyQt6.QtCore import QThread, pyqtSignal
class WorkerThread(QThread):
    result_ready = pyqtSignal(str)
    def run(self):
        # 执行耗时任务
        result = "Task result"
        self.result_ready.emit(result)
thread = WorkerThread()
thread.result_ready.connect(lambda result: print("Result:", result))
thread.start()

import sys
from PyQt6.QtCore import QThread, pyqtSignal
from PyQt6.QtWidgets import QApplication, QMainWindow, QPushButton
class WorkerThread(QThread):
    result_ready = pyqtSignal(str)
    def run(self):
        result = "Task result"
        self.result_ready.emit(result)
class MyWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Thread Communication Example")
        self.setGeometry(100, 100, 400, 300)
        self.button = QPushButton("Start Task", self)
        self.button.setGeometry(150, 150, 100, 30)
        self.button.clicked.connect(self.start_thread)
    def start_thread(self):
        self.thread = WorkerThread()
        self.thread.result_ready.connect(self.handle_result)
        self.thread.start()
    def handle_result(self, result):
        print("Result:", result)
if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyWindow()
    window.show()
    sys.exit(app.exec())

import sys
from PyQt6.QtCore import QThread, QMutex, QMutexLocker
class SharedData:
    def __init__(self):
        self.mutex = QMutex()  # 用于保护共享数据的互斥锁
        self.data = 0
    def increment(self):
        locker = QMutexLocker(self.mutex)  # 加锁
        self.data += 1
class WorkerThread(QThread):
    def __init__(self, shared_data):
        super().__init__()
        self.shared_data = shared_data
    def run(self):
        for _ in range(10):
            self.shared_data.increment()
if __name__ == "__main__":
    shared_data = SharedData()  # 创建共享数据对象
    threads = [WorkerThread(shared_data) for _ in range(4)]  # 创建多个工作线程
    for thread in threads:
        thread.start()  # 启动工作线程
    for thread in threads:
        thread.wait()  # 等待所有工作线程完成
    print("Shared data:", shared_data.data)  # 打印最终共享数据的值

import sys
from PyQt6.QtCore import QThread, QMutex, QMutexLocker
# 共享资源类，用于展示互斥锁的使用来避免死锁和线程饥饿
class SharedResource:
    def __init__(self):
        self.mutex1 = QMutex()  # 第一个互斥锁
        self.mutex2 = QMutex()  # 第二个互斥锁
    def process1(self):
        with QMutexLocker(self.mutex1):  # 获取第一个锁
            print("Process 1: Mutex 1 locked")
            QThread.msleep(100)  # 模拟处理时间
            with QMutexLocker(self.mutex2):  # 获取第二个锁
                print("Process 1: Mutex 2 locked")
    def process2(self):
        with QMutexLocker(self.mutex2):  # 获取第二个锁
            print("Process 2: Mutex 2 locked")
            QThread.msleep(100)  # 模拟处理时间
            with QMutexLocker(self.mutex1):  # 获取第一个锁
                print("Process 2: Mutex 1 locked")
class WorkerThread(QThread):
    def __init__(self, shared_resource, process_func):
        super().__init__()
        self.shared_resource = shared_resource
        self.process_func = process_func
    def run(self):
        self.process_func()
if __name__ == "__main__":
    shared_resource = SharedResource()
    thread1 = WorkerThread(shared_resource, shared_resource.process1)
    thread2 = WorkerThread(shared_resource, shared_resource.process2)
    thread1.start()  # 启动线程1
    thread2.start()  # 启动线程2
    thread1.wait()  # 等待线程1完成
    thread2.wait()  # 等待线程2完成
    print("Main thread exited")  # 主线程退出

from PyQt6.QtCore import QMutex
mutex = QMutex()
def thread_function():
    mutex.lock()
    # 访问共享资源
    mutex.unlock()
# 创建多个线程，每个线程执行 thread_function

from PyQt6.QtCore import QMutex, QMutexLocker
mutex = QMutex()
def thread_function():
    with QMutexLocker(mutex):  # 进入作用域时自动锁定，离开作用域时自动释放
        # 访问共享资源
# 创建多个线程，每个线程执行 thread_function