Python多线程任务中如何实时更新共享状态

兄弟，完全懂你看一堆资料越看越头大的感受！先给你捋清楚几个核心问题，再给你落地的解决方案：

首先得纠正一个小误区：你代码里用的multiprocessing.Pool是多进程池，不是多线程！进程之间的内存是完全隔离的，所以你想让子进程直接修改主进程的self.state根本行不通——每个子进程里的self.state都是独立的副本，改了也不会同步到主进程。而爬虫这类IO密集型任务，其实用多线程更合适，资源开销小，还能直接共享内存（注意线程安全就行）。

解决方案一：用多线程实时更新状态（推荐，适合爬虫场景） #

我们用concurrent.futures.ThreadPoolExecutor来实现，它能让我们在任务完成的第一时间就把结果加到state里，不用等所有任务跑完。而且代码比你原来的更简洁：

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

def some_function(job_dict: dict) -> dict:
    # 你的爬虫逻辑，比如返回结果或者抛出异常
    # 示例：
    # import requests
    # try:
    #     resp = requests.get(job_dict['url'])
    #     return {'url': job_dict['url'], 'status': resp.status_code}
    # except Exception as e:
    #     print(f"爬取{job_dict['url']}出错: {e}")
    #     return None

class RunJobs:
    def __init__(self, jobs: list[dict], threads=1):
        self.jobs = jobs
        self.threads = threads
        self.state = []
        # 加个锁保证线程安全，避免极端情况下的列表竞争问题
        self.state_lock = threading.Lock()

    def _run_sequentially(self):
        for job in self.jobs:
            self.state.append(some_function(job))

    def _run_multithreaded(self):
        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            # 把所有任务提交到线程池，映射任务和future对象
            future_map = {executor.submit(some_function, job): job for job in self.jobs}
            
            # 遍历所有完成的任务，实时更新state
            for future in as_completed(future_map):
                job = future_map[future]
                try:
                    result = future.result()
                    # 加锁修改共享状态
                    with self.state_lock:
                        self.state.append(result)
                    print(f"任务 {job} 完成，已更新状态")
                except Exception as exc:
                    print(f"任务 {job} 执行失败: {exc}")
                    # 出错时也可以添加一个标记到state里
                    with self.state_lock:
                        self.state.append(None)

    def run(self):
        try:
            if self.threads == 1:
                self._run_sequentially()
            else:
                self._run_multithreaded()
        except KeyInterrupt:
            print("用户中断，当前状态:", self.state)

为什么这个方案能实时更新？ #

as_completed会在每个任务完成时立刻返回对应的future对象，我们调用future.result()拿到结果后，马上就把它加到self.state里——完全不用等所有任务都结束，真正做到“边跑边更”。

如果坚持要用多进程怎么办？ #

要是你因为某些原因必须用多进程，那得用进程间通信的方式（比如队列）来传递结果，主进程专门开个线程监听队列，实时把结果加到state里：

from multiprocessing import Pool, Queue
import threading

def some_function(job_dict: dict) -> dict:
    # 你的爬虫逻辑，同上

# 多进程的worker函数，把结果放到队列里
def process_worker(job, queue):
    try:
        result = some_function(job)
        queue.put(result)
    except Exception as exc:
        print(f"任务出错: {exc}")
        queue.put(None)

class RunJobs:
    def __init__(self, jobs: list[dict], threads=1):
        self.jobs = jobs
        self.threads = threads
        self.state = []
        self.queue = Queue()

    # 监听队列的线程函数，实时更新state
    def _queue_listener(self):
        while True:
            item = self.queue.get()
            if item is None:  # 用None作为任务结束的信号
                break
            self.state.append(item)
            print(f"收到结果，状态已更新")

    def _run_multiprocessed(self):
        # 启动监听队列的线程
        listener_thread = threading.Thread(target=self._queue_listener)
        listener_thread.start()

        with Pool(self.threads) as pool:
            for job in self.jobs:
                # 把队列传给每个worker进程
                pool.apply_async(process_worker, args=(job, self.queue))
            pool.close()
            pool.join()
        
        # 发送结束信号，让监听线程退出
        self.queue.put(None)
        listener_thread.join()

    def _run_sequentially(self):
        for job in self.jobs:
            self.state.append(some_function(job))

    def run(self):
        try:
            if self.threads == 1:
                self._run_sequentially()
            else:
                self._run_multiprocessed()
        except KeyInterrupt:
            print("用户中断，当前状态:", self.state)

最后补个你原来代码里的小bug #

你写的pool.apply_async(some_function, args=(job))是错的！args需要传元组，正确写法是args=(job,)——少个逗号的话，Python会把job这个字典拆成一堆参数传给some_function，肯定会报错。

备注：内容来源于stack exchange，提问作者probablyjg