Python迭代器的一次性消耗与多进程starmap的陷阱

x = (0, 1, 2)
y = "ABC"
zipper = zip(x, y)

# 第一次消耗：将迭代器转换为列表
print("转换为列表后：", list(zipper))

# 尝试第二次遍历：迭代器已耗尽，不会打印任何内容
print("尝试第二次遍历：")
for n, s in zipper:
    print(n, s)

from itertools import repeat
import multiprocessing

def starmap_with_kwargs(pool, fn, args_iter, kwargs_iter):
    # args_for_starmap 是一个zip迭代器
    args_for_starmap = zip(repeat(fn), args_iter, kwargs_iter)
    print(args_iter) # 打印的是zip对象本身，不是其内容
    return pool.starmap(apply_args_and_kwargs, args_for_starmap)

def apply_args_and_kwargs(fn, args, kwargs):
    print('test') # 只有当任务被实际执行时才会打印
    return fn(*args, **kwargs)

def func(path, dictArg, **kwargs):
    # 原始代码中这里存在一个TypeError的潜在问题
    # dictArg 是 {'a: 2'} 这样的字符串，而不是字典
    for i in dictArg: # 如果dictArg是字符串，这里会遍历字符串的字符
        print(i['a']) # 尝试对字符进行字典索引，导致TypeError
        print(kwargs['yes'])

def funcWrapper(path, dictList, **kwargs):
    args_iter = zip(repeat(path), dictList) # 第一个zip迭代器
    kwargs_iter = repeat(kwargs)

    # 关键行：list(args_iter)
    # 如果这行被执行，args_iter会被耗尽

    pool = multiprocessing.Pool()
    starmap_with_kwargs(pool, func, args_iter, kwargs_iter)
    pool.close() # 最佳实践：关闭进程池
    pool.join()  # 最佳实践：等待所有任务完成

# 原始数据中dictList的结构需要修正，否则func会报错
# 修正后的dictList应为字典列表，而不是字符串列表
dictList = [{'a': 2}, {'a': 65}, {'a': 213}, {'a': 3218}]
path = 'some/path/to/something'

funcWrapper(path, dictList, yes=1)

dictList = [{'a': 2}, {'a': 65}, {'a': 213}, {'a': 3218}]

def func(path, dictArg, **kwargs):
    # 假设dictArg是 {'a': 2} 这样的字典
    print(dictArg['a']) # 直接访问字典的'a'键
    print(kwargs['yes'])

from itertools import repeat
import multiprocessing

def starmap_with_kwargs(pool, fn, args_iter, kwargs_iter):
    # 这里我们确保args_iter和kwargs_iter在创建args_for_starmap时都是可用的
    # 如果args_iter或kwargs_iter是单次消耗的迭代器，且之前已被使用，
    # 那么此处需要重新生成或转换为列表

    # 假设args_iter和kwargs_iter在这里是新生成的迭代器，或者已经是列表
    args_for_starmap = zip(repeat(fn), args_iter, kwargs_iter)
    return pool.starmap(apply_args_and_kwargs, args_for_starmap)

def apply_args_and_kwargs(fn, args, kwargs):
    print(f'Executing task for args={args}, kwargs={kwargs}')
    return fn(*args, **kwargs)

def func(path, dictArg, **kwargs):
    # 修正后的func，假设dictArg是字典
    print(f"Path: {path}, Dict Arg 'a': {dictArg['a']}, Kwarg 'yes': {kwargs['yes']}")
    return f"Processed {path} with {dictArg['a']}"

def funcWrapper(path, dictList, **kwargs):
    # 每次需要时重新生成迭代器，或者将原始数据转换为列表
    # 这里为了演示，我们假设每次调用funcWrapper时，args_iter和kwargs_iter都是新创建的

    # 如果dictList是一个可迭代对象且需要多次使用，应先转换为列表
    # processed_dict_list = list(dictList) 

    args_iter = zip(repeat(path), dictList) # 创建新的zip迭代器
    kwargs_iter = repeat(kwargs)

    pool = multiprocessing.Pool()
    results = starmap_with_kwargs(pool, func, args_iter, kwargs_iter)

    pool.close()
    pool.join()

    print("\nAll tasks completed. Results:")
    for res in results:
        print(res)

# 修正后的dictList，确保每个元素都是一个字典
dictList = [{'a': 2}, {'a': 65}, {'a': 213}, {'a': 3218}]
path = 'some/path/to/something'

if __name__ == '__main__':
    funcWrapper(path, dictList, yes=1)