1.執行緒概述
1.執行緒是實作多任務編程的一種方法,可以使用計算機多核資源,是計算機核心分配的最小單位,執行緒由代碼段,資料段,和TCB(執行緒控制塊)組成
2.執行緒又稱為輕量級行程,在創建和洗掉時消耗的計算機資源小,理論上創建和銷毀執行緒的消耗是創建和銷毀行程消耗的二十分之一
3.一個行程中的所有執行緒共享行程的空間資源(空間,全域變數,分配的記憶體等),行程中每個執行緒有自己的特有屬性,如指令集TID等
4.多執行緒程式的執行順序是不確定的,主執行緒會等待所有的子執行緒結束后才結束
5.計算機開啟的執行緒數量建議: CPU核數 * 5
2.threading模塊語法概述
import threading t = thread.Thread() # 創建執行緒并回傳執行緒物件 引數: target: 執行緒函式 args: 給執行緒函式的位置引數(型別為元組) kwargs: 給執行緒函式的字典傳參(型別為字典) name: 給執行緒取名字(默認為Thread-1) t.start(): 啟動執行緒 t.join(timeout): 回收執行緒 t.is_alive(): 查看執行緒狀態 t.name: 查看執行緒名稱 threading.currentThread(): 得到執行緒物件 t.setName(): 設定執行緒名稱 t.daemon = True: 守護執行緒,默認為False主執行緒執行完畢不會影響分支執行緒的執行,True則表示主執行緒執行完畢其它執行緒也會終止 設定方法: t.daemon = True 或者 t.setDaemon(True) t.isDaemon(True): 判斷daemon屬性是 True or False
3.執行緒屬性示例
from threading import Thread from threading import currentThread from time import sleep def func(sec): print("執行緒屬性測驗") sleep(sec) print("%s執行緒結束" % currentThread().getName()) def main(): thread = list() for i in range(3): t = Thread(name="t-" + str(i), target=func, args=(5,)) # 設定deamon屬性為True,此時主執行緒結束,子執行緒也會結束 # t.setDaemon(True) # 默認為False,主執行緒等待子執行緒結束 print("isDaemon", t.isDaemon()) # 也可以在定義后設定執行緒名稱 # t.setName = "t-" + str(i) t.start() thread.append(t) for i in thread: i.join(1) print("thread name:", i.name) print("alive:", i.is_alive()) print("主執行緒結束") if __name__ == '__main__': main() """執行結果 isDaemon False 執行緒屬性測驗 isDaemon False 執行緒屬性測驗 isDaemon False 執行緒屬性測驗 thread name: t-0 alive: True thread name: t-1 alive: True thread name: t-2 alive: True 主執行緒結束 t-0執行緒結束 t-2執行緒結束 t-1執行緒結束 """
4.執行緒定時器
from threading import Timer # 定時器 def func(): print('此生一入IT們,從此不愛任何人!') Timer(2.5, func).start() # Timer(time, func) # time: 睡眠的時間,以秒為單位 # func: 睡眠時間之后,需要執行的任務
5.驗證守護執行緒
from threading import Thread import time def func(): time.sleep(2) print(123) def func1(): time.sleep(1) print('abc') # 守護執行緒不是根據主執行緒的代碼執行結束而結束,而是根據主執行緒執行結束才結束 # 主執行緒會等待普通執行緒執行結束再結束,守護執行緒會等待主執行緒結束再結束,所以一般把不重要的事情設定為守護執行緒 # 守護行程是根據主行程的代碼執行完畢,守護行程就結束 if __name__ == '__main__': t = Thread(target=func) t.daemon = True t.start() t1 = Thread(target=func1) t1.start() print(456)
6.自定義執行緒函式
# 將函式作為實參傳遞給執行緒類,通過執行緒類的實體化物件啟動執行緒 import threading import time def sing(): """唱歌5秒鐘""" for i in range(5): print("---正在唱歌---%s" % i) time.sleep(1) def dance(): """跳舞10秒鐘""" for i in range(10): print("---正在跳舞---%s" % i) time.sleep(1) def main(): # 創建執行緒 t1 = threading.Thread(target=sing) t2 = threading.Thread(target=dance) # 啟動執行緒 t1.start() t2.start() for i in range(15): # 查看當前執行緒數量 print(threading.enumerate()) time.sleep(1) t1.join() # 等待回收執行緒 t2.join() # 等待回收執行緒 if __name__ == "__main__": main()
7.自定義執行緒類
1.繼承Thread類重寫 run 方法
import threading import time # threading.Thread類有一個run方法,用于定義執行緒的功能函式,可以在自己的執行緒類中覆寫該方法 # 而創建自己的執行緒實體后,通過Thread類的start方法啟動該執行緒 class MyThread(threading.Thread): def run(self): for i in range(3): time.sleep(1) msg = "I'm "+self.name+' @ '+str(i) #name屬性中保存的是當前執行緒的名字 print(msg) if __name__ == "__main__": t = MyThread() t.start()
2.繼承Thread類重寫 run 方法和 __init__ 方法
from threading import Thread from time import ctime from time import sleep class MyThread(Thread): def __init__(self, func, args, name="Tedu"): super().__init__() self.func = func self.args = args self.name = name def run(self): self.func(*self.args) @staticmethod # 宣告此方法是靜態方法 def plary(file, sec): for _ in range(sec): print("%s: %s" % (file, ctime())) sleep(sec) def main(): t = MyThread(MyThread.plary, ("echo", 3)) t.start() t.join() if __name__ == "__main__": main() """ echo: Sun Jul 19 23:13:55 2020 echo: Sun Jul 19 23:13:58 2020 echo: Sun Jul 19 23:14:01 2020 """
8.Python執行緒的全域解釋器鎖(GIL)
全域解釋器鎖-產生的原因
GIL是CPython解釋器存在的問題
Python支持多執行緒 同步互斥 加鎖 超級鎖 但在同一時刻解釋器只能解釋一個執行緒
大量的Python庫為了省事沿用了這種方法導致了Python多執行緒效率低下
全域解釋器鎖-用htop命令查看CPU占有率驗證
1.單執行緒死回圈
def main(): # 主執行緒死回圈 while True: pass if __name__ == "__main__": main()
2.多執行緒死回圈
import threading # 子執行緒死回圈 def test(): pass def main(): t1 = threading.Thread(target=test) t1.start() # 主執行緒死回圈 while True: pass if __name__ == "__main__": main()
3.多行程死回圈
import multiprocessing # 子行程死回圈 def test(): pass def main(): p1 = multiprocessing.Process(target=test) p1.start() # 主行程死回圈 while True: pass if __name__ == "__main__": main()
全域解釋器鎖-通過計算密集耗時和IO操作耗時驗證
import time import threading import multiprocessing # 計算密集 def calculation(num, x=1, y=1): for _ in range(num): x += 1 y += 1 # IO密集-讀寫 def io_write_read(num): # IO密集-寫 f = open("./test.txt", "w") for _ in range(num): f.write("hello world!\n") f.close() # IO密集-讀 f = open("./test.txt", "r") for _ in range(num): f.readline() f.close() # 單任務耗時 def single_task_time(num, func): t_start = time.time() for _ in range(10): func(num) t_end = time.time() - t_start print("單執行緒CPU執行1億次%s耗時: %s" % (func.__name__, t_end)) # 多執行緒耗時 def threading_time(num, func): t_start = time.time() counts = list() for _ in range(10): # t = threading.Thread(target=func, args=(num,), kwargs={"x": 1, "y": 1}) t = threading.Thread(target=func, args=(num,)) t.start() counts.append(t) for i in counts: i.join() t_end = time.time() - t_start print("10個執行緒CPU執行1億次%s耗時: %s" % (func.__name__, t_end)) # 多行程耗時 def multiprocessing_time(num, func): t_start = time.time() counts = list() for _ in range(10): p = multiprocessing.Process(target=func, args=(num,)) p.start() counts.append(p) for i in counts: i.join() t_end = time.time() - t_start print("10個行程CPU執行1億次%s耗時: %s" % (func.__name__, t_end)) def main(): # 單執行緒耗時 single_task_time(5000000, calculation) # 單執行緒CPU執行1億次calculation耗時: 5.152180910110474 single_task_time(5000000, io_write_read) # 單執行緒CPU執行1億次io_write_read耗時: 22.120678186416626 # 多執行緒耗時 threading_time(5000000, calculation) # 10個執行緒CPU執行1億次calculation耗時: 5.205517053604126 threading_time(5000000, io_write_read) # 10個執行緒CPU執行1億次io_write_read耗時: 22.218310356140137 # 多行程耗時 multiprocessing_time(5000000, calculation) # 10個行程CPU執行1億次calculation耗時: 1.7116732597351074 multiprocessing_time(5000000, io_write_read) # 10個行程CPU執行1億次io_write_read耗時: 6.813052177429199 if __name__ == "__main__": main() """執行結果 單執行緒CPU執行1億次calculation耗時: 5.335744142532349 單執行緒CPU執行1億次io_write_read耗時: 21.98353886604309 10個執行緒CPU執行1億次calculation耗時: 5.344666814804077 10個執行緒CPU執行1億次io_write_read耗時: 22.28637194633484 10個行程CPU執行1億次calculation耗時: 1.8069920539855957 10個行程CPU執行1億次io_write_read耗時: 7.141486167907715 """
全域解釋器鎖-解決方法
1.不使用多執行緒,使用多行程
2.不使用C C++做的解釋器,用 C# Java
3.Python多執行緒適合高用時的網路IO操作,不適用CPU密集型程式
4.使用其它語言實作多執行緒,Python去呼叫
GIL解決方案示例-C語言寫子執行緒執行的函式由Python呼叫
loop.c檔案代碼
#include<stdio.h> void DeadLoop() { while(1) { ; } } int main(int argc, char **argv) { printf("start DeadLoop\n"); DeadLoop(); return 0; }
命令列下編譯C語言寫的代碼
gcc loop.c # 編譯成功后當前目錄先會生成一個a.out的可執行檔案
gcc loop.c -shared -o libdead_loop.so # 執行命令把c語言檔案編譯成一個動態庫檔案
./a.out # 執行檔案
Python呼叫編譯后的C語言的動態庫檔案
import ctypes import threading def main(): # 加載動態庫 lib = ctypes.cdll.LoadLibrary("libdead_loop.so") # 創建一個子執行緒,讓子執行緒執行C語言寫的死回圈函式 t = threading.Thread(target=lib.DeadLoop) t.start() if __name__ == "__main__": main()
9.多執行緒-udp聊天器
import socket import threading def send_msg(udp_socket): """獲取鍵盤資料,并將其發送給對方""" while True: # 1. 從鍵盤輸入資料 msg = input("\n請輸入要發送的資料:") # 2. 輸入對方的ip地址 dest_ip = input("\n請輸入對方的ip地址:") # 3. 輸入對方的port dest_port = int(input("\n請輸入對方的port:")) # 4. 發送資料 udp_socket.sendto(msg.encode("utf-8"), (dest_ip, dest_port)) def recv_msg(udp_socket): """接收資料并顯示""" while True: # 1. 接收資料 recv_msg = udp_socket.recvfrom(1024) # 2. 解碼 recv_ip = recv_msg[1] recv_msg = recv_msg[0].decode("utf-8") # 3. 顯示接收到的資料 print(">>>%s:%s" % (str(recv_ip), recv_msg)) def main(): # 1. 創建套接字 udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 2. 系結本地資訊 udp_socket.bind(("", 7890)) # 3. 創建一個子執行緒用來接收資料 t = threading.Thread(target=recv_msg, args=(udp_socket,)) t.start() # 4. 讓主執行緒用來檢測鍵盤資料并且發送 send_msg(udp_socket) if __name__ == "__main__": main()
10.多執行緒-簡單tcp服務器
from socket import * import os import sys from threading import Thread def client_handler(c): try: print("子執行緒接收%s客戶端的請求" % str(c.getpeername())) while True: data = c.recv(1024) if not data: break print(data.decode("utf-8")) c.send(b"receive your message") except (KeyboardInterrupt, SystemError): raise except Exception as e: print(e) # 關閉客戶端套接字 c.close() def main(): # 創建套接字 s = socket() # 埠重用 s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # 系結本地資訊 s.bind(("", 7890)) # 監聽 s.listen(128) print("主執行緒%d等待客戶端的鏈接" % os.getpid()) while True: try: c, addr = s.accept() except KeyboardInterrupt: raise except Exception as e: print(e) continue t = Thread(target=client_handler, args=(c,)) t.setDaemon(True) # 主執行緒執行完畢其它執行緒也會終止 t.start() # 關閉監聽套接字 s.close() if __name__ == "__main__": main()
11.多執行緒-簡單tcp服務器的測驗客戶端
import socket def main(): # 創建資料流套接字 tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 連接服務器 server_ip = input("請輸入要連接的服務器的ip:") serve_port = int(input("請輸入要連接的服務器的port:")) server_addr = (server_ip, serve_port) tcp_client_socket.connect(server_addr) # 發送資料 send_data = https://www.cnblogs.com/tangxuecheng/p/input("請輸入要發生的資料:") tcp_client_socket.send(send_data.encode("utf-8")) # 接收服務器發送過來的資料 recv_data = https://www.cnblogs.com/tangxuecheng/p/tcp_client_socket.recv(1024) print("接收到的資料為:%s" % recv_data.decode("utf-8")) # 關閉套接字 tcp_client_socket.close() if __name__ == "__main__": main()
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/27472.html
標籤：Python