快捷導(dǎo)航

Python實(shí)現(xiàn)線程池之線程安全隊(duì)列

更新時(shí)間：2022年05月25日 15:58:28 作者：旺旺小小超

這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)線程池之線程安全隊(duì)列，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

一、線程池組成

一個(gè)完整的線程池由下面幾部分組成，線程安全隊(duì)列、任務(wù)對(duì)象、線程處理對(duì)象、線程池對(duì)象。其中一個(gè)線程安全的隊(duì)列是實(shí)現(xiàn)線程池和任務(wù)隊(duì)列的基礎(chǔ)，本節(jié)我們通過threading包中的互斥量threading.Lock()和條件變量threading.Condition()來(lái)實(shí)現(xiàn)一個(gè)簡(jiǎn)單的、讀取安全的線程隊(duì)列。

二、線程安全隊(duì)列的實(shí)現(xiàn)

包括put、pop、get等方法，為保證線程安全，讀寫操作時(shí)要添加互斥鎖；并且pop操作可以設(shè)置等待時(shí)間以阻塞當(dāng)前獲取元素的線程，當(dāng)新元素寫入隊(duì)列時(shí)通過條件變量通知解除等待操作。

class ThreadSafeQueue(object):

? ? def __init__(self, max_size=0):
? ? ? ? self.queue = []
? ? ? ? self.max_size = max_size ?# max_size為0表示無(wú)限大
? ? ? ? self.lock = threading.Lock() ?# 互斥量
? ? ? ? self.condition = threading.Condition() ?# 條件變量

? ? def size(self):
? ? ? ? """
? ? ? ? 獲取當(dāng)前隊(duì)列的大小
? ? ? ? :return: 隊(duì)列長(zhǎng)度
? ? ? ? """
? ? ? ? # 加鎖
? ? ? ? self.lock.acquire()
? ? ? ? size = len(self.queue)
? ? ? ? self.lock.release()
? ? ? ? return size

? ? def put(self, item):
? ? ? ? """
? ? ? ? 將單個(gè)元素放入隊(duì)列
? ? ? ? :param item:
? ? ? ? :return:
? ? ? ? """
? ? ? ? # 隊(duì)列已滿 max_size為0表示無(wú)限大
? ? ? ? if self.max_size != 0 and self.size() >= self.max_size:
? ? ? ? ? ? return ThreadSafeException()

? ? ? ? # 加鎖
? ? ? ? self.lock.acquire()
? ? ? ? self.queue.append(item)
? ? ? ? self.lock.release()
? ? ? ? self.condition.acquire()
? ? ? ? # 通知等待讀取的線程
? ? ? ? self.condition.notify()
? ? ? ? self.condition.release()

? ? ? ? return item

? ? def batch_put(self, item_list):
? ? ? ? """
? ? ? ? 批量添加元素
? ? ? ? :param item_list:
? ? ? ? :return:
? ? ? ? """
? ? ? ? if not isinstance(item_list, list):
? ? ? ? ? ? item_list = list(item_list)

? ? ? ? res = [self.put(item) for item in item_list]

? ? ? ? return res

? ? def pop(self, block=False, timeout=0):
? ? ? ? """
? ? ? ? 從隊(duì)列頭部取出元素
? ? ? ? :param block: 是否阻塞線程
? ? ? ? :param timeout: 等待時(shí)間
? ? ? ? :return:
? ? ? ? """
? ? ? ? if self.size() == 0:
? ? ? ? ? ? if block:
? ? ? ? ? ? ? ? self.condition.acquire()
? ? ? ? ? ? ? ? self.condition.wait(timeout)
? ? ? ? ? ? ? ? self.condition.release()
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? return None

? ? ? ? # 加鎖
? ? ? ? self.lock.acquire()
? ? ? ? item = None
? ? ? ? if len(self.queue):
? ? ? ? ? ? item = self.queue.pop()
? ? ? ? self.lock.release()

? ? ? ? return item

? ? def get(self, index):
? ? ? ? """
? ? ? ? 獲取指定位置的元素
? ? ? ? :param index:
? ? ? ? :return:
? ? ? ? """
? ? ? ? if self.size() == 0 or index >= self.size():
? ? ? ? ? ? return None

? ? ? ? # 加鎖
? ? ? ? self.lock.acquire()
? ? ? ? item = self.queue[index]
? ? ? ? self.lock.release()

? ? ? ? return item


class ThreadSafeException(Exception):
? ? pass

三、測(cè)試邏輯

3.1、測(cè)試阻塞邏輯

def thread_queue_test_1():
? ? thread_queue = ThreadSafeQueue(10)

? ? def producer():
? ? ? ? while True:
? ? ? ? ? ? thread_queue.put(random.randint(0, 10))
? ? ? ? ? ? time.sleep(2)

? ? def consumer():
? ? ? ? while True:
? ? ? ? ? ? print('current time before pop is %d' % time.time())
? ? ? ? ? ? item = thread_queue.pop(block=True, timeout=3)
? ? ? ? ? ? # item = thread_queue.get(2)
? ? ? ? ? ? if item is not None:
? ? ? ? ? ? ? ? print('get value from queue is %s' % item)
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? print(item)
? ? ? ? ? ? print('current time after pop is %d' % time.time())

? ? t1 = threading.Thread(target=producer)
? ? t2 = threading.Thread(target=consumer)
? ? t1.start()
? ? t2.start()
? ? t1.join()
? ? t2.join()

測(cè)試結(jié)果：

我們可以看到生產(chǎn)者線程每隔2s向隊(duì)列寫入一個(gè)元素，消費(fèi)者線程當(dāng)無(wú)數(shù)據(jù)時(shí)默認(rèn)阻塞3s。通過執(zhí)行時(shí)間發(fā)現(xiàn)消費(fèi)者線程確實(shí)發(fā)生了阻塞，當(dāng)生產(chǎn)者寫入數(shù)據(jù)時(shí)結(jié)束當(dāng)前等待操作。

3.2、測(cè)試讀寫加鎖邏輯

def thread_queue_test_2():
? ? thread_queue = ThreadSafeQueue(10)

? ? def producer():
? ? ? ? while True:
? ? ? ? ? ? thread_queue.put(random.randint(0, 10))
? ? ? ? ? ? time.sleep(2)

? ? def consumer(name):
? ? ? ? while True:
? ? ? ? ? ? item = thread_queue.pop(block=True, timeout=1)
? ? ? ? ? ? # item = thread_queue.get(2)
? ? ? ? ? ? if item is not None:
? ? ? ? ? ? ? ? print('%s get value from queue is %s' % (name, item))
? ? ? ? ? ? else:
? ? ? ? ? ? ? ? print('%s get value from queue is None' % name)

? ? t1 = threading.Thread(target=producer)
? ? t2 = threading.Thread(target=consumer, args=('thread1',))
? ? t3 = threading.Thread(target=consumer, args=('thread2',))
? ? t1.start()
? ? t2.start()
? ? t3.start()
? ? t1.join()
? ? t2.join()
? ? t3.join()

測(cè)試結(jié)果：

生產(chǎn)者還是每2s生成一個(gè)元素寫入隊(duì)列，消費(fèi)者開啟兩個(gè)線程進(jìn)行消費(fèi)，默認(rèn)阻塞時(shí)間為1s，打印結(jié)果顯示通過加鎖確保每次只有一個(gè)線程能獲取數(shù)據(jù)，保證了線程讀寫的安全。