1.今天来利用queue来实现一个线程安全队列。

2.场景：有时我们需要将一个大任务划分成很多小任务，每个小任务执行完得到结果需要放入一个安全队列里面，一个收集结果的线程就实时从队列中取结果将子任务结果聚合，形成大任务计算的结果。

3.代码实现

import queueimport threadingimport time## 封装的安全队列类（十分通用）class SafeQueue(threading.Thread):    # 退出队列的信号    SIG_QUIT = 'sig_quit'    def __init__(self,recv_calback):        threading.Thread.__init__(self)        ## 构造线程安全队列        self.Q = queue.Queue()        self.recv_calback = recv_calback        self.start()    #放入队列    def put(self,datas):        threadName = threading.currentThread().name        self.Q.put(datas)    #关闭队列    def close(self):        self.put(SafeQueue.SIG_QUIT)    ##主循环,处理队列接收    def run(self):        while True:           try:               datas = self.Q.get()               if datas == SafeQueue.SIG_QUIT: #收到退出队列信号                   break                #回调客户端               self.recv_calback(datas)           except: # 抛出打断异常               break##队列回调函数def queue_callback(datas):    print("接收到数据:",datas)    ## 将子任务结果加入 全局集合    try:        array_mutex.acquire()#锁定        datas_array.append(datas)        if len(datas_array) == 4:            safeQueue.close()            print("=======大任务计算结束===========  result:", datas_array)    finally:        array_mutex.release()#释放## 子任务计算函数def calclulate():    threadName = threading.currentThread().name    print(threadName , ' 正在计算')    time.sleep(2)    print(threadName, ' 计算完成,加入队列')    #将结果放入队列    safeQueue.put(threadName+"' result")####  ----------  main start   ----------#创建锁array_mutex = threading.Lock()## 存储 子任务计算结果的 集合datas_array = []##构造安全队列safeQueue = SafeQueue(queue_callback)##开启4 个子任务,开始计算for i in range(1,5):    threading.Thread(target=calclulate).start()

这个SafeQueue模型在项目开发中非常常见，这里有什么封装的不足的谢谢指出。今天队列就到这里，谢谢大家。