高并发场景下多线程共享函数执行结果的广播方案问询

需要实现一套数据广播机制：数据生成器（generator）持续生成数据，无论是否有消费者（consumer）存在；新加入的消费者从加入时刻开始接收数据。常规方案采用观察者模式，但当消费者数量极多、数据生成速率较高时，会出现数据1尚未完成向所有观察者广播，数据2就已生成的问题。

想咨询对应的解决方案，比如是否可以让多个线程共享执行同一个函数的结果——即仅运行一次函数，所有线程就能同时获取相同结果？

附参考代码：

import threading
from time import sleep

class data_provider():

    def __init__(self) -> None:
        self.cnt = 0
        self.stop = False
        self.data_received = threading.Event()

        self.data_received.clear()

        data_generator_thread = threading.Thread(target= self.data_generator)
        data_generator_thread.start()

    def data_sender(self):

        self.data_received.wait()
        
        while not self.stop:
            yield { 'Name': 'Data', 'Count': self.cnt}
            self.data_received.clear()

            self.data_received.wait()

    def data_generator(self):
        while(self.cnt < 20):
            self.cnt += 1
            self.data_received.set()
            sleep(1)

        self.stop = True
        self.data_received.set()

main.py代码：

from threading import Thread
obj = data_provider()

def printing(id):
    for data in obj.data_sender():
        print(str(id) + ' : ' + str(data))

our_threads = []

for i in range(100):
    our_threads.append(Thread(target= printing, args=(i,)))
    our_threads[i].start()

for i in range(100):
    our_threads[i].join()

print('End')

针对高并发消费者、高速数据生成的场景，核心是避免生成器向每个消费者重复推送数据，而是让所有消费者共享同一份数据快照，通过同步机制确保数据读取的一致性。以下是几种可行的实现思路：

1. 共享数据快照+事件同步 #

维护一份全局的共享数据快照，生成器仅需更新快照一次，所有消费者通过事件触发读取最新快照，无需生成器逐个推送。同时通过锁保护数据更新，避免读写冲突：

修改后的代码示例 #

import threading
from time import sleep

class data_provider():
    def __init__(self) -> None:
        self.cnt = 0
        self.stop = False
        # 通知消费者数据已更新
        self.data_ready = threading.Event()
        # 保护共享数据的读写锁
        self.data_lock = threading.Lock()
        # 维护活跃消费者数量
        self.active_consumers = 0
        self.consumer_count_lock = threading.Lock()

        data_generator_thread = threading.Thread(target=self.data_generator)
        data_generator_thread.start()

    def _add_consumer(self):
        with self.consumer_count_lock:
            self.active_consumers += 1

    def _remove_consumer(self):
        with self.consumer_count_lock:
            self.active_consumers -= 1

    def data_sender(self):
        self._add_consumer()
        try:
            while not self.stop:
                # 等待数据生成完成
                self.data_ready.wait()
                # 加锁读取当前数据快照
                with self.data_lock:
                    current_data = {'Name': 'Data', 'Count': self.cnt}
                yield current_data
                # 通知生成器当前消费者已读取完成（这里用事件清零实现批量同步）
                self.data_ready.clear()
        finally:
            self._remove_consumer()

    def data_generator(self):
        while self.cnt < 20:
            # 加锁更新共享数据
            with self.data_lock:
                self.cnt += 1
            # 通知所有消费者数据已就绪
            self.data_ready.set()
            # 等待所有活跃消费者都读取完当前数据，再生成下一条
            while self.data_ready.is_set() and self.active_consumers > 0:
                sleep(0.01)
            sleep(1)
        # 通知消费者停止
        self.stop = True
        self.data_ready.set()

2. 屏障同步（Barrier） #

使用threading.Barrier实现生成器与所有消费者的同步：生成器生成数据后，等待所有消费者都读取完当前数据，再继续生成下一条。注意需要动态维护屏障的参与者数量（消费者加入/退出时更新）：

• 消费者注册时，重新创建新的Barrier（Python原生Barrier不支持动态调整参与者数量）
• 生成器生成数据后，调用barrier.wait()等待所有消费者就绪
• 消费者读取数据后，同样调用barrier.wait()，所有线程到达屏障后，生成器继续执行

3. 共享内存原子操作 #

如果数据是简单数值类型，可以直接使用线程安全的共享内存变量（如配合锁的全局变量、multiprocessing.Value），生成器更新变量后通过事件通知消费者，消费者直接读取共享内存中的值，无需复制数据。

内容的提问来源于stack exchange，提问作者Sasan