1.進(jìn)程
進(jìn)程是一個(gè)具有一定獨(dú)立功能的程序在一個(gè)數(shù)據(jù)集上的一次動態(tài)執(zhí)行的過程，是操作系統(tǒng)進(jìn)行資源分配和調(diào)度的一個(gè)獨(dú)立單位，是應(yīng)用程序運(yùn)行的載體。進(jìn)程是一種抽象的概念，從來沒有統(tǒng)一的標(biāo)準(zhǔn)定義。進(jìn)程一般由程序、數(shù)據(jù)集合和進(jìn)程控制塊三部分組成。程序用于描述進(jìn)程要完成的功能，是控制進(jìn)程執(zhí)行的指令集；數(shù)據(jù)集合是程序在執(zhí)行時(shí)所需要的數(shù)據(jù)和工作區(qū)；程序控制塊包含進(jìn)程的描述信息和控制信息是進(jìn)程存在的唯一標(biāo)志。
進(jìn)程具有的特征：
- 動態(tài)性：進(jìn)程是程序的一次執(zhí)行過程，是臨時(shí)的，有生命期的，是動態(tài)產(chǎn)生，動態(tài)消亡的。
- 并發(fā)性：任何進(jìn)程都可以同其他進(jìn)程一起并發(fā)執(zhí)行。
- 獨(dú)立性：進(jìn)程是系統(tǒng)進(jìn)行資源分配和調(diào)度的一個(gè)獨(dú)立單位。
- 結(jié)構(gòu)性：進(jìn)程由程序、數(shù)據(jù)和進(jìn)程控制塊三部分組成。
實(shí)現(xiàn)多進(jìn)程
import multiprocessing
import time
def run1(sleep_time):
while True:
print("-- 1 --")
time.sleep(sleep_time)
def run2(sleep_time):
while True:
print("-- 2 --")
time.sleep(sleep_time)
def main():
# 創(chuàng)建進(jìn)程對象。
# target：指定線程調(diào)用的函數(shù)名。注:等號后跟方法名不能加括號，如果加了也能執(zhí)行函數(shù)但threading功能無效
# args：指定調(diào)用函數(shù)時(shí)傳遞的參數(shù)。注:args是一個(gè)數(shù)組變量參數(shù)，只傳一個(gè)參數(shù)時(shí)，需要在參數(shù)后面添加逗號
p1 = multiprocessing.Process(target=run1, args=(1,))
p2 = multiprocessing.Process(target=run2, args=(1,))
# 啟用子進(jìn)程
p1.start()
p2.start()
# join方法等待子進(jìn)程執(zhí)行結(jié)束
p1.join()
p2.join()
print("子進(jìn)程結(jié)束")
if __name__ == "__main__":
main()
運(yùn)行上面代碼，查看任務(wù)管理器python的啟動進(jìn)程數(shù)。
代碼中只啟動了兩個(gè)子進(jìn)程，但是為什么有3個(gè)python進(jìn)程？這是因?yàn)椋琾ython會創(chuàng)建一個(gè)主進(jìn)程（第1個(gè)進(jìn)程），當(dāng)運(yùn)行到p1.start()時(shí)會創(chuàng)建一個(gè)子進(jìn)程（第2個(gè)進(jìn)程），當(dāng)運(yùn)行到p2.start()時(shí)又會創(chuàng)建一個(gè)子進(jìn)程（第3個(gè)進(jìn)程）
2.進(jìn)程池
進(jìn)程的創(chuàng)建和刪除是需要消耗計(jì)算機(jī)資源的，如果有大量任務(wù)需要多進(jìn)程完成，則可能需要頻繁的創(chuàng)建刪除進(jìn)程，這會給計(jì)算機(jī)帶來較多的資源消耗。進(jìn)程池的出現(xiàn)解決了這個(gè)問題，它的原理是創(chuàng)建適當(dāng)?shù)倪M(jìn)程放入進(jìn)程池，等待待處理的事件，當(dāng)處理完事件后進(jìn)程不會銷毀，仍然在進(jìn)程池中等待處理其他事件，直到事件全部處理完畢，進(jìn)程退出。 進(jìn)程的復(fù)用降低了資源的消耗。
實(shí)現(xiàn)進(jìn)程池
import time, os
from multiprocessing import Pool
def worker(msg):
start_time = time.time()
print(F"{msg}開始執(zhí)行，進(jìn)程pid為{os.getpid()}")
time.sleep(1)
end_time = time.time()
print(F"{msg}執(zhí)行完畢，耗時(shí){end_time - start_time}")
def main():
po = Pool(3) # 定義進(jìn)程池最大進(jìn)程數(shù)為3
for i in range(10):
# 每次循環(huán)會用空閑出的子進(jìn)程調(diào)用目標(biāo)
po.apply_async(worker, args=(i,)) # 若調(diào)用的函數(shù)報(bào)錯(cuò)，進(jìn)程池中不會打印報(bào)錯(cuò)信息
po.close() # 關(guān)閉進(jìn)程池，關(guān)閉后，不再接收新的目標(biāo)
po.join() # 等待進(jìn)程池中所有子進(jìn)程執(zhí)行完，必須放在close()之后。若沒有join()操作，主進(jìn)程執(zhí)行完后直接關(guān)閉
print("--end--")
if __name__ == "__main__":
main()
3.線程
在早期的操作系統(tǒng)中并沒有線程的概念，進(jìn)程是擁有資源和獨(dú)立運(yùn)行的最小單位，也是程序執(zhí)行的最小單位。任務(wù)調(diào)度采用的是時(shí)間片輪轉(zhuǎn)的搶占式調(diào)度方式，而進(jìn)程是任務(wù)調(diào)度的最小單位，每個(gè)進(jìn)程有各自獨(dú)立的一塊內(nèi)存，使得各個(gè)進(jìn)程之間內(nèi)存地址相互隔離。后來，隨著計(jì)算機(jī)的發(fā)展，對CPU的要求越來越高，進(jìn)程之間的切換開銷較大，已經(jīng)無法滿足越來越復(fù)雜的程序的要求了。于是就發(fā)明了線程，線程是程序執(zhí)行中一個(gè)單一的順序控制流程，是程序執(zhí)行流的最小單元，是處理器調(diào)度和分派的基本單位。一個(gè)進(jìn)程可以有一個(gè)或多個(gè)線程，各個(gè)線程之間共享程序的內(nèi)存空間(也就是所在進(jìn)程的內(nèi)存空間)。一個(gè)標(biāo)準(zhǔn)的線程由線程ID，當(dāng)前指令指針PC，寄存器和堆棧組成。而進(jìn)程由內(nèi)存空間(代碼，數(shù)據(jù)，進(jìn)程空間，打開的文件)和一個(gè)或多個(gè)線程組成。
實(shí)現(xiàn)多線程
import time
import threading
def say(sleep_time):
for i in range(5):
print(f"說{i+1}下")
time.sleep(sleep_time)
def dance():
for i in range(10):
print(f"跳{i+1}下")
time.sleep(1)
def main():
# 創(chuàng)建線程對象
# target：指定線程調(diào)用的函數(shù)名。注:等號后跟方法名不能加括號，如果加了也能執(zhí)行函數(shù)但threading功能無效
# args：指定調(diào)用函數(shù)時(shí)傳遞的參數(shù)。注:args是一個(gè)數(shù)組變量參數(shù)，只傳一個(gè)參數(shù)時(shí)，需要在參數(shù)后面添加逗號
t1 = threading.Thread(target=say, args=(1,))
t2 = threading.Thread(target=dance)
# 啟動線程
t1.start()
t2.start()
# 查看正在運(yùn)行的線程
while True:
now_threading = threading.enumerate()
print(now_threading)
# 當(dāng)子線程全部運(yùn)行結(jié)束后，僅剩1個(gè)主線程
if len(now_threading) <= 1:
break
time.sleep(1)
if __name__ == "__main__":
main()
多線程的資源競爭問題
因?yàn)槎嗑€程共享全局變量，當(dāng)線程還沒執(zhí)行完當(dāng)前任務(wù)，操作系統(tǒng)就自動輪流調(diào)度執(zhí)行其他任務(wù)，就可能會產(chǎn)生資源競爭的問題。
比如下例中，執(zhí)行 g_num+=1 時(shí)，會將其分成3步執(zhí)行：1.取值；2.運(yùn)算;3.保存運(yùn)算結(jié)果，在CPU執(zhí)行任務(wù)時(shí)，若剛運(yùn)行1 2 步就交替執(zhí)行下一個(gè)任務(wù)，再返回來保存結(jié)果，因?yàn)楣蚕砣肿兞?，此時(shí)運(yùn)算結(jié)果可能已被重新賦值。
import time
import threading
g_num = 0
def sum1(num):
global g_num
for i in range(num):
g_num += 1
print(F"sum1：{g_num}")
def sum2(num):
global g_num
for i in range(num):
g_num += 1
print(F"sum2：{g_num}")
def main():
t1 = threading.Thread(target=sum1, args=(1000000,))
t2 = threading.Thread(target=sum2, args=(1000000,))
t1.start()
t2.start()
time.sleep(2)
print(g_num) # 執(zhí)行后，預(yù)期結(jié)果為2000000；實(shí)際不是
if __name__ == "__main__":
main()
執(zhí)行結(jié)果
從結(jié)果可以看出，sum1和sum2不為1000000，總和不為2000000，這就是上面說的資源競爭問題
互斥鎖解決資源競爭問題
import threading
import time
# 定義一個(gè)全局變量
g_num = 0
# 創(chuàng)建一個(gè)互斥鎖，默認(rèn)是沒有上鎖的
mutex = threading.Lock()
def sum1(num):
global g_num
# mutex.acquire() # 若在此處上鎖，要等下面循環(huán)執(zhí)行完才會解鎖，若循環(huán)時(shí)間太長，會導(dǎo)致另外的線程堵塞等待。
for i in range(num):
# 上鎖，如果之前沒有被上鎖，那么此時(shí)上鎖成功。 上鎖原則：一般對產(chǎn)生資源競爭的代碼上鎖。如果上鎖之前 已經(jīng)被上鎖了，那么此時(shí)會堵塞在這里，直到 這個(gè)鎖被解開為止。
mutex.acquire()
g_num += 1
# 解鎖
mutex.release()
print("-----in test1 g_num=%d----" % g_num)
def sum2(num):
global g_num
for i in range(num):
mutex.acquire()
g_num += 1
mutex.release()
print("-----in test2 g_num=%d=----" % g_num)
def main():
t1 = threading.Thread(target=sum1, args=(1000000,))
t2 = threading.Thread(target=sum2, args=(1000000,))
t1.start()
t2.start()
# 等待上面的2個(gè)線程執(zhí)行完畢....
time.sleep(2)
print("-----in main Thread g_num = %d---" % g_num)
if __name__ == "__main__":
main()
運(yùn)行結(jié)果
死鎖
在線程間共享多個(gè)資源的時(shí)候，如果兩個(gè)線程分別占用部分資源并且同時(shí)等待對方的資源，就會造成死鎖。盡管死鎖很少發(fā)生，但一旦發(fā)生就會造成應(yīng)用停止響應(yīng)。下面看一個(gè)死鎖例子。
import time
import threading
# 創(chuàng)建多個(gè)鎖
mutexA = threading.Lock()
mutexB = threading.Lock()
def print1():
mutexA.acquire()
time.sleep(2) # 等待B鎖穩(wěn)定
print("打印A1")
mutexB.acquire()
print("打印B1")
mutexB.release()
mutexA.release()
def print2():
mutexB.acquire()
time.sleep(1) # 等待A鎖穩(wěn)定
print("打印B2")
mutexA.acquire()
print("打印A2")
mutexA.release()
mutexB.release()
def main():
t1 = threading.Thread(target=print1)
t2 = threading.Thread(target=print2)
t1.start()
t2.start()
if __name__ == "__main__":
main()
執(zhí)行結(jié)果
避免死索辦法：1、添加超時(shí)時(shí)間；2、銀行家算法（讓鎖按預(yù)期上鎖和解鎖）
4.協(xié)程
協(xié)程，又稱微線程。協(xié)程的作用是在執(zhí)行函數(shù)A時(shí)可以隨時(shí)中斷去執(zhí)行函數(shù)B，然后中斷函數(shù)B繼續(xù)執(zhí)行函數(shù)A（可以自由切換）。但這一過程并不是函數(shù)調(diào)用，這一整個(gè)過程看似像多線程，然而協(xié)程只有一個(gè)線程執(zhí)行。
協(xié)程的優(yōu)勢：
- 執(zhí)行效率極高，因?yàn)樽映绦蚯袚Q（函數(shù)）不是線程切換，由程序自身控制，沒有切換線程的開銷。所以與多線程相比，線程的數(shù)量越多，協(xié)程性能的優(yōu)勢越明顯。
- 不需要多線程的鎖機(jī)制，因?yàn)橹挥幸粋€(gè)線程，也不存在同時(shí)寫變量沖突，在控制共享資源時(shí)也不需要加鎖，因此執(zhí)行效率高很多。
gevent
gevent是第三方庫，通過 greenlet 實(shí)現(xiàn) coroutine，創(chuàng)建、調(diào)度的開銷比 線程(thread) 還小，因此程序內(nèi)部的 執(zhí)行流 效率高。
其基本思想是：當(dāng)一個(gè)greenlet遇到IO操作時(shí)，比如訪問網(wǎng)絡(luò)，就自動切換到其他的greenlet，等到IO操作完成，再在適當(dāng)?shù)臅r(shí)候切換回來繼續(xù)執(zhí)行。由于IO操作非常耗時(shí)，經(jīng)常使程序處于等待狀態(tài)，有了gevent為我們自動切換協(xié)程，就保證總有g(shù)reenlet在運(yùn)行，而不是等待IO。
gevent常用方法：
- gevent.spawn() 創(chuàng)建一個(gè)普通的Greenlet對象并切換
- gevent.spawn_later(seconds=3) 延時(shí)創(chuàng)建一個(gè)普通的Greenlet對象并切換
- gevent.spawn_raw() 創(chuàng)建的協(xié)程對象屬于一個(gè)組
- gevent.getcurrent() 返回當(dāng)前正在執(zhí)行的greenlet
- gevent.joinall(jobs) 將協(xié)程任務(wù)添加到事件循環(huán)，接收一個(gè)任務(wù)列表
- gevent.wait() 可以替代join函數(shù)等待循環(huán)結(jié)束，也可以傳入?yún)f(xié)程對象列表
- gevent.kill() 殺死一個(gè)協(xié)程
- gevent.killall() 殺死一個(gè)協(xié)程列表里的所有協(xié)程
- monkey.patch_all() 非常重要，會自動將python的一些標(biāo)準(zhǔn)模塊替換成gevent框架
import gevent
def task(n):
for i in range(n):
print(gevent.getcurrent(), i)
if __name__ == '__main__':
g1 = gevent.spawn(task, 3)
g2 = gevent.spawn(task, 3)
g3 = gevent.spawn(task, 3)
g1.join()
g2.join()
g3.join()
運(yùn)行結(jié)果
可以看到3個(gè)greenlet是依次運(yùn)行而不是交替運(yùn)行。要讓greenlet交替運(yùn)行，可以通過gevent.sleep()交出控制權(quán)：
import gevent
def task(n):
for i in range(n):
print(gevent.getcurrent(), i)
gevent.sleep(1)
if __name__ == '__main__':
g1 = gevent.spawn(task, 3)
g2 = gevent.spawn(task, 3)
g3 = gevent.spawn(task, 3)
g1.join()
g2.join()
g3.join()
運(yùn)行結(jié)果
當(dāng)然在實(shí)際的代碼里，我們不會用gevent.sleep()去切換協(xié)程，而是在執(zhí)行到IO操作時(shí)gevent會自動完成，所以gevent需要將Python自帶的一些標(biāo)準(zhǔn)庫的運(yùn)行方式由阻塞式調(diào)用變?yōu)閰f(xié)作式運(yùn)行。這一過程在啟動時(shí)通過monkey patch完成：
import time
import gevent
from gevent import monkey
# 猴子補(bǔ)丁，會自動將python的一些標(biāo)準(zhǔn)模塊替換成gevent框架。慎用，它創(chuàng)造了“隱式的副作用”，如果出現(xiàn)問題 它很多時(shí)候是極難調(diào)試的。
monkey.patch_all() # 注意：若導(dǎo)出的模塊函數(shù)不會被替換，比如from time import sleep，sleep不會被替換
def task(n):
for i in range(n):
print(gevent.getcurrent(), i)
time.sleep(1) # 會被gevent自動替換為gevent.sleep()
if __name__ == '__main__':
g1 = gevent.spawn(task, 3)
g2 = gevent.spawn(task, 3)
g3 = gevent.spawn(task, 3)
g1.join()
g2.join()
g3.join()
執(zhí)行結(jié)果
上面的流程看起來比較繁瑣，可以使用 gevent.joinall() 方法簡化流程：文章來源：http://www.zghlxwxcb.cn/news/detail-533444.html
import time
import gevent
from gevent import monkey
# 猴子補(bǔ)丁，會自動將python的一些標(biāo)準(zhǔn)模塊替換成gevent框架。慎用，它創(chuàng)造了“隱式的副作用”，如果出現(xiàn)問題 它很多時(shí)候是極難調(diào)試的。
monkey.patch_all() # 注意：若導(dǎo)出的模塊函數(shù)不會被替換，比如from time import sleep，sleep不會被替換
'''
學(xué)習(xí)中遇到問題沒人解答？小編創(chuàng)建了一個(gè)Python學(xué)習(xí)交流群：711312441
尋找有志同道合的小伙伴，互幫互助,群里還有不錯(cuò)的視頻學(xué)習(xí)教程和PDF電子書！
'''
def task(n):
for i in range(n):
print(gevent.getcurrent(), i)
time.sleep(1) # 會被gevent自動替換為gevent.sleep()
if __name__ == '__main__':
gevent.joinall([
gevent.spawn(task, 4),
gevent.spawn(task, 4),
gevent.spawn(task, 4),
])
執(zhí)行結(jié)果
文章來源地址http://www.zghlxwxcb.cn/news/detail-533444.html
到了這里，關(guān)于Python多任務(wù)教程：進(jìn)程、線程、協(xié)程的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
