建站
咨詢
售后
建站咨詢
新聞中心
Python中如何使用threading，針對(duì)這個(gè)問題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問題的小伙伴找到更簡(jiǎn)單易行的方法。
站在用戶的角度思考問題，與客戶深入溝通，找到黃山網(wǎng)站設(shè)計(jì)與黃山網(wǎng)站推廣的解決方案，憑借多年的經(jīng)驗(yàn)，讓設(shè)計(jì)與互聯(lián)網(wǎng)技術(shù)結(jié)合，創(chuàng)造個(gè)性化、用戶體驗(yàn)好的作品，建站類型包括：成都網(wǎng)站設(shè)計(jì)、成都網(wǎng)站制作、企業(yè)官網(wǎng)、英文網(wǎng)站、手機(jī)端網(wǎng)站、網(wǎng)站推廣、申請(qǐng)域名、網(wǎng)頁空間、企業(yè)郵箱。業(yè)務(wù)覆蓋黃山地區(qū)。
Python 版本是3.7.4
前面的文章記錄了網(wǎng)絡(luò)請(qǐng)求(urllib,requests)、數(shù)據(jù)提取(beautiful,xpath,正則)、數(shù)據(jù)存儲(chǔ)(json,csv)的學(xué)習(xí)，下面進(jìn)行一個(gè)多線程的學(xué)習(xí)。
多線程爬蟲
有些時(shí)候，比如下載圖片，因?yàn)橄螺d圖片是一個(gè)耗時(shí)的操作，如果采用之前那種同步的方式下載，那效率會(huì)特別慢。這時(shí)候我們就可以考慮使用多線程的方式來下載圖片。
多線程介紹
多線程是為了同步完成多項(xiàng)任務(wù)，通過提高資源使用來提高系統(tǒng)的效率，線程是在同一時(shí)間需要完成多項(xiàng)任務(wù)的時(shí)候是西納的，最簡(jiǎn)單的比喻多線程就像火車的每一節(jié)車廂，二進(jìn)程就是火車。車廂離開火車是無法跑動(dòng)的，同理火車可以有多節(jié)車廂，多線程的出現(xiàn)是為了提高效率，同時(shí)他的出現(xiàn)也帶來一些問題。
簡(jiǎn)單來講，多線程就相當(dāng)于你原來開了一個(gè)窗口爬取，限制開了十個(gè)窗口來爬取。
threading模塊介紹
threading模塊是Python中專門提供用來做多線程的模塊。threading模塊中最常用的類是Thread。下面一個(gè)簡(jiǎn)單的多線程程序：
# 引入所需庫
import threading
import time
def coding():
"""
coding函數(shù)
:return:
"""
for x in range(5):
print('%s 號(hào)程序員正則寫代碼...' % x)
time.sleep(1)
def drawing():
"""
drawing函數(shù)
:return:
"""
for x in range(5):
print('%s 號(hào)設(shè)計(jì)師正在設(shè)計(jì)圖片...' % x)
time.sleep(1)
def single_thread():
"""
單線程執(zhí)行
:return:
"""
coding()
drawing()
def multi_thread():
"""
多線程執(zhí)行
:return:
"""
# 創(chuàng)建線程
# 注意:target參數(shù)是函數(shù)名,不能帶括號(hào)
t1 = threading.Thread(target=coding, name='coding')
t2 = threading.Thread(target=drawing, name='drawing')
# 啟動(dòng)線程
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類，繼承自這個(gè)類，然后實(shí)現(xiàn)run()方法，線程就會(huì)自動(dòng)運(yùn)行run()方法中的代碼，示例代碼如下：
# 引入所需庫
import threading
import time
class CodingThread(threading.Thread):
"""
寫程序進(jìn)程類
"""
def run(self):
for x in range(5):
print('%s 號(hào)程序員正則寫代碼...' % threading.current_thread())
time.sleep(1)
class DrawingThread(threading.Thread):
"""
設(shè)計(jì)進(jìn)程類
"""
def run(self):
for x in range(5):
print('%s 號(hào)設(shè)計(jì)師正在設(shè)計(jì)圖片...' % threading.current_thread())
time.sleep(1)
def multi_thread():
t1 = CodingThread()
t2 = DrawingThread()
t1.start()
t2.start()
if __name__ == '__main__':
multi_thread()
多線程共享全局變量問題
多線程都是在同一個(gè)進(jìn)程中運(yùn)行的，因此在進(jìn)程中的全局變量所有的線程都是可以共享的。這就造就了一個(gè)問題，因?yàn)榫€程執(zhí)行的順序是無序的，有可能會(huì)造成數(shù)據(jù)錯(cuò)誤。例如如下代碼：
# 引入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
但是由于多線程運(yùn)行的不確定性，因此結(jié)果可能是隨機(jī)的。
鎖機(jī)制
為了解決上述問題由于多線程運(yùn)行的不確定性，threading庫增加了Lock類鎖機(jī)制進(jìn)行處理，當(dāng)某個(gè)線程對(duì)全局變量進(jìn)行修改時(shí)則將此變量加鎖不允許其他線程進(jìn)行修改，知道當(dāng)前線程修改完這個(gè)變量之后再進(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)者和消費(fèi)者模式
生產(chǎn)者和消費(fèi)者模式時(shí)多線程開發(fā)中的經(jīng)常見到的一種模式。生產(chǎn)者的線程專門用來生產(chǎn)一些數(shù)據(jù)，然后存放到一個(gè)中間的變量中。消費(fèi)者再從這個(gè)中間的變量中取出數(shù)據(jù)進(jìn)行消費(fèi)，但是因?yàn)橐褂弥虚g變量，中間變量經(jīng)常是一些全局變量，因此需要使用鎖來保證數(shù)據(jù)的完整性。以下是使用threading.Lock()鎖實(shí)現(xiàn)“生產(chǎn)者與消費(fèi)者模式”的一個(gè)例子：
# 引入所需庫
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):
"""
消費(fèi)者
"""
def run(self):
global gMoney
while True:
money = random.randint(100, 1000)
gLock.acquire()
if gMoney >= money:
gMoney -= money
print('%s消費(fèi)了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))
else:
if gTimes >= 10:
gLock.release()
break
print("%s消費(fèi)者消費(fèi)錢不夠,不消費(fèi)" % threading.current_thread())
gLock.release()
time.sleep(0.5)
def main():
# 定義三個(gè)消費(fèi)者
for x in range(3):
t = Consumer(name='消費(fèi)者線程%d' % x)
t.start()
# 定義五個(gè)生產(chǎn)者
for x in range(5):
t = Producer(name='生產(chǎn)者線程%d' % x)
t.start()
if __name__ == '__main__':
main()
Condition版生產(chǎn)者與消費(fèi)者模式
Lock()版的生產(chǎn)者與消費(fèi)者模式可以正常的運(yùn)行，但是存在一個(gè)不足，在消費(fèi)者中，總是通過while True死循環(huán)并且上鎖的方法去判斷錢夠不夠，上鎖是一個(gè)很好CPU資源的行為。因此這種方式不是最好的，還有一種更好的方式便是使用threading.Condition來實(shí)現(xiàn)。
threading.Condition可以在沒有數(shù)據(jù)的時(shí)候處于阻塞等等狀態(tài)。一旦有合適的數(shù)據(jù)了，還可以使用notify相關(guān)的函數(shù)來通知其他處于等待狀態(tài)的線程，這樣就可以不用做一些無用的上鎖和解鎖的操作，可以提高程序的性能。
首先對(duì)threading.Condition相關(guān)的函數(shù)做個(gè)介紹，threading.Condition類似threading.Lock，可以在修改全部數(shù)據(jù)的時(shí)候進(jìn)行上鎖，也可以在修改完畢后進(jìn)行解鎖。以下將一些常用的函數(shù)做個(gè)簡(jiǎn)單的介紹：
acquire : 上鎖
release : 解鎖
wait : 將當(dāng)前線程處于等待狀態(tài)，并且會(huì)釋放鎖?？梢员黄渌€程使用notify和notify_all函數(shù)喚醒，被喚醒后會(huì)繼續(xù)等待上鎖，上鎖后繼續(xù)執(zhí)行后續(xù)的代碼
notify : 通知某個(gè)正在等待的線程，默認(rèn)是第1個(gè)等待的線程
notify_all : 通知所有正在等待的線程。notify和notify_all不會(huì)釋放鎖，并且需要在release之前掉用
Condition版生產(chǎn)者與消費(fèi)者模式示例代碼如下：
# 引入所需庫
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):
"""
消費(fèi)者
"""
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消費(fèi)者消費(fèi)錢不夠,不消費(fèi)" % threading.current_thread())
gCondition.wait()
gMoney -= money
print('%s消費(fèi)了%d元錢,剩余%d元錢' % (threading.current_thread(), money, gMoney))
gCondition.release()
time.sleep(0.5)
def main():
# 定義三個(gè)消費(fèi)者
for x in range(3):
t = Consumer(name='消費(fèi)者線程%d' % x)
t.start()
# 定義五個(gè)生產(chǎn)者
for x in range(5):
t = Producer(name='生產(chǎn)者線程%d' % x)
t.start()
if __name__ == '__main__':
main()
Queue線程安全隊(duì)列
在線程中，訪問一些全局變量，加鎖是一個(gè)經(jīng)常的過程，如果你想把一些數(shù)據(jù)存儲(chǔ)到莫格隊(duì)列中，那么Python內(nèi)置了一個(gè)線程安全的模塊叫queue模塊。Python中的queue模塊中提供了同步的、線程安全的對(duì)咧咧，包括FIFO(先進(jìn)先出)隊(duì)列Queue，LIFO(后入先出)隊(duì)列LifoQueue。這些隊(duì)列都實(shí)現(xiàn)了所原理(可以理解為原子操作，即要么不做，要么都做完)，能夠在多線程中直接使用。可以使用隊(duì)列來實(shí)現(xiàn)線程間的同步，相關(guān)函數(shù)如下：
初始化Queue(maxsize) : 創(chuàng)建一個(gè)先進(jìn)先出的隊(duì)列
qsize() : 返回隊(duì)列的大小
empty() : 判斷隊(duì)列是否為空
full() : 判斷隊(duì)列是否已滿
get() : 從隊(duì)列中獲取最后一個(gè)數(shù)據(jù)
put() : 將一個(gè)數(shù)據(jù)放到隊(duì)列中
使用代碼示例：
# 引入所需庫
import threading
import time
from queue import Queue
def set_value(q):
"""
寫入隊(duì)列
:param q:
:return:
"""
index = 1
while True:
q.put(index)
index += 1
time.sleep(3)
def get_value(q):
"""
從隊(duì)列取值
: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()
使用實(shí)例
單線程爬取表情包，實(shí)例代碼如下：
# 引入所需庫
import os
import re
import requests
from lxml import etree
def parse_page(url):
"""
請(qǐng)求 解析 下載
:param url:
:return:
"""
# 聲明定義請(qǐng)求頭
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()
多線程爬取表情包，實(shí)例代碼如下：
# 引入所需庫
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):
"""
請(qǐng)求 解析 下載
:param url:
:return:
"""
# 聲明定義請(qǐng)求頭
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):
"""
消費(fèi)者 - 下載表情包圖片
"""
def __init__(self, page_queue, img_queue):
super(Consumer, self).__init__()
self.page_queue = page_queue
self.img_queue = img_queue
def run(self):
# 聲明定義請(qǐng)求頭
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)
# 定義五個(gè)生產(chǎn)者
for x in range(6):
t = Producer(page_queue=page_queue, img_queue=img_queue)
t.start()
# 定義三個(gè)消費(fèi)者
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解釋器的多線程實(shí)際上是一個(gè)家的多線程(在多核CPU中，只能利用一核，不能利用多核)。同一時(shí)刻只有一個(gè)線程在執(zhí)行，為了保證同一時(shí)刻只有一個(gè)線程在執(zhí)行，在CPython解釋器中有一個(gè)功能叫做GIL，叫做全局解釋器鎖。這個(gè)解釋器鎖是有必要的，因?yàn)镃Python解釋器的內(nèi)存管理不是線程安全的，當(dāng)然除了CPython解釋器，還有其他的解釋器，有些解釋器是沒有GIL鎖的，見下面：
Jpython : 用Java實(shí)現(xiàn)的Python解釋器，不存在GIL鎖。更多詳情請(qǐng)見：https://zh.wikipedia.org/wiki/Jpython
IronPython : 用.NET實(shí)現(xiàn)的Python解釋器，不存在GIL鎖。更多詳情請(qǐng)見：https://zh.wikipedia.org/wiki/IronPython
PyPy : 用Python實(shí)現(xiàn)的Python解釋器，存在GIL鎖。更多詳情請(qǐng)見：https://zh.wikipedia.org/wiki/PyPy
GIL雖然是一個(gè)假的多線程，但是在處理IO操作
關(guān)于Python中如何使用threading問題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識(shí)。
新聞名稱：Python中如何使用threading
瀏覽路徑：http://www.dlmjj.cn/article/jpdohs.html
