Pyqt中的永久后台QThread向Main类发出信号

Question

我正在尝试创建一个具有 Main 类（具有永久计数方法和从 Worker 类方法接收的永久发射信号）和 Worker 类（正在计数并永久发射到 Main 类）的程序

我是编程新手所以我不知道异步或并行程序但我必须使用信号/插槽这就是我决定使用pyqt的原因

当我使用 QThread 时它阻塞了 main_cnt()

class Worker(QThread):
    tr_pass = pyqtSignal(int)

    def __init__(self):
        super().__init__()

    def run(self):
        cnt = 0
        while True:
            cnt += 1
            print("Worker cnt : ", cnt)
            self.tr_pass.emit(cnt)
            time.sleep(2)

class Main:
    def __init__(self):
        self.worker = Worker()
        self.worker.tr_pass.connect(self.worker_connect)
        self.worker.run()

        self.main_cnt()

    def main_cnt(self):
        cnt = 0
        while True:
            cnt += 1
            print("main cnt : ", cnt)
            time.sleep(3)

    def worker_connect(self, df):
        print("Connected by Worker : ", df)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    main = Main()
    app.exec_()

所以我使用了 moveToThread 异步工作但是效果不好

终端指示这样，'self.worker_.tr_pass.connect(self.worker_connect)' 不起作用

1. 主要数量：1
2. 工人数量：1
3. 工人数量：1
4. 工人数量：2
5. 主要数量：2
6. 工人数量：3
7. 工人 cnt : 4main cnt : 3
9. 工人数量：5
10. 主要数量：4
11. 主要数量：4
12. 工人数量：6

class Worker_(QObject):
    tr_pass = pyqtSignal(int)

    def __init__(self):
        super().__init__()

    def run(self):
        cnt = 0
        while True:
            cnt += 1
            print("Worker cnt : ", cnt)
            self.tr_pass.emit(cnt)
            time.sleep(2)
class Main:
    def __init__(self):
        self.worker_ = Worker_()
        self.thread = QThread()
        self.worker_.moveToThread(self.thread)
        self.thread.started.connect(self.worker_.run)
        self.worker_.tr_pass.connect(self.worker_connect)
        self.thread.start()

        self.main_cnt()

    def main_cnt(self):
        cnt = 0
        while True:
            cnt += 1
            print("main cnt : ", cnt)
            time.sleep(3)

    def worker_connect(self, df):
        print("Connected by Worker : ", df)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    main = Main()
    app.exec_()

我该如何解决这个问题 谢谢你的回答

Answer 1

如果您需要同时运行这两个功能，那么我可以使用QThreadpool 和QRunnable 来回答。

首先我们创建QThreadpool并调用线程运行函数

class Main:
def __init__(self):
    self.threadpool = QThreadPool()
    self.run_function_in_thread(self.main_cnt)

然后添加将启动线程中每个函数的函数

def run_function_in_thread(self,function):
    worker = Worker(fn=function)
    self.threadpool.start(worker)

通过创建 Worker

class Worker(QRunnable):

    def __init__(self,fn):
        super().__init__()
        self.fn = fn

    def run(self):
        self.fn()

然后我根据您的 Worker 类的逻辑添加了新功能

def another_cnter(self):
    cnt = 0
    while True:
        cnt += 1
        print("Worker cnt : ", cnt)
        self.worker_connect(cnt)
        time.sleep(2)

完整代码：

class Worker(QRunnable):

    def __init__(self,fn):
        super().__init__()
        self.fn = fn

    def run(self):
        self.fn()

class Main:
    def __init__(self):
        self.threadpool = QThreadPool()
        self.run_function_in_thread(self.another_cnter)
        time.sleep(0.5)
        self.run_function_in_thread(self.another_cnter)
        time.sleep(0.5)
        self.run_function_in_thread(self.main_cnt)

    def run_function_in_thread(self,function):
        worker = Worker(fn=function)
        self.threadpool.start(worker)

    def main_cnt(self):
        cnt = 0
        while True:
            cnt += 1
            print("main cnt : ", cnt)
            time.sleep(3)

    def another_cnter(self):
        cnt = 0
        while True:
            cnt += 1
            print("Worker cnt : ", cnt)
            self.worker_connect(cnt)
            time.sleep(2)

    def worker_connect(self,df):
        print("Connected by Worker : ", df)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    main = Main()
    app.exec_()

额外
添加信号和槽有不同的方法。一种方法是：
创建一个新类：

class WorkerSignals(QObject):
    counter = pyqtSignal(int)

初始化并传递给函数

class Worker(QRunnable):
    def __init__(self,fn):
        super().__init__()
        self.signals = WorkerSignals()
    def run(self):
        self.fn(self.signals)

连接到函数

def run_function_in_thread(self,function):
    worker = Worker(fn=function)
    worker.signals.counter.connect(self.print_out_line)
    self.threadpool.start(worker)

然后在函数中发出信号

def main_cnt(self,signals):
    cnt = 0
    signals.counter.emit(cnt)