Python中如何使用threading

Python中如何使用threading，針對這個問題，這篇文章詳細(xì)介紹了相對應(yīng)的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

　　Python 版本是3.7.4

　　前面的文章記錄了網(wǎng)絡(luò)請求(urllib,requests)、數(shù)據(jù)提取(beautiful,xpath,正則)、數(shù)據(jù)存儲(json,csv)的學(xué)習(xí)，下面進(jìn)行一個多線程的學(xué)習(xí)。

　　多線程爬蟲

　　有些時候，比如下載圖片，因為下載圖片是一個耗時的操作，如果采用之前那種同步的方式下載，那效率會特別慢。這時候我們就可以考慮使用多線程的方式來下載圖片。

　　多線程介紹

　　多線程是為了同步完成多項任務(wù)，通過提高資源使用來提高系統(tǒng)的效率，線程是在同一時間需要完成多項任務(wù)的時候是西納的，最簡單的比喻多線程就像火車的每一節(jié)車廂，二進(jìn)程就是火車。車廂離開火車是無法跑動的，同理火車可以有多節(jié)車廂，多線程的出現(xiàn)是為了提高效率，同時他的出現(xiàn)也帶來一些問題。

　　簡單來講，多線程就相當(dāng)于你原來開了一個窗口爬取，限制開了十個窗口來爬取。

　　threading模塊介紹

　　threading模塊是Python中專門提供用來做多線程的模塊。threading模塊中最常用的類是Thread。下面一個簡單的多線程程序：

　　# 引入所需庫

　　import threading

　　import time

　　def coding():

　　"""

　　coding函數(shù)

　　:return:

　　"""

　　for x in range(5):

　　print('%s 號程序員正則寫代碼...' % x)

　　time.sleep(1)

　　def drawing():

　　"""

　　drawing函數(shù)

　　:return:

　　"""

　　for x in range(5):

　　print('%s 號設(shè)計師正在設(shè)計圖片...' % x)

　　time.sleep(1)

　　def single_thread():

　　"""

　　單線程執(zhí)行

　　:return:

　　"""

　　coding()

　　drawing()

　　def multi_thread():

　　"""

　　多線程執(zhí)行

　　:return:

　　"""

　　# 創(chuàng)建線程

　　# 注意:target參數(shù)是函數(shù)名,不能帶括號

　　t1 = threading.Thread(target=coding, name='coding')

　　t2 = threading.Thread(target=drawing, name='drawing')

　　# 啟動線程

　　t1.start()

　　t2.start()

　　if __name__ == '__main__':

　　# single_thread()

　　multi_thread()

　　查看線程數(shù)：

　　num = threading.enumerate()

　　print(num)

　　查看當(dāng)前進(jìn)程名字：

　　threading.current_thread()

　　Thread類的使用

　　為了讓線程代碼更好的封裝，可以使用threading模塊下的Thread類，繼承自這個類，然后實現(xiàn)run()方法，線程就會自動運行run()方法中的代碼，示例代碼如下：

　　# 引入所需庫

　　import threading

　　import time

　　class CodingThread(threading.Thread):

　　"""

　　寫程序進(jìn)程類

　　"""

　　def run(self):

　　for x in range(5):

　　print('%s 號程序員正則寫代碼...' % threading.current_thread())

　　time.sleep(1)

　　class DrawingThread(threading.Thread):

　　"""

　　設(shè)計進(jìn)程類

　　"""

　　def run(self):

　　for x in range(5):

　　print('%s 號設(shè)計師正在設(shè)計圖片...' % threading.current_thread())

　　time.sleep(1)

　　def multi_thread():

　　t1 = CodingThread()

　　t2 = DrawingThread()

　　t1.start()

　　t2.start()

　　if __name__ == '__main__':

　　multi_thread()

　　多線程共享全局變量問題

　　多線程都是在同一個進(jìn)程中運行的，因此在進(jìn)程中的全局變量所有的線程都是可以共享的。這就造就了一個問題，因為線程執(zhí)行的順序是無序的，有可能會造成數(shù)據(jù)錯誤。例如如下代碼：

　　# 引入threading庫

　　import threading

　　# 定義全局變量

　　VALUE = 0

　　def add_value():

　　"""

　　增加數(shù)值

　　:return:

　　"""

　　global VALUE

　　for x in range(1000000):

　　VALUE += 1

　　print(VALUE)

　　def main():

　　for x in range(2):

　　t = threading.Thread(target=add_value)

　　t.start()

　　if __name__ == '__main__':

　　main()

　　以上的代碼結(jié)果正常來講應(yīng)該是：

　　1000000

　　2000000

　　但是由于多線程運行的不確定性，因此結(jié)果可能是隨機(jī)的。

　　鎖機(jī)制

　　為了解決上述問題由于多線程運行的不確定性，threading庫增加了Lock類鎖機(jī)制進(jìn)行處理，當(dāng)某個線程對全局變量進(jìn)行修改時則將此變量加鎖不允許其他線程進(jìn)行修改，知道當(dāng)前線程修改完這個變量之后再進(jìn)行解鎖釋放，之后其他線程才可進(jìn)行修改，這就保證了數(shù)據(jù)的安全性。修改上述代碼如下：

　　# 引入threading庫

　　import threading

　　# 定義全局變量

　　VALUE = 0

　　# 創(chuàng)建鎖

　　gLock = threading.Lock()

　　def add_value():

　　"""

　　增加數(shù)值

　　:return:

　　"""

　　global VALUE

　　# 加鎖

　　gLock.acquire()

　　for x in range(1000000):

　　VALUE += 1

　　# 解鎖

　　gLock.release()

　　print(VALUE)

　　def main():

　　for x in range(2):

　　t = threading.Thread(target=add_value)

　　t.start()

　　if __name__ == '__main__':

　　main()

　　Lock版生產(chǎn)者和消費者模式

　　生產(chǎn)者和消費者模式時多線程開發(fā)中的經(jīng)常見到的一種模式。生產(chǎn)者的線程專門用來生產(chǎn)一些數(shù)據(jù)，然后存放到一個中間的變量中。消費者再從這個中間的變量中取出數(shù)據(jù)進(jìn)行消費，但是因為要使用中間變量，中間變量經(jīng)常是一些全局變量，因此需要使用鎖來保證數(shù)據(jù)的完整性。以下是使用threading.Lock()鎖實現(xiàn)“生產(chǎn)者與消費者模式”的一個例子：

　　# 引入所需庫

　　import random

　　import threading

　　import time

　　gMoney = 1000

　　gTimes = 0

　　# 定義鎖

　　gLock = threading.Lock()

　　class Producer(threading.Thread):

　　"""

　　生產(chǎn)者

　　"""

　　def run(self):

　　global gMoney

　　global gTimes

　　while True:

　　money = random.randint(100, 1000)

　　gLock.acquire()

　　# 僅允許生產(chǎn)10次

　　if gTimes >= 10:

　　gLock.release()

　　break

　　gMoney += money

　　print('%s生產(chǎn)了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))

　　gTimes += 1

　　gLock.release()

　　time.sleep(0.5)

　　class Consumer(threading.Thread):

　　"""

　　消費者

　　"""

　　def run(self):

　　global gMoney

　　while True:

　　money = random.randint(100, 1000)

　　gLock.acquire()

　　if gMoney >= money:

　　gMoney -= money

　　print('%s消費了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))

　　else:

　　if gTimes >= 10:

　　gLock.release()

　　break

　　print("%s消費者消費錢不夠,不消費" % threading.current_thread())

　　gLock.release()

　　time.sleep(0.5)

　　def main():

　　# 定義三個消費者

　　for x in range(3):

　　t = Consumer(name='消費者線程%d' % x)

　　t.start()

　　# 定義五個生產(chǎn)者

　　for x in range(5):

　　t = Producer(name='生產(chǎn)者線程%d' % x)

　　t.start()

　　if __name__ == '__main__':

　　main()

　　Condition版生產(chǎn)者與消費者模式

　　Lock()版的生產(chǎn)者與消費者模式可以正常的運行，但是存在一個不足，在消費者中，總是通過while True死循環(huán)并且上鎖的方法去判斷錢夠不夠，上鎖是一個很好CPU資源的行為。因此這種方式不是最好的，還有一種更好的方式便是使用threading.Condition來實現(xiàn)。

　　threading.Condition可以在沒有數(shù)據(jù)的時候處于阻塞等等狀態(tài)。一旦有合適的數(shù)據(jù)了，還可以使用notify相關(guān)的函數(shù)來通知其他處于等待狀態(tài)的線程，這樣就可以不用做一些無用的上鎖和解鎖的操作，可以提高程序的性能。

　　首先對threading.Condition相關(guān)的函數(shù)做個介紹，threading.Condition類似threading.Lock，可以在修改全部數(shù)據(jù)的時候進(jìn)行上鎖，也可以在修改完畢后進(jìn)行解鎖。以下將一些常用的函數(shù)做個簡單的介紹：

　　acquire : 上鎖

　　release : 解鎖

　　wait : 將當(dāng)前線程處于等待狀態(tài)，并且會釋放鎖?？梢员黄渌€程使用notify和notify_all函數(shù)喚醒，被喚醒后會繼續(xù)等待上鎖，上鎖后繼續(xù)執(zhí)行后續(xù)的代碼

　　notify : 通知某個正在等待的線程，默認(rèn)是第1個等待的線程

　　notify_all : 通知所有正在等待的線程。notify和notify_all不會釋放鎖，并且需要在release之前掉用

　　Condition版生產(chǎn)者與消費者模式示例代碼如下：

　　# 引入所需庫

　　import random

　　import threading

　　import time

　　gMoney = 1000

　　gTimes = 0

　　# 定義Condition

　　gCondition = threading.Condition()

　　class Producer(threading.Thread):

　　"""

　　生產(chǎn)者

　　"""

　　def run(self):

　　global gMoney

　　global gTimes

　　while True:

　　money = random.randint(100, 1000)

　　gCondition.acquire()

　　# 僅允許生產(chǎn)10次

　　if gTimes >= 10:

　　gCondition.release()

　　break

　　gMoney += money

　　print('%s生產(chǎn)了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))

　　gTimes += 1

　　gCondition.notify_all()

　　gCondition.release()

　　time.sleep(0.5)

　　class Consumer(threading.Thread):

　　"""

　　消費者

　　"""

　　def run(self):

　　global gMoney

　　while True:

　　money = random.randint(100, 1000)

　　gCondition.acquire()

　　while gMoney < money:

　　if gTimes >= 10:

　　gCondition.release()

　　return

　　print("%s消費者消費錢不夠,不消費" % threading.current_thread())

　　gCondition.wait()

　　gMoney -= money

　　print('%s消費了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))

　　gCondition.release()

　　time.sleep(0.5)

　　def main():

　　# 定義三個消費者

　　for x in range(3):

　　t = Consumer(name='消費者線程%d' % x)

　　t.start()

　　# 定義五個生產(chǎn)者

　　for x in range(5):

　　t = Producer(name='生產(chǎn)者線程%d' % x)

　　t.start()

　　if __name__ == '__main__':

　　main()

　　Queue線程安全隊列

　　在線程中，訪問一些全局變量，加鎖是一個經(jīng)常的過程，如果你想把一些數(shù)據(jù)存儲到莫格隊列中，那么Python內(nèi)置了一個線程安全的模塊叫queue模塊。Python中的queue模塊中提供了同步的、線程安全的對咧咧，包括FIFO(先進(jìn)先出)隊列Queue，LIFO(后入先出)隊列LifoQueue。這些隊列都實現(xiàn)了所原理(可以理解為原子操作，即要么不做，要么都做完)，能夠在多線程中直接使用。可以使用隊列來實現(xiàn)線程間的同步，相關(guān)函數(shù)如下：

　　初始化Queue(maxsize) : 創(chuàng)建一個先進(jìn)先出的隊列

　　qsize() : 返回隊列的大小

　　empty() : 判斷隊列是否為空

　　full() : 判斷隊列是否已滿

　　get() : 從隊列中獲取最后一個數(shù)據(jù)

　　put() : 將一個數(shù)據(jù)放到隊列中

　　使用代碼示例：

　　# 引入所需庫

　　import threading

　　import time

　　from queue import Queue

　　def set_value(q):

　　"""

　　寫入隊列

　　:param q:

　　:return:

　　"""

　　index = 1

　　while True:

　　q.put(index)

　　index += 1

　　time.sleep(3)

　　def get_value(q):

　　"""

　　從隊列取值

　　:param q:

　　:return:

　　"""

　　while True:

　　print(q.get())

　　# time.sleep(4)

　　# print("qsize:", q.qsize())

　　def main():

　　"""

　　主函數(shù)

　　:return:

　　"""

　　q = Queue(5)

　　t1 = threading.Thread(target=set_value, args=[q])

　　t2 = threading.Thread(target=get_value, args=[q])

　　t1.start()

　　t2.start()

　　if __name__ == '__main__':

　　main()

　　使用實例

　　單線程爬取表情包，實例代碼如下：

　　# 引入所需庫

　　import os

　　import re

　　import requests

　　from lxml import etree

　　def parse_page(url):

　　"""

　　請求 解析 下載

　　:param url:

　　:return:

　　"""

　　# 聲明定義請求頭

　　headers = {

　　'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36',

　　}無錫人流醫(yī)院哪家好 http://www.bhnfk.com

　　req = requests.get(url=url, headers=headers)

　　html = req.text

　　tree = etree.HTML(html)

　　imgs = tree.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]')

　　for img in imgs:

　　img_url = img.get('data-original')

　　alt = img.get('alt')

　　alt = re.sub(r'[\??\..,!!]]', '', alt)

　　suffix = os.path.splitext(img_url)[1]

　　file_name = alt + suffix

　　req_img = requests.get(url=img_url, headers=headers)

　　with open('images/' + file_name, 'wb') as fp:

　　fp.write(req_img.content)

　　print(file_name)

　　def main():

　　"""

　　主函數(shù)

　　:return:

　　"""

　　for x in range(1, 101):

　　print("第%d頁開始下載..." % x)

　　url = 'http://www.doutula.com/photo/list/?page=%d' % x

　　parse_page(url)

　　print("第%d頁結(jié)束下載..." % x)

　　if __name__ == '__main__':

　　main()

　　多線程爬取表情包，實例代碼如下：

　　# 引入所需庫

　　import os

　　import re

　　import threading

　　from queue import Queue

　　import requests

　　from lxml import etree

　　class Producer(threading.Thread):

　　"""

　　生產(chǎn)者 - 手機(jī)表情包圖片地址

　　"""

　　def __init__(self, page_queue, img_queue):

　　super(Producer, self).__init__()

　　self.page_queue = page_queue

　　self.img_queue = img_queue

　　def run(self):

　　while True:

　　if self.page_queue.empty():

　　break

　　url = self.page_queue.get()

　　self.parse_page(url)

　　def parse_page(self, url):

　　"""

　　請求 解析 下載

　　:param url:

　　:return:

　　"""

　　# 聲明定義請求頭

　　headers = {

　　'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36',

　　}

　　req = requests.get(url=url, headers=headers)

　　html = req.text

　　tree = etree.HTML(html)

　　imgs = tree.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]')

　　for img in imgs:

　　img_url = img.get('data-original')

　　alt = img.get('alt')

　　alt = re.sub(r'[\??\..,!!\*]]', '', alt)

　　suffix = os.path.splitext(img_url)[1]

　　file_name = alt + suffix

　　self.img_queue.put((img_url, file_name))

　　class Consumer(threading.Thread):

　　"""

　　消費者 - 下載表情包圖片

　　"""

　　def __init__(self, page_queue, img_queue):

　　super(Consumer, self).__init__()

　　self.page_queue = page_queue

　　self.img_queue = img_queue

　　def run(self):

　　# 聲明定義請求頭

　　headers = {

　　'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.110 Safari/537.36',

　　}

　　while True:

　　if self.img_queue.empty() and self.page_queue.empty():

　　break

　　img_url, file_name = self.img_queue.get()

　　req_img = requests.get(url=img_url, headers=headers)

　　with open('images/' + file_name, 'wb') as fp:

　　fp.write(req_img.content)

　　print(file_name + '下載完成...')

　　def main():

　　"""

　　主函數(shù)

　　:return:

　　"""

　　page_queue = Queue(100)

　　img_queue = Queue(1000)

　　for x in range(1, 101):

　　url = 'http://www.doutula.com/photo/list/?page=%d' % x

　　page_queue.put(url)

　　# 定義五個生產(chǎn)者

　　for x in range(6):

　　t = Producer(page_queue=page_queue, img_queue=img_queue)

　　t.start()

　　# 定義三個消費者

　　for x in range(4):

　　t = Consumer(page_queue=page_queue, img_queue=img_queue)

　　t.start()

　　if __name__ == '__main__':

　　main()

　　GIL全局解釋器鎖

　　Python自帶的解釋器是CPython。CPython解釋器的多線程實際上是一個家的多線程(在多核CPU中，只能利用一核，不能利用多核)。同一時刻只有一個線程在執(zhí)行，為了保證同一時刻只有一個線程在執(zhí)行，在CPython解釋器中有一個功能叫做GIL，叫做全局解釋器鎖。這個解釋器鎖是有必要的，因為CPython解釋器的內(nèi)存管理不是線程安全的，當(dāng)然除了CPython解釋器，還有其他的解釋器，有些解釋器是沒有GIL鎖的，見下面：

　　Jpython : 用Java實現(xiàn)的Python解釋器，不存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/Jpython

　　IronPython : 用.NET實現(xiàn)的Python解釋器，不存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/IronPython

　　PyPy : 用Python實現(xiàn)的Python解釋器，存在GIL鎖。更多詳情請見：https://zh.wikipedia.org/wiki/PyPy

　　GIL雖然是一個假的多線程，但是在處理IO操作

關(guān)于Python中如何使用threading問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識。