|
一,、簡介 多個線程之間的數(shù)據(jù)是共享的，多個線程進行數(shù)據(jù)交換的時候,，不能夠保證數(shù)據(jù)的安全性和一致性,，所以當多個線程需要進行數(shù)據(jù)交換的時候，隊列就出現(xiàn)了,，隊列可以完美解決線程間的數(shù)據(jù)交換,，保證線程間數(shù)據(jù)的安全性和一致性。 二,、隊列類型 - queue.Queue(maxsize=0) 先進先出
import randomqueue_data = [1,2,3,4,5,6]random.shuffle(queue_data)print('原數(shù)據(jù)：',queue_data)queue_test = queue.Queue()def queuePut(value): queue_test.put(value)for i in queue_data: queuePut(i)for i in range(len(queue_data)): print(queue_test.get())
- queue.LifoQueue(maxsize=0) 先進后出
import queueimport randomqueue_data = [1,2,3,4,5,6]random.shuffle(queue_data)print('原數(shù)據(jù)：',queue_data)queue_test = queue.LifoQueue()def queuePut(value): queue_test.put(value)for i in queue_data: queuePut(i)for i in range(len(queue_data)): print(queue_test.get())
- queue.PriorityQueue(maxsize=0) 優(yōu)先級隊列級別越低越先出來
import queueimport randomqueue_data = [1,2,3,4,5,6]random.shuffle(queue_data)print('原數(shù)據(jù)：',queue_data)queue_test = queue.PriorityQueue()def queuePut(value): queue_test.put(value)for i in queue_data: queuePut(i)for i in range(len(queue_data)): print(queue_test.get())
- queue.SimpleQueue(maxsize=0) 簡化的隊列,，無跟蹤任務的功能，只具有empty,、get,、put3個方法
三、隊列方法 - queue.qsize() 返回隊列的大小
- queue.empty() 如果隊列為空,，返回True,反之False
- queue.full() 如果隊列滿了,，返回True,反之False
- queue.get(block,timeout) 獲取隊列，block：從隊列里面取值,，如果取不到值的話,，程序不會結束， timeout：當block的值為真的時候,，timeout是用來等待多少秒
- queue.put(value,block,timeout) 寫入隊列,，value：寫入隊列的值，,block：隊列如果滿了的話,，再往隊列里放值的話會等待,， timeout：當block的值為真的時候,，timeout是用來等待多少秒
- queue.get_nowait() 相當于queue.get(False)
- queue.put_nowait(value) 相當于queue.put(value, False)
- queue.join() 阻塞調用線程,，直到隊列中的所有任務被處理掉, 實際上意味著等到隊列為空,，再執(zhí)行別的操作
- queue.task_done() 在完成一項工作之后,，queue.task_done()函數(shù)向任務已經完成的隊列發(fā)送一個信號
- queue.full 與 maxsize 大小對應
四、隊列+線程,、隊列+線程池 案例：簡單的兩人對話 - 隊列+線程,，會導致線程頻繁的創(chuàng)建,，且主線程不關閉,，其他線程就會一直開啟
import queueimport threadingimport timeclass ChatSend(threading.Thread): def __init__(self, name, message_queue): super().__init__() self.name = name self.message_queue = message_queue def run(self): '''使用隊列中的任務跟蹤''' message = input(f'{self.name}：') self.message_queue.put({ 'name': self.name, 'message': message, 'time': time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()) }) self.message_queue.join()class ChatReceive(threading.Thread): def __init__(self, name, message_queue): super().__init__() self.name = name self.message_queue = message_queue def run(self): message_info = self.message_queue.get() print(f'\n{message_info['time']}') print(f'{self.name}接收到{message_info['name']}消息：{message_info['message']}\n') self.message_queue.task_done()def chat(chat_person1, chat_person2): message_queue = queue.Queue() print(f'----------{chat_person1}和{chat_person2}正在對話----------') while True: send1 = ChatSend(chat_person1, message_queue) receive1 = ChatReceive(chat_person2, message_queue) send1.start() receive1.start() receive1.join() send2 = ChatSend(chat_person2,message_queue) receive2 = ChatReceive(chat_person1,message_queue) send2.start() receive2.start() receive2.join()chat('張三', '李四')
- 隊列+線程+鎖
import queueimport threadingimport timeclass ChatSend(threading.Thread): def __init__(self, name, message_queue, lock=None): super().__init__() self.name = name self.message_queue = message_queue self.lock = lock def run(self): self.lock.acquire() message = input(f'{self.name}：') self.message_queue.put({ 'name': self.name, 'message': message, 'time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) }) self.lock.notify() self.lock.release() class ChatReceive(threading.Thread): def __init__(self, name, message_queue, lock=None): super().__init__() self.name = name self.message_queue = message_queue self.lock = lock def run(self): self.lock.acquire() self.lock.wait() message_info = self.message_queue.get() print(f'\n{message_info['time']}') print(f'{self.name}接收到{message_info['name']}消息：{message_info['message']}\n') self.lock.release()def chat(chat_person1, chat_person2): message_queue = queue.Queue() lock = threading.Condition() print(f'----------{chat_person1}和{chat_person2}正在對話----------') while True: receive1 = ChatReceive(chat_person2, message_queue, lock) send1 = ChatSend(chat_person1, message_queue, lock) receive1.start() send1.start() receive1.join() receive2 = ChatReceive(chat_person1, message_queue, lock) send2 = ChatSend(chat_person2,message_queue, lock) receive2.start() send2.start() receive2.join()chat('張三', '李四')
- 隊列+池,，不會頻繁創(chuàng)建線程
import queueimport threadingimport timefrom concurrent.futures import ThreadPoolExecutor,waitlock = threading.Condition()class ChatSend: def __init__(self, name, message_queue): self.send_person = name self.message_queue = message_queue def run(self): message = input(f'{self.send_person}：') self.message_queue.put({ 'name': self.send_person, 'message': message, 'time': time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()) }) self.message_queue.join()class ChatReceive: def __init__(self, name, message_queue): self.receive_person = name self.message_queue = message_queue def run(self): message_info = self.message_queue.get() print(f'\n{message_info['time']}') print(f'{message_info['name']}對{self.receive_person}說：{message_info['message']}\n') self.message_queue.task_done()class Chat(ThreadPoolExecutor): def __init__(self,person1, person2): ThreadPoolExecutor.__init__(self,max_workers=2) message_queue = queue.Queue() print(f'----------{person1}和{person2}正在對話----------') self.send1 = ChatSend(person1, message_queue) self.receive1 = ChatReceive(person2, message_queue) self.send2 = ChatSend(person2, message_queue) self.receive2 = ChatReceive(person1, message_queue) def start(self): while True: self.submit(self.send1.run) self.submit(self.receive1.run) self.submit(self.send2.run) self.submit(self.receive2.run)chat = Chat('張三', '李四')chat.start()
|
