Shelve模块：使用deque.append()时如何实现文件即时增量写入

shelve的writeback=True机制会将修改的对象缓存到内存，直到关闭shelf时才全量序列化写入磁盘，这对于10GB级的大型deque来说完全不可行——不仅每次暂停时写入耗时极长，中途崩溃还可能丢失未缓存的修改。要实现每次append后即时更新、仅写入新增元素的需求，可以用以下两种方案：

方案一：手动拆分元素存储 #

直接将deque的每个元素单独存为shelf中的一个键，用计数器记录元素数量，每次新增元素时只写入对应的新键：

import shelve
from collections import deque

# 增量写入示例
with shelve.open('123') as DATA:
    # 初始化元素计数器（首次运行时创建）
    if 'TEST_counter' not in DATA:
        DATA['TEST_counter'] = 0
    
    # 新增元素1，即时写入磁盘
    DATA[f'TEST_{DATA["TEST_counter"]}'] = 1
    DATA['TEST_counter'] += 1
    
    # 新增元素2，仅写入该元素，不覆盖原有数据
    DATA[f'TEST_{DATA["TEST_counter"]}'] = 2
    DATA['TEST_counter'] += 1

# 读取时重建deque
with shelve.open('123') as DATA:
    test_deque = deque()
    for idx in range(DATA['TEST_counter']):
        test_deque.append(DATA[f'TEST_{idx}'])
    print(test_deque)  # 输出: deque([1, 2])

这个方案的核心是避免存储整个deque对象，转而拆分存储单个元素，每次append只做一次小数据写入，耗时可以忽略，且数据即时持久化，暂停程序也不会丢失进度。

方案二：封装为类，模拟原生Deque操作 #

如果希望使用起来更接近原生deque的体验，可以封装一个ShelveDeque类，内部处理增量存储的逻辑：

import shelve
from collections import deque

class ShelveDeque:
    def __init__(self, shelf_path, key_prefix='TEST'):
        self.shelf_path = shelf_path
        self.key_prefix = key_prefix
        self.counter_key = f'{key_prefix}_counter'
        
        # 初始化计数器
        with shelve.open(self.shelf_path) as db:
            if self.counter_key not in db:
                db[self.counter_key] = 0

    def append(self, item):
        """新增单个元素，即时写入磁盘"""
        with shelve.open(self.shelf_path, writeback=False) as db:
            current_idx = db[self.counter_key]
            db[f'{self.key_prefix}_{current_idx}'] = item
            db[self.counter_key] = current_idx + 1

    def extend(self, items):
        """批量新增元素"""
        with shelve.open(self.shelf_path, writeback=False) as db:
            current_idx = db[self.counter_key]
            for item in items:
                db[f'{self.key_prefix}_{current_idx}'] = item
                current_idx += 1
            db[self.counter_key] = current_idx

    def to_deque(self):
        """从磁盘读取所有元素，重建deque"""
        result = deque()
        with shelve.open(self.shelf_path) as db:
            for idx in range(db[self.counter_key]):
                result.append(db[f'{self.key_prefix}_{idx}'])
        return result

# 使用示例
sd = ShelveDeque('123')
sd.append(1)  # 即时写入
sd.append(2)  # 仅新增元素
print(sd.to_deque())  # 输出: deque([1, 2])

关键注意点 #

• 彻底放弃shelve的writeback=True：该机制是为小对象设计的，对超大型集合完全不适用，会导致内存爆炸和写入超时。
• 拆分存储的优势：每次操作只处理单个元素，写入速度极快，训练循环中每次迭代后都可以安全暂停，无需等待全量写入。

内容的提问来源于stack exchange，提问作者Nico Carrión