一、 什么是進程 / 線程

1、 引論

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

2、 線程

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

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

3、 進程

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

4、 區(qū)別

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

5、 使用

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

二、 多線程使用

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

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

from threading import Thread

1、 常用方法

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

2、 常用參數

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

? ? # 生產者模型
? ? def submit_tasks(self, func, args=(), kwargs={}): ?# func為要執(zhí)行的任務，加入不定長參數使用（默認使用默認參數）
? ? ? ? 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})

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

守護模式：

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

3.2 直接調用法

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、 線程間數據的共享

現在我們程序代碼中，有多個線程， 并且在這個幾個線程中都會去 操作同一部分內容，那么如何實現這些數據的共享呢？

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

Lock 對象的acquire方法 是申請鎖

每個線程在操作共享數據對象之前，都應該申請獲取操作權，也就是調用該共享數據對象對應的鎖對象的acquire方法，如果線程A 執(zhí)行了acquire() 方法，別的線程B 已經申請到了這個鎖， 并且還沒有釋放，那么 線程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)

到在使用多線程時，如果數據出現和自己預期不符的問題，就可以考慮是否是共享的數據被調用覆蓋的問題

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

三、 多進程使用

1、 簡介

Python中的多進程是通過multiprocessing包來實現的，和多線程的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() ?# 讓進程等待

守護模式：

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

2.2 直接調用法

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、 進程之間的數據共享

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 函數提交給線程池/進程池，剩下的事情就由線程池/進程池來搞定

Exectuor 提供了如下常用方法：

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

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

Future 提供了如下方法：

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

2、 獲取 CPU 數量

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

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

3、 線程池

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

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

from concurrent.futures import ThreadPoolExecutor
import threading
import time
# 定義一個準備作為線程任務的函數
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()函數的參數
future1 = pool.submit(action, 50)
# 向線程池再提交一個task, 100會作為action()函數的參數
future2 = pool.submit(action, 100)
def get_result(future):
?? ?print(future.result())
? ??
# 為future1添加線程完成的回調函數
future1.add_done_callback(get_result)
# 為future2添加線程完成的回調函數
future2.add_done_callback(get_result)
# 判斷future1代表的任務是否結束
print(future1.done())
time.sleep(3)
# 判斷future2代表的任務是否結束
print(future2.done())
# 查看future1代表的任務返回的結果
print(future1.result())
# 查看future2代表的任務返回的結果
print(future2.result())
# 關閉線程池
pool.shutdown() ?# 序可以使用 with 語句來管理線程池，這樣即可避免手動關閉線程池

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

也可以低于 CPU 核心數

3、 進程池

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

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

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

開啟進程的技巧：

from concurrent.futures import ProcessPoolExecutor

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

開啟進程的數量最好低于最大 CPU 核心數

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

最新評論