使用 python 在 concurrent.futures 中使用

Question

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

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

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)问题解决了。