PyQt6多线程运行无限循环数据采集仅GUI正常工作问题求助

• 实现带无限循环逻辑的类，每分钟执行一次数据采集
  • 逻辑中包含sleep阻塞调用，因此采用多线程方案实现
• 采集逻辑运行时，同步运行基于PyQt6开发的GUI界面

已编写通过无限循环实现采集逻辑的API数据采集类，初始代码如下：

class Get4hForexData:
    """
    Instantiating the different market_data objects
    """

    def __init__(self):
        # Gets list containing the markets
        self.markets = get_forex_market_to_process()  # Returns a list containing tuples
        self.list_with_market_data_objects = []

    def collect_market_data(self):

        # Loop does something once every minute in the first 3-8 seconds of each minute
        while True:
            seconds_in_current_minute = dt.datetime.now().second

            if 3 < seconds_in_current_minute < 8:
                print("now")  # Testing print statement

                # Creating all 4h timeframe marketdata objects
                for market in self.markets:
                    self.list_with_market_data_objects.append(
                        MarketData(from_currency=market[1], to_currency=market[2], timeframe="4h", outputsize=50)
                    )

                print(self.list_with_market_data_objects)  # Testing print statement
                print(self.list_with_market_data_objects[0].structure_x_coord)  # Testing print statement

            time.sleep(10)  # Stop maxing out CPU

main.py中实例化采集类，通过threading.Thread分别启动采集逻辑和GUI逻辑，初始代码如下：

def main():
    data = datacontroller.Get4hForexData()

    t1 = threading.Thread(target=data.collect_market_data)  # calling the method here
    t2 = threading.Thread(target=gui.testing)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

采集类中添加了多个print语句用于测试，GUI为带按钮的简单界面，点击按钮也会触发print输出。预期采集逻辑、GUI按钮点击操作都会向控制台打印内容，但实际运行时只有GUI正常工作，采集线程无任何输出。

• 单独运行GUI、单独运行Get4hForexData类，两者独立运行时均符合预期
• 使用简单无限循环函数替换原有采集逻辑测试多线程代码，该场景下多线程可正常运行，测试代码如下：

def loop():
    while True:

        print("Now")
        time.sleep(1)


def main():

    t1 = threading.Thread(target=loop)
    t2 = threading.Thread(target=gui.testing)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

main.py #

import gui
import threading
import datacontroller


def main():
    data = datacontroller.Get4hForexData()

    t1 = threading.Thread(target=data.collect_market_data)
    t2 = threading.Thread(target=gui.testing)

    t1.start()
    t2.start()


if __name__ == '__main__':
    main()

datacontroller.py #

import datetime as dt
from data import MarketData
import time


class Get4hForexData:
    """
    Instantiating the different market_data objects
    """
    def __init__(self):
        # Gets list containing the markets
        self.list_with_market_data_objects = []

    def collect_market_data(self):

        # Loop does something once every minute in the first 3-8 seconds of each minute
        while True:
            seconds_in_current_minute = dt.datetime.now().second

            if 3 < seconds_in_current_minute < 8:
                print("now")  # Testing print statement

                # Creating all 4h timeframe marketdata objects
                for _ in range(10):
                    self.list_with_market_data_objects.append(MarketData())

                print(self.list_with_market_data_objects)  # Testing print statement
                print(self.list_with_market_data_objects[0].name)  # Testing print statement

            time.sleep(10)  # Stop maxing out CPU

data.py #

class MarketData:

    def __init__(self):
        super().__init__()
        self.name = "test"

gui.py #

from PyQt6.QtWidgets import QApplication, QMainWindow, QPushButton


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()

        self.setWindowTitle("Market Screener")

        self.button = QPushButton("Press me.")
        self.button.clicked.connect(self.the_button_was_clicked)  # Connect button click with function

        # Set the central widget of the window.
        self.setCentralWidget(self.button)

    def the_button_was_clicked(self):
        self.button.setText("You clicked me.")
        print("Clicked")


def testing():
    app = QApplication([])

    window = MainWindow()
    window.show()

    # Start the event loop.
    app.exec()

核心根因 #

采集循环的时间判定逻辑和休眠时长不匹配，导致永远错过触发窗口：
循环每次执行完时间判断就休眠10秒，大概率直接跳过每分钟3-8秒的触发区间。举个典型运行时序：

1. 某次循环读取当前秒数为2，不满足3 < seconds < 8的条件，直接休眠10秒
2. 休眠结束后当前秒数为12，仍不满足条件，再休眠10秒
3. 后续唤醒时秒数依次为22、32、42、52、02，循环往复永远无法进入触发分支，自然不会有任何print输出。
   测试用的简单循环每次只休眠1秒，不会错过时间窗口，因此可以正常打印。

另外部分Python运行环境下，子线程的标准输出默认开启缓冲，print内容不会实时推送到控制台，会进一步掩盖逻辑问题。

修复代码 #

1. 调整循环休眠时长为1秒，CPU占用几乎可以忽略，不会错过时间窗口
2. 增加分钟标记位，避免同一分钟内重复触发采集
3. print语句增加flush=True参数，强制刷新输出缓冲
4. 采集线程设置为守护线程，关闭GUI时自动退出，避免残留后台进程

修复后的collect_market_data方法：

def collect_market_data(self):
    last_trigger_minute = -1  # 记录上次触发的分钟值，避免同一分钟重复采集
    while True:
        now = dt.datetime.now()
        current_second = now.second
        current_minute = now.minute

        # 满足时间窗口、且本分钟未触发过采集时执行逻辑
        if 3 < current_second < 8 and current_minute != last_trigger_minute:
            print("now", flush=True)
            # 如果不需要保留历史采集数据，每次采集前清空列表，避免内存无限增长
            self.list_with_market_data_objects.clear()
            for _ in range(10):
                self.list_with_market_data_objects.append(MarketData())
            print(self.list_with_market_data_objects, flush=True)
            print(self.list_with_market_data_objects[0].name, flush=True)
            last_trigger_minute = current_minute

        time.sleep(1)

调整main.py中的线程启动逻辑，设置守护线程：

def main():
    data = datacontroller.Get4hForexData()
    # 采集线程设为守护线程，GUI退出时自动终止
    t1 = threading.Thread(target=data.collect_market_data, daemon=True)
    t2 = threading.Thread(target=gui.testing)

    t1.start()
    t2.start()

额外注意事项 #

后续如果需要把采集到的数据更新到PyQt界面，禁止直接在采集线程中操作GUI控件，必须通过PyQt的信号槽机制跨线程传递数据，否则会出现界面卡死、无响应、崩溃的问题。

内容的提问来源于stack exchange，提问作者som3k