使用全局时，Python运行threadpoolexecutor crosswise

Question

在我的Python应用程序中，我使用芹菜处理长期运行的任务（其中许多）。在我的一项任务中，我有以下代码将段上传到S3存储，请一般可以看一下，

    if not debug_level['DONT_SEND']:
        count_lock = threading.Lock()
        obj_count = 0

        def __upload(object_path_pair):
            def percentage(part, whole):
                percentage = 100 * int(part) / int(whole)
                return str(percentage) + "%"
            global obj_count
            if obj_count > 0:
                print(f'Uploading segment #{obj_count} to CDN.')
            libera_resource.upload_file(*object_path_pair)
            sleep(random.uniform(1, 5))
            with count_lock:
                if obj_count > 0:
                    print(f' Segment {obj_count} of {len(segment_upload_list)} ({percentage(obj_count, whole=len(segment_upload_list))}) uploaded successfully.')
                obj_count += 1

        def upload_segments(segment_upload_list):
            global obj_count
            obj_count = 0
            with ThreadPoolExecutor(max_workers=50) as executor:
                executor.map(__upload, segment_upload_list)
            print('\n!!! All Segments Uploaded !!!')

        upload_segments(segment_upload_list)

但是一旦任务运行两次，我会从我的打印语句中收到奇怪的消息

“ print（f'segment {obj_count} （{百分比（obj_count，thoter = len（segment_upload_list））}）上传 成功。'）。

成功的 我会回来（如果它运行的时间更多，则同时运行） “ 100（140％）的第233段成功上传。”

如您所见，打印说明没有多大意义，如果任务在后台运行两次，为什么要像这样打印它？ 这是由于我设定的全球人数吗？如果是这样，这里可能有什么解决方法？ 我目前唯一可以想象的是，任务调用1和在运行时呼叫2使用了全局计数，然后导致异常计数，这再次导致了这种丑陋的输出。

提前致谢

Answer 1

如果您正在运行多个上传线程，则应该为每个线程提供一个计数器。

一种可能的解决方案是使用 thread noreferrer“> thread-local data 对于计数。

编辑：
线程本地数据的一个重要限制是，只能通过其自己的线程才能访问。

因此，如果您需要来自多个线程的数据， 另一个线程可访问，则必须创建一个线程特定数据的全局字典。
有关此词典的一些观点：

• 使用线程的esident作为字典的键。或者，只要您将其设置为描述性，name。
• 每个线程都应 更新其自己的数据！
• 每个线程都可以读取来自其他线程的数据。

这应该主要是 阻止字典被拧紧。
但是，在更新字典时，线程会被中断有很小的机会。
那将使数据处于不一致的状态。

因此，您可能应该通过lock序列化对字典的访问。

edit2：

线程的小示例，该线程在锁定锁定时会更新全局数据结构。

def target(args):
    '''Function to run in a thread.'''
    tid = threading.get_ident()
    running = True
    while running:
        # do stuff...
        # Update the global data structure.
        data_modify_lock.acquire()
        tread_data[tid]['obj_count'] = ...
        # et cetera
        data_modify_lock.release()


data_modify_lock = threading.Lock()
thread_data = {
   2: {'obj_count': 6, 'segment_list': [...]},
   3: {'obj_count': 2, 'segment_list': [...]}
}