Python多进程异步读文件：单例类实现无重复行批量读取方案问询

原代码存在的问题 #

• 使用threading.Lock而非multiprocessing.Lock：多进程环境下，线程锁无法跨进程同步，导致锁完全失效。
• 多进程下的“单例”无效：Python多进程会复制父进程内存空间，每个进程拥有独立的类实例、文件指针和锁，无法实现跨进程的状态同步。
• 文件读取逻辑未被完整保护：for line in self.file_pointer循环未被锁覆盖，多个进程同时读取会导致文件指针位置混乱，出现重复/漏读。
• 异常处理存在错误：self.function_name未定义，应使用局部变量function_name。

优化方案与实现思路 #

针对多进程下无重复批量读文件的需求，推荐两种可靠实现方式：

方式一：共享偏移量+跨进程锁（单例类模式） #

通过跨进程共享的文件偏移量和锁，保证每个进程读取未被处理的行，同时实现单例类的跨进程唯一性。

import multiprocessing
import traceback

class ListFromTextFile:
    _instance = None
    _instance_lock = multiprocessing.Lock()  # 保证单例创建的跨进程安全

    def __new__(cls, config):
        with cls._instance_lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._init(config)
        return cls._instance

    def _init(self, config):
        self.file_path = config["file_path"]
        self.batch_size = config["batch_size"]
        # 用无符号长整型共享文件当前偏移量，所有进程可见
        self.current_offset = multiprocessing.Value('Q', 0)
        self.read_lock = multiprocessing.Lock()  # 保护文件读取和偏移量更新的跨进程锁

    def get_list_of_pids(self):
        function_name = "get_list_of_pids"
        while True:
            pids = []
            try:
                with self.read_lock:
                    # 每次读取重新打开文件，避免多进程共享文件指针的混乱
                    with open(self.file_path, "r") as f:
                        f.seek(self.current_offset.value)
                        for line in f:
                            pid = line.strip("\n")
                            pids.append(pid)
                            if len(pids) == self.batch_size:
                                # 更新偏移量为当前读取位置
                                self.current_offset.value = f.tell()
                                break
                        # 处理剩余不足批量的行
                        if len(pids) > 0 and self.current_offset.value != f.tell():
                            self.current_offset.value = f.tell()
                if not pids:
                    break  # 无更多数据，退出循环
                yield pids
            except Exception as e:
                err = str(e)
                trace_back = traceback.format_exc()
                log_exception()(function_name, err, trace_back)
                break

# 测试用的工作进程函数
def worker(config):
    reader = ListFromTextFile(config)
    for batch in reader.get_list_of_pids():
        print(f"进程 {multiprocessing.current_process().pid} 获取到批量数据: {batch}")

if __name__ == "__main__":
    config = {
        "file_path": "test.txt",
        "batch_size": 2
    }
    # 启动3个工作进程
    processes = [multiprocessing.Process(target=worker, args=(config,)) for _ in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

关键点说明

• 用multiprocessing.Lock保护单例实例的创建，确保整个进程组仅存在一个ListFromTextFile实例。
• 通过multiprocessing.Value共享文件偏移量，所有进程读取前都会定位到最新的未读取位置。
• 每次读取重新打开文件，避免多进程共享文件指针导致的状态混乱，读取过程全程由跨进程锁保护，确保无重复读取。

方式二：生产者-消费者模式（更简洁可靠） #

由一个专门的生产者进程读取文件并将批量数据放入队列，多个消费者进程从队列中取数据处理，彻底避免跨进程共享文件的问题。

import multiprocessing
import traceback

def producer(file_path, batch_size, queue):
    """生产者进程：读取文件并批量放入队列"""
    try:
        with open(file_path, "r") as f:
            batch = []
            for line in f:
                pid = line.strip("\n")
                batch.append(pid)
                if len(batch) == batch_size:
                    queue.put(batch)
                    batch = []
            # 放入剩余不足批量的数据
            if batch:
                queue.put(batch)
        # 放入结束标记，通知消费者进程退出
        queue.put(None)
    except Exception as e:
        err = str(e)
        trace_back = traceback.format_exc()
        log_exception()("producer", err, trace_back)
        queue.put(None)

def worker(queue):
    """消费者进程：从队列获取批量数据并处理"""
    while True:
        batch = queue.get()
        if batch is None:
            # 将结束标记传递给下一个进程
            queue.put(None)
            break
        print(f"进程 {multiprocessing.current_process().pid} 获取到批量数据: {batch}")

if __name__ == "__main__":
    config = {
        "file_path": "test.txt",
        "batch_size": 2
    }
    # 创建跨进程队列
    queue = multiprocessing.Queue()
    # 启动生产者进程
    producer_proc = multiprocessing.Process(target=producer, args=(config["file_path"], config["batch_size"], queue))
    producer_proc.start()
    # 启动3个消费者进程
    processes = [multiprocessing.Process(target=worker, args=(queue,)) for _ in range(3)]
    for p in processes:
        p.start()
    # 等待生产者和消费者进程结束
    producer_proc.join()
    for p in processes:
        p.join()

关键点说明

• 生产者负责所有文件读取逻辑，消费者只需专注于数据处理，职责分离更清晰。
• 利用multiprocessing.Queue实现跨进程数据传递，天然保证数据的唯一性和顺序性。
• 无需处理复杂的锁和偏移量同步，代码更简洁，可靠性更高。

总结建议 #

• 优先选择生产者-消费者模式，实现简单且不易出错。
• 若必须使用单例类读取文件，务必使用multiprocessing模块的跨进程同步原语（Lock、Value等），避免使用线程锁和进程内状态。
• 原代码的锁逻辑需重构，确保文件读取和状态更新的全程被跨进程锁保护。

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