一、 什么是進程 / 線程

1、 引論

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

2、 線程

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

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

3、 進程

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

4、 區(qū)別

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

5、 使用

在Python中，通過兩個標準庫 thread 和 Threading 提供對線程的支持, threading 對 thread 進行了封裝。 threading 模塊中提供了 Thread , Lock , RLOCK , Condition 等組件

二、 多線程使用

在Python中線程和進程的使用就是通過 Thread 這個類。這個類在我們的 thread 和 threading 模塊中。我們一般通過 threading 導入

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

from threading import Thread

1、 常用方法

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

2、 常用參數(shù)

3、 多線程的應用

3.1 重寫線程法

import time, queue, threading


class MyThread(threading.Thread):

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

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

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

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

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

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

守護模式：

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

3.2 直接調(diào)用法

def f2(i):
? ? time.sleep(2)
? ? print("任務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)容，那么如何實現(xiàn)這些數(shù)據(jù)的共享呢？

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

Lock 對象的acquire方法 是申請鎖

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

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

如：

import threading

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

# 定義兩個線程要用做的任務
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)和自己預期不符的問題，就可以考慮是否是共享的數(shù)據(jù)被調(diào)用覆蓋的問題

使用 threading 庫里面的鎖對象 Lock 去保護

三、 多進程使用

1、 簡介

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

2、 應用

2.1 重寫進程法

import time
from multiprocessing import Process


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

? ? 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() ?# 讓進程等待

守護模式：

主進程在其代碼結(jié)束后就已經(jīng)算運行完畢了（守護進程在此時就被回收）,然后主進程會一直等非守護的子進程都運行完畢后回收子進程的資源(否則會產(chǎ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() ?# 讓進程等待

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

3.1 Lock 方法

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

3.2 Manager 方法

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

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個進程，不同的進程對d和l中的不同元素進行操作
? ? ? ? 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)建進程池

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

Exectuor 提供了如下常用方法：

• submit(fn, *args, * kwargs)：將 fn 函數(shù)提交給線程池。 args 代表傳給 fn 函數(shù)的參數(shù)，*kwargs 代表以關鍵字參數(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)：關閉線程池

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

Future 提供了如下方法：

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

2、 獲取 CPU 數(shù)量

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

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

3、 線程池

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

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

from concurrent.futures import ThreadPoolExecutor
import threading
import time
# 定義一個準備作為線程任務的函數(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代表的任務是否結(jié)束
print(future1.done())
time.sleep(3)
# 判斷future2代表的任務是否結(jié)束
print(future2.done())
# 查看future1代表的任務返回的結(jié)果
print(future1.result())
# 查看future2代表的任務返回的結(jié)果
print(future2.result())
# 關閉線程池
pool.shutdown() ?# 序可以使用 with 語句來管理線程池，這樣即可避免手動關閉線程池

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

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

3、 進程池

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

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

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

開啟進程的技巧：

from concurrent.futures import ProcessPoolExecutor

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

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

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

最新評論