CodePattern/13 多计算单元/数据共享/1 内存/1 队列.py

import  threading, queue
from cppy.cp_util import *
from collections import Counter

stop_words = get_stopwords()

# 待处理数据放一个队列，多个线程轮流计数，最后合并统一计数
class WordFrequencyCounter:
    def __init__(self, input_file):
        self.word_space = queue.Queue()
        self.freq_space = queue.Queue()                
        for chunk in get_chunks(input_file,3000):                        
            self.word_space.put(chunk)        

    def process_words(self):
        while not self.word_space.empty():
            try:
                chunk = self.word_space.get_nowait()  # 不使用超时，持续获取数据
            except queue.Empty:
                break  # 队列为空，退出循环
            # print(f"Worker thread ID: {threading.get_ident()}",len(chunk))  
            words = [ w for w in chunk if w not in stop_words and len(w) >= 3 ]
            word_freqs = Counter(words)
            self.freq_space.put(dict(word_freqs))  # 将Counter对象转换为字典

    def run(self):
        workers = [ threading.Thread(target=self.process_words) for _ in range(5)]
        for worker in workers:  worker.start()
        for worker in workers:  worker.join()

        word_freqs = Counter()  # 初始化一个空的Counter对象
        while not self.freq_space.empty():
            freqs = self.freq_space.get()            
            if freqs:  # 确保freqs非空
                word_freqs.update(freqs)  

        print_word_freqs ( sort_dict (word_freqs) )
        

@timing_decorator
def main(): 
    counter = WordFrequencyCounter( testfilepath )
    counter.run()

if __name__ == '__main__':
    main()

'''
在多线程之间传递数据，建议使用线程安全的队列，如queue.Queue或multiprocessing.Queue（后者也适用于多进程环境）。
这些队列提供了线程安全的数据传输机制，可以避免竞态条件和数据损坏。
全局变量不可预测

multiprocessing.Queue 利用了操作系统提供的进程间通信（IPC, Inter-Process Communication）机制，具体实现取决于不同操作系统的支持。
在Unix/Linux系统中，multiprocessing.Queue通常基于管道（pipes）、共享内存和/或消息队列等机制实现。
而在Windows系统上，可能使用命名管道（named pipes）或者内存映射文件（memory-mapped files），以及某些版本的Windows特有的进程间同步对象如Mutexes、Semaphores和事件。
'''