一、 什么是進(jìn)程 / 線程

1、 引論

眾所周知，CPU是計算機(jī)的核心，它承擔(dān)了所有的計算任務(wù)。而操作系統(tǒng)是計算機(jī)的管理者，是一個大管家，它負(fù)責(zé)任務(wù)的調(diào)度，資源的分配和管理，統(tǒng)領(lǐng)整個計算機(jī)硬件。應(yīng)用程序是具有某種功能的程序，程序運(yùn)行與操作系統(tǒng)之上

2、 線程

在很早的時候計算機(jī)并沒有線程這個概念，但是隨著時代的發(fā)展，只用進(jìn)程來處理程序出現(xiàn)很多的不足。如當(dāng)一個進(jìn)程堵塞時，整個程序會停止在堵塞處，并且如果頻繁的切換進(jìn)程，會浪費(fèi)系統(tǒng)資源。所以線程出現(xiàn)了

線程是能擁有資源和獨(dú)立運(yùn)行的最小單位，也是程序執(zhí)行的最小單位。一個進(jìn)程可以擁有多個線程，而且屬于同一個進(jìn)程的多個線程間會共享該進(jìn)行的資源

3、 進(jìn)程

進(jìn)程時一個具有一定功能的程序在一個數(shù)據(jù)集上的一次動態(tài)執(zhí)行過程。進(jìn)程由程序，數(shù)據(jù)集合和進(jìn)程控制塊三部分組成。程序用于描述進(jìn)程要完成的功能，是控制進(jìn)程執(zhí)行的指令集；數(shù)據(jù)集合是程序在執(zhí)行時需要的數(shù)據(jù)和工作區(qū)；程序控制塊（PCB）包含程序的描述信息和控制信息，是進(jìn)程存在的唯一標(biāo)志

4、 區(qū)別

一個進(jìn)程由一個或者多個線程組成，線程是一個進(jìn)程中代碼的不同執(zhí)行路線
切換進(jìn)程需要的資源比切換線程的要多的多
進(jìn)程之間相互獨(dú)立，而同一個進(jìn)程下的線程共享程序的內(nèi)存空間（如代碼段，數(shù)據(jù)集，堆棧等）。某進(jìn)程內(nèi)的線程在其他進(jìn)程不可見。換言之，線程共享同一片內(nèi)存空間，而進(jìn)程各有獨(dú)立的內(nèi)存空間

5、 使用

在Python中，通過兩個標(biāo)準(zhǔn)庫 thread 和 Threading 提供對線程的支持, threading 對 thread 進(jìn)行了封裝。 threading 模塊中提供了 Thread , Lock , RLOCK , Condition 等組件

二、 多線程使用

在Python中線程和進(jìn)程的使用就是通過 Thread 這個類。這個類在我們的 thread 和 threading 模塊中。我們一般通過 threading 導(dǎo)入

默認(rèn)情況下，只要在解釋器中，如果沒有報錯，則說明線程可用

from threading import Thread

1、 常用方法

• Thread.run(self)  # 線程啟動時運(yùn)行的方法，由該方法調(diào)用 target 參數(shù)所指定的函數(shù)
• Thread.start(self)  # 啟動線程，start 方法就是去調(diào)用 run 方法
• Thread.terminate(self)  # 強(qiáng)制終止線程
• Thread.join(self, timeout)  # 阻塞調(diào)用，主線程進(jìn)行等待
• Thread.setDaemon(self, daemonic)  # 將子線程設(shè)置為守護(hù)線程
• Thread.getName(self, name)  # 獲取線程名稱
• Thread.setName(self, name)  # 設(shè)置線程名稱

2、 常用參數(shù)

3、 多線程的應(yīng)用

3.1 重寫線程法

import time, queue, threading


class MyThread(threading.Thread):

? ? def __init__(self):
? ? ? ? super().__init__()
? ? ? ? self.daemon = True ?# 開啟守護(hù)模式
? ? ? ? self.queue = queue.Queue(3) ?# 開啟隊列對象，存儲三個任務(wù)
? ? ? ? self.start() ?# 實(shí)例化的時候直接啟動線程，不需要手動啟動線程

? ? def run(self) -> None: ?# run方法線程自帶的方法,內(nèi)置方法，在線程運(yùn)行時會自動調(diào)用
? ? ? ? while True: ?# 不斷處理任務(wù)
? ? ? ? ? ? func, args, kwargs = self.queue.get()
? ? ? ? ? ? func(*args, **kwargs) ?# 調(diào)用函數(shù)執(zhí)行任務(wù) 元組不定長記得一定要拆包
? ? ? ? ? ? self.queue.task_done() ?# 解決一個任務(wù)就讓計數(shù)器減一，避免阻塞

? ? # 生產(chǎn)者模型
? ? def submit_tasks(self, func, args=(), kwargs={}): ?# func為要執(zhí)行的任務(wù)，加入不定長參數(shù)使用（默認(rèn)使用默認(rèn)參數(shù)）
? ? ? ? self.queue.put((func, args, kwargs)) ?# 提交任務(wù)

? ? # 重寫join方法
? ? def join(self) -> None:
? ? ? ? self.queue.join() ?# 查看隊列計時器是否為0 任務(wù)為空 為空關(guān)閉隊列
? ? ? ??
? ? ? ??
def f2(*args, **kwargs):
? ? time.sleep(2)
? ? print("任務(wù)2完成", args, kwargs)

# 實(shí)例化線程對象
mt = MyThread()
# 提交任務(wù)
mt.submit_tasks(f2, args=("aa", "aasd"), kwargs={"a": 2, "s": 3})

# 讓主線程等待子線程結(jié)束再結(jié)束
mt.join()

守護(hù)模式：

主線程在其他非守護(hù)線程運(yùn)行完畢后才算運(yùn)行完畢（守護(hù)線程在此時就被回收）。因?yàn)橹骶€程的結(jié)束意味著進(jìn)程的結(jié)束，進(jìn)程整體的資源都將被回收，而進(jìn)程必須保證非守護(hù)線程都運(yùn)行完畢后才能結(jié)束

3.2 直接調(diào)用法

def f2(i):
? ? time.sleep(2)
? ? print("任務(wù)2完成", i)

lis = []
for i in range(5):
? ? t = Thread(target=f2, args=(i,))
? ? t.start() ?# 啟動 5 個線程
? ? lis.append(t)

for i in lis:
? ? i.join() ?# 線程等待

4、 線程間數(shù)據(jù)的共享

現(xiàn)在我們程序代碼中，有多個線程， 并且在這個幾個線程中都會去 操作同一部分內(nèi)容，那么如何實(shí)現(xiàn)這些數(shù)據(jù)的共享呢？

這時，可以使用 threading庫里面的鎖對象 Lock 去保護(hù)

Lock 對象的acquire方法 是申請鎖

每個線程在操作共享數(shù)據(jù)對象之前，都應(yīng)該申請獲取操作權(quán)，也就是調(diào)用該共享數(shù)據(jù)對象對應(yīng)的鎖對象的acquire方法，如果線程A 執(zhí)行了acquire() 方法，別的線程B 已經(jīng)申請到了這個鎖， 并且還沒有釋放，那么 線程A的代碼就在此處 等待 線程B 釋放鎖，不去執(zhí)行后面的代碼。

直到線程B 執(zhí)行了鎖的 release 方法釋放了這個鎖， 線程A 才可以獲取這個鎖，就可以執(zhí)行下面的代碼了

如：

import threading

var = 1
# 添加互斥鎖，并且拿到鎖
lock = threading.Lock()

# 定義兩個線程要用做的任務(wù)
def func1():
? ? global var ?# 聲明全局變量
? ? for i in range(1000000):
? ? ? ? lock.acquire() ?# 操作前上鎖
? ? ? ? var += i
? ? ? ? lock.release() ?# 操作完后釋放鎖
? ? ? ??
def func2():
? ? global var ?# 聲明全局變量
? ? for i in range(1000000): ? ? ??
?? ??? ?lock.acquire() ?# 操作前上鎖 ? ?
? ? ? ? var -= i
? ? ? ? lock.release() ?# 操作完后釋放鎖
? ? ? ??
# 創(chuàng)建2個線程
t1 = threading.Thread(target=func1)
t2 = threading.Thread(target=func2)
t1.start()
t2.start()
t1.join()
t2.join()
print(var)

到在使用多線程時，如果數(shù)據(jù)出現(xiàn)和自己預(yù)期不符的問題，就可以考慮是否是共享的數(shù)據(jù)被調(diào)用覆蓋的問題

使用 threading 庫里面的鎖對象 Lock 去保護(hù)

三、 多進(jìn)程使用

1、 簡介

Python中的多進(jìn)程是通過multiprocessing包來實(shí)現(xiàn)的，和多線程的threading.Thread差不多，它可以利用multiprocessing.Process對象來創(chuàng)建一個進(jìn)程對象。這個進(jìn)程對象的方法和線程對象的方法差不多也有start(), run(), join()等方法，其中有一個方法不同Thread線程對象中的守護(hù)線程方法是setDeamon，而Process進(jìn)程對象的守護(hù)進(jìn)程是通過設(shè)置daemon屬性來完成的

2、 應(yīng)用

2.1 重寫進(jìn)程法

import time
from multiprocessing import Process


class MyProcess(Process): ?# 繼承Process類
? ? def __init__(self, target, args=(), kwargs={}):
? ? ? ? super(MyProcess, self).__init__()
? ? ? ? self.daemon = True ?# 開啟守護(hù)進(jìn)程
? ? ? ? self.target = target
? ? ? ? self.args = args
? ? ? ? self.kwargs = kwargs
? ? ? ? self.start() ?# 自動開啟進(jìn)程

? ? def run(self):
? ? ? ? self.target(*self.args, **self.kwargs)


def fun(*args, **kwargs):
? ? print(time.time())
? ? print(args[0])


if __name__ == '__main__':
? ? lis = []
? ? for i in range(5):
? ? ? ? p = MyProcess(fun, args=(1, ))
? ? ? ? lis.append(p)
? ? for i in lis:
? ? ? ? i.join() ?# 讓進(jìn)程等待

守護(hù)模式：

主進(jìn)程在其代碼結(jié)束后就已經(jīng)算運(yùn)行完畢了（守護(hù)進(jìn)程在此時就被回收）,然后主進(jìn)程會一直等非守護(hù)的子進(jìn)程都運(yùn)行完畢后回收子進(jìn)程的資源(否則會產(chǎn)生僵尸進(jìn)程)，才會結(jié)束

2.2 直接調(diào)用法

import time
from multiprocessing import ?Process

def fun(*args, **kwargs):
? ? print(time.time())
? ? print(args[0])

if __name__ == '__main__':
? ? lis = []
? ? for i in range(5):
? ? ? ? p = Process(target=fun, args=(1, ))
? ? ? ? lis.append(p)
? ? for i in lis:
? ? ? ? i.join() ?# 讓進(jìn)程等待

3、 進(jìn)程之間的數(shù)據(jù)共享

3.1 Lock 方法

其使用方法和線程的那個 Lock 使用方法類似

3.2 Manager 方法

Manager的作用是提供多進(jìn)程共享的全局變量，Manager()方法會返回一個對象，該對象控制著一個服務(wù)進(jìn)程，該進(jìn)程中保存的對象運(yùn)行其他進(jìn)程使用代理進(jìn)行操作

Manager支持的類型有：list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array

語法：

from multiprocessing import Process, Lock, Manager

def f(n, d, l, lock):
? ? lock.acquire()
? ? d[str(n)] = n
? ? l[n] = -99
? ? lock.release()

if __name__ == '__main__':
? ? lock = Lock()
? ? with Manager() as manager:
? ? ? ? d = manager.dict() ?# 空字典
? ? ? ? l = manager.list(range(10)) ?# [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
? ? ? ? # 啟動10個進(jìn)程，不同的進(jìn)程對d和l中的不同元素進(jìn)行操作
? ? ? ? for i in range(10):
? ? ? ? ? ? p = Process(target=f, args=(i, d, l, lock))
? ? ? ? ? ? p.start()
? ? ? ? ? ? p.join()

? ? ? ? print(d)
? ? ? ? print(l)

四、 池并發(fā)

1、 語法

線程池的基類是 concurrent.futures 模塊中的 Executor ， Executor 提供了兩個子類，即 ThreadPoolExecutor 和 ProcessPoolExecutor ，其中 ThreadPoolExecutor 用于創(chuàng)建線程池，而ProcessPoolExecutor 用于創(chuàng)建進(jìn)程池

如果使用線程池/進(jìn)程池來管理并發(fā)編程，那么只要將相應(yīng)的 task 函數(shù)提交給線程池/進(jìn)程池，剩下的事情就由線程池/進(jìn)程池來搞定

Exectuor 提供了如下常用方法：

• submit(fn, *args, * kwargs)：將 fn 函數(shù)提交給線程池。 args 代表傳給 fn 函數(shù)的參數(shù)，*kwargs 代表以關(guān)鍵字參數(shù)的形式為 fn 函數(shù)傳入?yún)?shù)
• map(func, *iterables, timeout=None, chunksize=1)：該函數(shù)類似于全局函數(shù) map(func, *iterables)，只是該函數(shù)將會啟動多個線程，以異步方式立即對 iterables 執(zhí)行 map 處理。
• shutdown(wait=True)：關(guān)閉線程池

程序?qū)?task 函數(shù)提交（submit）給線程池后，submit 方法會返回一個 Future 對象，F(xiàn)uture 類主要用于獲取線程任務(wù)函數(shù)的返回值。由于線程任務(wù)會在新線程中以異步方式執(zhí)行，因此，線程執(zhí)行的函數(shù)相當(dāng)于一個“將來完成”的任務(wù)，所以 Python 使用 Future 來代表

Future 提供了如下方法：

• cancel()：取消該 Future 代表的線程任務(wù)。如果該任務(wù)正在執(zhí)行，不可取消，則該方法返回 False；否則，程序會取消該任務(wù)，并返回 True。
• cancelled()：返回 Future 代表的線程任務(wù)是否被成功取消。
• running()：如果該 Future 代表的線程任務(wù)正在執(zhí)行、不可被取消，該方法返回 True。
• done()：如果該 Funture 代表的線程任務(wù)被成功取消或執(zhí)行完成，則該方法返回 True。
• result(timeout=None)：獲取該 Future 代表的線程任務(wù)最后返回的結(jié)果。如果 Future 代表的線程任務(wù)還未完成，該方法將會阻塞當(dāng)前線程，其中 timeout 參數(shù)指定最多阻塞多少秒。
• exception(timeout=None)：獲取該 Future 代表的線程任務(wù)所引發(fā)的異常。如果該任務(wù)成功完成，沒有異常，則該方法返回 None。
• add_done_callback(fn)：為該 Future 代表的線程任務(wù)注冊一個“回調(diào)函數(shù)”，當(dāng)該任務(wù)成功完成時，程序會自動觸發(fā)該 fn 函數(shù)

2、 獲取 CPU 數(shù)量

from multiprocessing import cpu_count ?# cpu核心數(shù)模塊，其可以獲取 CPU 核心數(shù)

n = cpu_count() ?# 獲取cpu核心數(shù)

3、 線程池

使用線程池來執(zhí)行線程任務(wù)的步驟如下：

• 調(diào)用 ThreadPoolExecutor 類的構(gòu)造器創(chuàng)建一個線程池
• 定義一個普通函數(shù)作為線程任務(wù)
• 調(diào)用 ThreadPoolExecutor 對象的 submit() 方法來提交線程任務(wù)
• 當(dāng)不想提交任何任務(wù)時，調(diào)用 ThreadPoolExecutor 對象的 shutdown() 方法來關(guān)閉線程池

from concurrent.futures import ThreadPoolExecutor
import threading
import time
# 定義一個準(zhǔn)備作為線程任務(wù)的函數(shù)
def action(max):
? ? my_sum = 0
? ? for i in range(max):
? ? ? ? print(threading.current_thread().name + ' ?' + str(i))
? ? ? ? my_sum += i
? ? return my_sum

# 創(chuàng)建一個包含2條線程的線程池
pool = ThreadPoolExecutor(max_workers=2)
# 向線程池提交一個task, 50會作為action()函數(shù)的參數(shù)
future1 = pool.submit(action, 50)
# 向線程池再提交一個task, 100會作為action()函數(shù)的參數(shù)
future2 = pool.submit(action, 100)
def get_result(future):
?? ?print(future.result())
? ??
# 為future1添加線程完成的回調(diào)函數(shù)
future1.add_done_callback(get_result)
# 為future2添加線程完成的回調(diào)函數(shù)
future2.add_done_callback(get_result)
# 判斷future1代表的任務(wù)是否結(jié)束
print(future1.done())
time.sleep(3)
# 判斷future2代表的任務(wù)是否結(jié)束
print(future2.done())
# 查看future1代表的任務(wù)返回的結(jié)果
print(future1.result())
# 查看future2代表的任務(wù)返回的結(jié)果
print(future2.result())
# 關(guān)閉線程池
pool.shutdown() ?# 序可以使用 with 語句來管理線程池，這樣即可避免手動關(guān)閉線程池

最佳線程數(shù)目 = （（線程等待時間+線程CPU時間）/線程CPU時間 ）* CPU數(shù)目

也可以低于 CPU 核心數(shù)

3、 進(jìn)程池

使用線程池來執(zhí)行線程任務(wù)的步驟如下：

• 調(diào)用 ProcessPoolExecutor 類的構(gòu)造器創(chuàng)建一個線程池
• 定義一個普通函數(shù)作為進(jìn)程程任務(wù)
• 調(diào)用 ProcessPoolExecutor 對象的 submit() 方法來提交線程任務(wù)
• 當(dāng)不想提交任何任務(wù)時，調(diào)用 ProcessPoolExecutor 對象的 shutdown() 方法來關(guān)閉線程池

關(guān)于進(jìn)程的開啟代碼一定要放在 if __name__ == '__main__': 代碼之下，不能放到函數(shù)中或其他地方

開啟進(jìn)程的技巧：

from concurrent.futures import ProcessPoolExecutor

pool = ProcessPoolExecutor(max_workers=cpu_count()) ?# 根據(jù)cpu核心數(shù)開啟多少個進(jìn)程

開啟進(jìn)程的數(shù)量最好低于最大 CPU 核心數(shù)

到此這篇關(guān)于深入了解Python 中線程和進(jìn)程區(qū)別的文章就介紹到這了,更多相關(guān)Python 中線程和進(jìn)程內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論