python3 queue多線程通信

89542767 發(fā)布于2022-09-18 23:56 / 480人閱讀

　　小編寫這篇文章的主要目的，主要是給大家介紹關(guān)于python3 queue多線程通信，這里面有很多的技術(shù)性的難點(diǎn)，那么，該怎么去進(jìn)行處理呢，下面小編給大家進(jìn)行詳細(xì)的解答一下。

　　queue分類

　　python3 queue分三類：

　　先進(jìn)先出隊(duì)列

　　后進(jìn)先出的棧

　　優(yōu)先級(jí)隊(duì)列

　　他們的導(dǎo)入方式分別是：

　from queue import Queue
　　from queue import LifoQueue
　　from queue import

　　具體方法見下面引用說明。

　　Queue對(duì)象已經(jīng)包含了必要的鎖，所以你可以通過它在多個(gè)線程間多安全地共享數(shù)據(jù)。當(dāng)使用隊(duì)列時(shí)，協(xié)調(diào)生產(chǎn)者和消費(fèi)者的關(guān)閉問題可能會(huì)有一些麻煩。一個(gè)通用的解決方法是在隊(duì)列中放置一個(gè)特殊的值，當(dāng)消費(fèi)者讀到這個(gè)值的時(shí)候，終止執(zhí)行。

　　例如：

　　from queue import Queue
　　from threading import Thread
　　#用來表示終止的特殊對(duì)象
　　_sentinel=object()
　　#A thread that produces data
　　def producer(out_q):
　　for i in range(10):
　　print("生產(chǎn)")
　　out_q.put(i)
　　out_q.put(_sentinel)
　　#A thread that consumes data
　　def consumer(in_q):
　　while True:
　　data=in_q.get()
　　if data is _sentinel:
　　in_q.put(_sentinel)
　　break
　　else:
　　print("消費(fèi)",data)
　　#Create the shared queue and launch both threads
　　q=Queue()
　　t1=Thread(target=consumer,args=(q,))
　　t2=Thread(target=producer,args=(q,))
　　t1.start()
　　t2.start()

　　結(jié)果：

　　本例里面有一個(gè)不尋常的位置：購買者在學(xué)到這些特殊值過后馬上又將它放返回序列中，將它傳下去。那樣，任何竊聽這一個(gè)序列的用戶進(jìn)程就能夠關(guān)閉所有了。雖然序列是一種常見的線程間通信制度，但仍然能自己根據(jù)構(gòu)建自已的程序設(shè)計(jì)并添加所需的鎖和同步機(jī)制來實(shí)現(xiàn)線程間通信。最常見的方法是使用Condition變量來包裝你的程序設(shè)計(jì)。下邊這個(gè)例子演示了如何創(chuàng)建一個(gè)線程安全的優(yōu)先級(jí)隊(duì)列。

　import heapq
　　import threading
　　class PriorityQueue:
　　def __init__(self):
　　self._queue=[]
　　self._count=0
　　self._cv=threading.Condition()
　　def put(self,item,priority):
　　with self._cv:
　　heapq.heappush(self._queue,(-priority,self._count,item))
　　self._count+=1
　　self._cv.notify()
　　def get(self):
　　with self._cv:
　　while len(self._queue)==0:
　　self._cv.wait()
　　return heapq.heappop(self._queue)[-1]

　　例子二、task_done和join

　　使用隊(duì)列來進(jìn)行線程間通信是一個(gè)單向、不確定的過程。通常情況下，你沒有辦法知道接收數(shù)據(jù)的線程是什么時(shí)候接收到的數(shù)據(jù)并開始工作的。不過隊(duì)列對(duì)象提供一些基本完成的特性，比如下邊這個(gè)例子中的task_done()和join()：

　　from queue import Queue
　　from threading import Thread
　　class Producer(Thread):
　　def __init__(self,q):
　　super().__init__()
　　self.count=5
　　self.q=q
　　def run(self):
　　while self.count&gt;0:
　　print("生產(chǎn)")
　　if self.count==1:
　　self.count-=1
　　self.q.put(2)
　　else:
　　self.count-=1
　　self.q.put(1)
　　class Consumer(Thread):
　　def __init__(self,q):
　　super().__init__()
　　self.q=q
　　def run(self):
　　while True:
　　print("消費(fèi)")
　　data=self.q.get()
　　if data==2:
　　print("stop because data=",data)
　　#任務(wù)完成，從隊(duì)列中清除一個(gè)元素
　　self.q.task_done()
　　break
　　else:
　　print("data is good,data=",data)
　　#任務(wù)完成，從隊(duì)列中清除一個(gè)元素
　　self.q.task_done()
　　def main():
　　q=Queue()
　　p=Producer(q)
　　c=Consumer(q)
　　p.setDaemon(True)
　　c.setDaemon(True)
　　p.start()
　　c.start()
　　#等待隊(duì)列清空
　　q.join()
　　print("queue is complete")
　　if __name__=='__main__':
　　main()

　　結(jié)果：

　　例子三、多線程里用queue

　　設(shè)置倆隊(duì)列，一個(gè)是要做的任務(wù)隊(duì)列todo_queue，一個(gè)是已經(jīng)完成的隊(duì)列done_queue。

　　每次執(zhí)行線程，先從todo_queue隊(duì)列里取出一個(gè)值，然后執(zhí)行完，放入done_queue隊(duì)列。

　　如果todo_queue為空，就退出。

　　import logging
　　import logging.handlers
　　import threading
　　import queue
　　log_mgr=None
　　todo_queue=queue.Queue()
　　done_queue=queue.Queue()
　　class LogMgr:
　　def __init__(self,logpath):
　　self.LOG=logging.getLogger('log')
　　loghd=logging.handlers.RotatingFileHandler(logpath,"a",0,1)
　　fmt=logging.Formatter("%(asctime)s%(threadName)-10s%(message)s","%Y-%m-%d%H:%M:%S")
　　loghd.setFormatter(fmt)
　　self.LOG.addHandler(loghd)
　　self.LOG.setLevel(logging.INFO)
　　def info(self,msg):
　　if self.LOG is not None:
　　self.LOG.info(msg)
　　class Worker(threading.Thread):
　　global log_mgr
　　def __init__(self,name):
　　threading.Thread.__init__(self)
　　self.name=name
　　def run(self):
　　while True:
　　try:
　　task=todo_queue.get(False)
　　if task:
　　log_mgr.info("HANDLE_TASK:%s"%task)
　　done_queue.put(1)
　　except queue.Empty:
　　break
　　return
　　def main():
　　global log_mgr
　　log_mgr=LogMgr("mylog")
　　for i in range(30):
　　todo_queue.put("data"+str(i))
　　workers=[]
　　for i in range(3):
　　w=Worker("worker"+str(i))
　　workers.append(w)
　　for i in range(3):
　　workers&lt;i&gt;.start()
　　for i in range(3):
　　workers&lt;i&gt;.join()
　　total_num=done_queue.qsize()
　　log_mgr.info("TOTAL_HANDLE_TASK:%d"%total_num)
　　exit(0)
　　if __name__=='__main__':
　　main()

　　輸出日志文件結(jié)果：