PyQt5中如何使用Qt5的原子整数类？

PyQt5并未绑定Qt C中的QAtomicInteger类，这类底层原子操作原语不在PyQt的对外暴露接口中——原因是Python的GIL（全局解释器锁）机制与C的无锁并发模型存在差异，Python线程安全更适合通过标准库的同步工具实现。以下是可行的替代方案：

1. 使用Python标准库的线程锁保护整数访问 #

这是最通用的线程安全整数操作方案，适用于PyQt多线程场景：

from PyQt5.QtCore import QThread
import threading

class CounterWorker(QThread):
    def __init__(self):
        super().__init__()
        self._counter = 0
        self._lock = threading.Lock()

    def run(self):
        # 模拟多线程下的计数操作
        for _ in range(10000):
            with self._lock:
                self._counter += 1
        print(f"最终计数：{self._counter}")

# 测试代码
if __name__ == "__main__":
    worker1 = CounterWorker()
    worker2 = CounterWorker()
    worker1.start()
    worker2.start()
    worker1.wait()
    worker2.wait()

2. 封装自定义原子整数类 #

如果需要模仿QAtomicInt的接口风格，可以基于线程锁封装一个自定义类：

import threading

class QAtomicInt:
    def __init__(self, initial_value=0):
        self._value = initial_value
        self._lock = threading.Lock()

    def load(self):
        """读取当前值"""
        with self._lock:
            return self._value

    def store(self, new_value):
        """设置新值"""
        with self._lock:
            self._value = new_value

    def fetch_and_add(self, delta):
        """返回旧值并增加delta，对应QAtomicInt::fetchAndAdd"""
        with self._lock:
            old_value = self._value
            self._value += delta
            return old_value

    def increment_and_get(self):
        """自增并返回新值"""
        with self._lock:
            self._value += 1
            return self._value

3. 多进程场景使用multiprocessing.Value #

如果你的并发场景是多进程而非多线程，可使用Python标准库的multiprocessing.Value实现跨进程的原子整数操作：

from PyQt5.QtCore import QProcess
from multiprocessing import Value

def worker(counter):
    for _ in range(10000):
        with counter.get_lock():
            counter.value += 1

# 测试代码
if __name__ == "__main__":
    counter = Value('i', 0)
    # 注意：QProcess调用外部函数需要将逻辑写入独立脚本，此处仅示意核心逻辑
    process1 = QProcess()
    process2 = QProcess()
    process1.start("python", ["worker_script.py"])
    process2.start("python", ["worker_script.py"])
    process1.waitForFinished()
    process2.waitForFinished()
    print(f"跨进程最终计数：{counter.value}")

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