python - 使用 python 在并发.futures 中 as_completed 出现意外行为。

Question

我正在努力在Python中使用concurrent.futures。我正在尝试迭代大量 S3 对象。由于帐户、存储桶和对象的数量较多，这可能需要很长时间。比我的 STS 凭证有效的时间长，并且足够长，我不相信脚本不会被中断。

我希望以下代码能够工作，虽然它确实产生了我在使用少量存储桶进行测试时正在寻找的输出，但它仅在完全处理每个存储桶之后而不是在每个 future 之后写入已完成和输出文件返回。如果它被中断，则不会写入已完成的文件和输出文件。即使许多桶已经被成功处理。

if __name__ == '__main__':
    args_results = parser.parse_args()

    completed = open(args_results.completed, 'a+')
    out = open(args_results.out, 'a+')  

    done = getCompleted(args_results.completed) 
    todo = getBuckets(args_results.todo)

    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = []
        for item in todo: 
            if item not in done:
                account, bucket = item.split('|')
                futures.append(executor.submit(getBucketInfo, account, bucket))

        for x in as_completed(futures):
            result = x.result()
            out.write(result + '\n')
            completed.write(result['Account'] + '|' + result['Bucket'] + '\n')

我是否误解了 as_completed() 功能的工作原理？

Answer 1

我需要在打开文件时添加行缓冲，以便它们在写入的每一行时都会刷新到磁盘。

completed = open(args_results.completed, 'a+', buffering=1)
out = open(args_results.out, 'a+', buffering=1)

问题已解决。