多任務編程

cpu輪詢機制 ： cpu都在多個任務之間快速的切換執(zhí)行，切換速度在微秒級別，其實cpu同時只執(zhí)行一個任務，但是因為切換太快了，從應用層看好像所有任務同時在執(zhí)行。

并發(fā) ： 多個任務如果被分配給了一個cpu內(nèi)核，那么這多個任務之間就是并發(fā)關(guān)系，并發(fā)關(guān)系的多個任務之間并不是真正的"同時"。

并行 ： 多個任務如果被分配給了不同的cpu內(nèi)核，那么這多個任務之間執(zhí)行時就是并行關(guān)系，并行關(guān)系的多個任務時真正的“同時”執(zhí)行。

多任務編程：一個程序中編寫多個任務，在程序運行時讓這多個任務一起運行，而不是一個一個的順次執(zhí)行。比如微信視頻聊天，這時候在微信運行過程中涉及視頻、音頻、發(fā)消息。

實現(xiàn)多任務編程的方法 ： 多進程編程，多線程編程

多任務意義

• 提高了任務之間的配合，可以根據(jù)運行情況進行任務創(chuàng)建。

• 充分利用計算機資源，提高了任務的執(zhí)行效率。

  • 在任務中無阻塞時只有并行狀態(tài)才能提高效率

  • 在任務中有阻塞時并行并發(fā)都能提高效率

進程（Process）

進程概述：程序在計算機中的一次執(zhí)行過程。

• 程序是一個可執(zhí)行的文件，是靜態(tài)的占有磁盤。

• 進程是一個動態(tài)的過程描述，占有計算機運行資源，有一定的生命周期。

• 進程狀態(tài)

  三態(tài) 就緒態(tài) ： 進程具備執(zhí)行條件，等待系統(tǒng)調(diào)度分配cpu資源 運行態(tài) ： 進程占有cpu正在運行 等待態(tài) ： 進程阻塞等待，此時會讓出cpu五態(tài) (在三態(tài)基礎上增加新建和終止)新建 ： 創(chuàng)建一個進程，獲取資源的過程終止 ： 進程結(jié)束，釋放資源的過程進程命令

查看進程信息

ps -aux * USER ： 進程的創(chuàng)建者 * PID : 操作系統(tǒng)分配給進程的編號,大于0的整數(shù)，系統(tǒng)中每個進程的PID都不重復。PID也是重要的區(qū)分進程的標志。 * %CPU,%MEM : 占有的CPU和內(nèi)存 * STAT ： 進程狀態(tài)信息，S I 表示阻塞狀態(tài) ，R 表示就緒狀態(tài)或者運行狀態(tài) * START : 進程啟動時間 * COMMAND : 通過什么程序啟動的進程 pstree進程樹形結(jié)構(gòu)

• 父子進程：在Linux操作系統(tǒng)中，進程形成樹形關(guān)系，任務上一級進程是下一級的父進程，下一級進程是上一級的子進程。

多進程編程 multiprocessing

創(chuàng)建流程

1、將需要新進程執(zhí)行的事件封裝為函數(shù) 2、通過模塊的Process類創(chuàng)建進程對象，關(guān)聯(lián)函數(shù) 3、通過進程對象調(diào)用start啟動進程

主要類和函數(shù)使用

Process()功能 ： 創(chuàng)建進程對象參數(shù) ： target 綁定要執(zhí)行的目標函數(shù) args 元組，用于給target函數(shù)位置傳參kwargs 字典，給target函數(shù)鍵值傳參daemon bool值，讓子進程隨父進程退出p.start() 功能 ： 啟動進程注意: 啟動進程此時target綁定函數(shù)開始執(zhí)行，該函數(shù)作為新進程執(zhí)行內(nèi)容，此時進程真正被創(chuàng)建p.join([timeout])功能：阻塞等待子進程退出參數(shù)：最長等待時間

進程執(zhí)行現(xiàn)象理解 （難點）

新的進程是原有進程的子進程，子進程復制父進程全部內(nèi)存空間代碼段，一個進程可以創(chuàng)建多個子進程。 子進程只執(zhí)行指定的函數(shù)，其余內(nèi)容均是父進程執(zhí)行內(nèi)容，但是子進程也擁有其他父進程資源。 各個進程在執(zhí)行上互不影響，也沒有先后順序關(guān)系。 進程創(chuàng)建后，各個進程空間獨立，相互沒有影響。 multiprocessing 創(chuàng)建的子進程中無法使用標準輸入（即無法使用input）。 """ 進程創(chuàng)建示例 01 """ import multiprocessing as mp from time import sleepa = 1 # 全局變量# 進程目標函數(shù) def fun():print("開始運行一個進程")sleep(2) # 模擬事件執(zhí)行事件global aprint("a =",a) # Yesa = 10000print("進程執(zhí)行結(jié)束")# 實例化進程對象 windos系統(tǒng)需要放在main函數(shù)下 process = mp.Process(target=fun)# 啟動新進程 進程產(chǎn)生 執(zhí)行fun 與下面的第一個print搶占執(zhí)行沒關(guān)系 process.start() #如果使用函數(shù)fun()，執(zhí)行時間5秒print("我也做點事情") sleep(3) print("我也把事情做完了...")process.join() # 阻塞等待子進程結(jié)束,子進程不影響父進程 print("a:",a) # 1############## windos系統(tǒng)需要將多進程放在main函數(shù)下 if __name__ == '__main__':process = mp.Process(target=func)process.start() # 啟動進程 --》 執(zhí)行funcprint("哎呦，我也干點事吧")sleep(5)print("哈哈，我也干完了")

含有參數(shù)的進程函數(shù)練習

""" 進程創(chuàng)建示例02 ： 含有參數(shù)的進程函數(shù) """ from multiprocessing import Process from time import sleep#　含有參數(shù)的進程函數(shù) def worker(sec,name):for i in range(3):sleep(sec)print("I'm %s"%name)print("I'm working....")#　元組位置傳參 # p = Process(target=worker,args=(2,"Tom"))# 關(guān)鍵字傳參，或者同時使用 p = Process(target=worker,args = (2,),kwargs={"name":"Tom"},daemon=True) # 子進程伴隨父進程結(jié)束 p.start() sleep(3)

進程處理細節(jié)

進程相關(guān)函數(shù)

os.getpid()功能： 獲取一個進程的PID值返回值： 返回當前進程的PID os.getppid()功能： 獲取父進程的PID號返回值： 返回父進程PID sys.exit(info)功能：退出進程參數(shù)：字符串 表示退出時打印內(nèi)容 """ 創(chuàng)建多個子進程 """ from multiprocessing import Process from time import sleep import sys, os ## sys.exit("不能睡覺了") 結(jié)束進程并提示def th1():sleep(3)print("吃飯")print(os.getppid(), "--", os.getpid())def th2():# sys.exit("不能睡覺了") # 進程結(jié)束sleep(1)print("睡覺")print(os.getppid(), "--", os.getpid())def th3():sleep(2)print("打豆豆")print(os.getppid(), "--", os.getpid())# 循環(huán)創(chuàng)建子進程 jobs = [] # 存放每個進程對象def make_th():for th in [th1, th2, th3]:p = Process(target=th)jobs.append(p) # 存入jobsp.start() # 循環(huán)創(chuàng)建子進程 jobs = [] # 存放每個進程對象if __name__ == '__main__':# window是系統(tǒng)不能直接使用，需要放到main函數(shù)里使用或調(diào)用make_th()# 確保三件事都結(jié)束for i in jobs:i.join()print("三件事完成")

大文件拆分

""" 有一個大文件，將其拆分成上下兩個部分 （按照字節(jié)大小）， 要求兩個部分拆分要同步進行,不用合并 plus ： 假設文件很大不要一次read讀取全部 os.path.getsize() """ import os from multiprocessing import Process# 復制上半部分 def top(filename):fr = open(filename, 'rb')fw = open("top.jpeg", 'wb')n = os.path.getsize(filename) // 2while n >= 1024:fw.write(fr.read(1024))n -= 1024fw.write(fr.read(n))fr.close()fw.close()# 復制下半部分 def bot(filename):fr = open(filename, 'rb')fw = open("bot.jpeg", 'wb')n = os.path.getsize(filename) // 2fr.seek(n) # 文件偏移量到中間while True:data = fr.read(1024)if not data:breakfw.write(data)fr.close()fw.close()def main():jobs = []for func in [top, bot]:p = Process(target=func, args=("/home/下載/bizhi.jpeg",))jobs.append(p)p.start()[i.join() for i in jobs] # 和循環(huán)一個意思 酷print("文件拆分完成")if __name__ == '__main__':main()

孤兒進程和僵尸進程

• 孤兒進程： 父進程先于子進程退出時，子進程會成為孤兒進程，孤兒進程會被系統(tǒng)自動收養(yǎng)，成為孤兒進程新的父進程，并在孤兒進程退出時釋放其資源。

• 僵尸進程： 子進程先于父進程退出，父進程又沒有處理子進程的退出狀態(tài)，此時子進程就會成為僵尸進程。

  特點： 僵尸進程雖然結(jié)束，但是會存留部分進程資源在內(nèi)存中，大量的僵尸進程會浪費系統(tǒng)資源。Python模塊當中自動建立了僵尸處理機制，每次創(chuàng)建新進程都進行檢查，將之前產(chǎn)生的僵尸處理掉，而且父進程退出前，僵尸也會被自動處理。

  """ 僵尸進程 """ from multiprocessing import Process import os from time import sleepdef fun():print("子進程結(jié)束變?yōu)榻┦?#xff1a;", os.getpid())#while True: #使用這段循環(huán)，產(chǎn)生僵尸進程 # passwhile True:passsleep(3)p = Process(target=fun)p.start() # 創(chuàng)建進程前會自動檢測處理已有僵尸# p.join() # 處理僵尸

創(chuàng)建進程類

進程的基本創(chuàng)建方法將子進程執(zhí)行的內(nèi)容封裝為函數(shù)。如果我們更熱衷于面向?qū)ο蟮木幊趟枷?#xff0c;也可以使用類來封裝進程內(nèi)容。

創(chuàng)建步驟繼承Process類重寫`__init__`方法添加自己的屬性，使用super()加載父類屬性重寫run()方法使用方法實例化對象調(diào)用start自動執(zhí)行run方法

自定義進程類練習

""" 自定義進程類 --》 面向?qū)ο笏枷?""" from multiprocessing import Process from time import sleepclass MyProcess(Process):def __init__(self, value):self.value = valuesuper().__init__() # 加載調(diào)用父類方法# 父類提供的方法接口--》 我們重寫即可使用def run(self):self.func()def func(self):for i in range(self.value):sleep(2)print("自己進程的事情")if __name__ == '__main__':p = MyProcess(3)p.start() # 創(chuàng)建進程 -->執(zhí)行run方法作為進程內(nèi)容

判斷質(zhì)數(shù)、求和、計時

#判斷一個數(shù)是否為質(zhì)數(shù) 質(zhì)數(shù): 只能被1和其本身整除的整數(shù)且>1 def is_prime(num):if num <= 1:return Falsefor i in range(2,num // 2 + 1): #判斷到num的一半即可，eg：100 不能被50+1以上的數(shù)整除if num % i == 0:return Falsereturn Truedef sum_prime():prime = [] # 存放所有質(zhì)數(shù)for i in range(1,100001):if is_prime(i):prime.append(i)print(sum(prime)) # 求和begin = time.time() sum_prime() # 用時： 12.56395149230957 print("用時：",time.time() - begin)

練習

""" 1. 求100000以內(nèi)質(zhì)數(shù)之和，并且計算這個求和過程的時間 2. 將100000分成4份，創(chuàng)建4個進程，每個進程求其中一份的 質(zhì)數(shù)之和，統(tǒng)計4個進程執(zhí)行完的時間在無阻塞的任務執(zhí)行中，并不是創(chuàng)建的進程越多越好，而是受到cpu硬件的制約 """ import time from multiprocessing import Process# 求begin -- end 之間的質(zhì)數(shù)之和 class Prime(Process):@staticmethoddef is_prime(n):if n <= 1:return Falsefor i in range(2, n // 2 + 1):if n % i == 0:return Falsereturn Truedef __init__(self,begin,end):self.begin = begin # 起始數(shù)字self.end = end # 結(jié)尾數(shù)字super().__init__()def run(self):prime = [] # 存放所有質(zhì)數(shù)for i in range(self.begin,self.end):if Prime.is_prime(i):prime.append(i) # 存入列表print(sum(prime))if __name__ == '__main__':jobs = []b = time.time()for i in range(1,100001,10000): #利用步長分割,10進程 p = Prime(i,i + 10000)#for i in range(1,100001,25000): #利用步長分割 4進程 # p = Prime(i,i + 25000) jobs.append(p)p.start()[i.join() for i in jobs]print("用時:",time.time()-b)

進程間通信

進程間空間獨立，資源不共享，此時在需要進程間數(shù)據(jù)傳輸時就需要特定的手段進行數(shù)據(jù)通信。

常用進程間通信方法：消息隊列，套接字等。

消息隊列使用： 在內(nèi)存中開辟空間，建立隊列模型，進程通過隊列將消息存入，或者從隊列取出完成進程間通信。

實現(xiàn)方法

from multiprocessing import Queueq = Queue(maxsize=0)功能: 創(chuàng)建隊列對象參數(shù)：最多存放消息個數(shù)，列表、字典、字符串...返回值：隊列對象q.put(data)功能：向隊列存入消息,滿了阻塞參數(shù)：data 要存入的內(nèi)容q.get()功能：從隊列取出消息，空了阻塞，先進先出返回值： 返回獲取到的內(nèi)容q.full() 判斷隊列是否為滿 q.empty() 判斷隊列是否為空 q.qsize() 獲取隊列中消息個數(shù) q.close() 關(guān)閉隊列 進程間通信示例： from multiprocessing import Process,Queue# 創(chuàng)建消息隊列 q = Queue(5)# 子進程函數(shù) def handle():while True:cmd = q.get() # 取出指令if cmd == "1":print("\n完成指令1")elif cmd == "2":print("\n完成指令2")# 創(chuàng)建進程 p = Process(target=handle,daemon=True) p.start()while True:cmd = input("指令：")if not cmd:breakq.put(cmd) # 通過隊列給子進程

練習

""" 有一個目錄中有若干普通文件，將該目錄復制一份到當前程序所在位置 要求： 目標文件夾中每個文件復制都采用一個獨立的進程完成當所有文件復制完成之后，按復制完成順序打印所有文件名 思路提示：子進程負責拷貝文件父進程做文件是否重名判斷，接收用戶輸入指令創(chuàng)建文件夾 ： os.mkdir(dir) os.listdir() """from multiprocessing import Process, Queue import osold = "/home/FTP/" new = "./FTP/" q = Queue() # 消息隊列，用于傳遞文件名def copy():while True:filename = q.get() # 從消息隊列獲取名字if filename == '##':break # 文件已經(jīng)拷貝完成fr = open(old + filename, 'rb')fw = open(new + filename, 'wb')while True:data = fr.read(1024)if not data:breakfw.write(data)fr.close()fw.close()def select_file():new_files = os.listdir(new) # 當前目錄FTP下的文件for file in os.listdir(old):if file in new_files:print("已存在%s 1.替換 2.跳過" % file)cmd = input("請選擇:")if cmd == "1":q.put(file) # 選擇1 則也放入消息隊列else:q.put(file) # 要拷貝的文件名放入消息隊列q.put("##") # 子進程結(jié)束標志def main():p = Process(target=copy)p.start()select_file() # 文件篩選if __name__ == '__main__':main()

群聊聊天室框架搭建設計步驟

功能 ： 類似qq群功能 有人進入聊天室需要輸入姓名，姓名不能重復 有人進入聊天室時，其他人會收到通知：Lucy 進入了聊天室 一個人發(fā)消息，其他人會收到： Lucy ： 一起出去玩啊。 有人退出聊天室，則其他人也會收到通知 : Lucy 退出了聊天室 擴展功能：服務器可以向所有用戶發(fā)送公告: 管理員消息： 大家好，歡迎進入聊天室。 需求認知 ： C / S使用流程 ：開始——進入--聊天--退出--結(jié)束模塊劃分 ： 進入聊天室 聊天 退出函數(shù)技術(shù)點設計： 網(wǎng)絡 UDP（暫用，練習）存儲：姓名 地址 [(name,address)] {name:address}發(fā)送接收 ： 轉(zhuǎn)發(fā)收發(fā)互不影響-》分進程完成通信協(xié)議設計 （請求不止一種）請求類型 數(shù)據(jù)參數(shù)進入聊天室 LOGIN name聊天 CHAT content退出 EXIT name具體每個模塊邏輯設計--》編碼搭建框架 ： udp網(wǎng)絡循環(huán)模型進入聊天室客戶端 輸入名字發(fā)送給服務端接收結(jié)果--》 是否進程是 ： 功能結(jié)束否 ： 重新回到第一步服務端 接收名字判斷是否可以進入聊天室 （名字是否重復）發(fā)送結(jié)果是 ： 給其他人發(fā)通知，存儲用戶信息否 ： 功能結(jié)束聊天退出

網(wǎng)絡通信搭建

""" 慣例信息 姓名 ： name 郵箱 ：123456@qq.cn 時間 ： 2000-01-11 環(huán)境 ： Python3.6在線的群聊聊天室，鞏固網(wǎng)絡udp和進程知識 """ from socket import *# 服務地址 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT)# 建立存儲容器 {name:address} user = {}# 處理用戶進入 def do_login():passdef do_chat():passdef do_exit():pass# 入口函數(shù)，搭建網(wǎng)絡模型 def main():sock = socket(AF_INET, SOCK_DGRAM) # UDPsock.bind(ADDR)# 總體循環(huán)接收請求，分情況討論 總分while True:request, addr = sock.recvfrom(1024)print(request) #創(chuàng)建通信測試，OK后去掉if __name__ == '__main__':main() """ chat 客戶端 """ from socket import * from multiprocessing import Process# 服務器地址 ADDR = ("127.0.0.1", 8888)def do_login(sock):passdef do_chat():passdef do_exit():pass# 入口函數(shù) def main():sock = socket(AF_INET, SOCK_DGRAM) # UDPsock.sendto(b'test',ADDR) #創(chuàng)建通信測試，OK后去掉# 順次向下按照步驟執(zhí)行do_login(sock)do_chat()do_exit()if __name__ == '__main__':main()# ### 框架通信測試：先動服務端，再啟動客服端，看通信是否正常 服務端收到b'test'

login功能添加

"""server端""" from socket import *# 服務地址 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT)# 建立存儲容器 {name:address} user = {}# 處理用戶進入 def do_login(sock, name, addr):if name in user:sock.sendto(b"FAIL", addr)else:sock.sendto(b"OK", addr)msg = "歡迎 %s 進入聊天室" % namefor key, value in user.items():sock.sendto(msg.encode(), value)user[name] = addr # 增加用戶def do_chat():passdef do_exit():pass# 入口函數(shù)，搭建網(wǎng)絡模型 def main():sock = socket(AF_INET, SOCK_DGRAM) # UDPsock.bind(ADDR)# 總體接收請求，分情況討論 總分while True:request, addr = sock.recvfrom(1024)tmp = request.decode().split(' ') # 簡單的解析 tmp->[LOGIN,name]if tmp[0] == "LOGIN":do_login(sock, tmp[1], addr)elif tmp[0] == "CHAT":do_chat()elif tmp[0] == "EXIT":do_exit()if __name__ == '__main__':main() """ 客戶端 """ from socket import * from multiprocessing import Process# 服務器地址 ADDR = ("127.0.0.1",8888)def do_login(sock):while True:name = input("請輸入昵稱:")msg = "LOGIN " + name # 請求sock.sendto(msg.encode(),ADDR)result,addr = sock.recvfrom(1024)if result == b'OK':print("進入聊天室成功")breakelse:print("該昵稱已存在")def do_chat():passdef do_exit():pass# 入口函數(shù) def main():sock = socket(AF_INET,SOCK_DGRAM) # UDPsock.sendto(b'test', ADDR)# 順次向下按照步驟執(zhí)行do_login(sock)do_chat()do_exit()if __name__ == '__main__':main()

其他功能代碼

聊天 客戶端 創(chuàng)建一個子進程父進程負責循環(huán)發(fā)送子進程負責循環(huán)接收服務端 接收客戶端請求 簡單解析將內(nèi)容轉(zhuǎn)發(fā)給其他人退出 客戶端：　發(fā)送請求，結(jié)束服務端：　通知其他人　刪除用戶信息優(yōu)化完善 ################ 服務端參考代碼 ################### """ 慣例信息 姓名 ： name 郵箱 ：123456@qq.cn 時間 ： 2000-01-11 環(huán)境 ： Python3.6在線的群聊聊天室，鞏固網(wǎng)絡udp和進程知識 """ from socket import * from multiprocessing import Process# 服務器地址 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT)# 存儲用戶信息 {name:address} user = {}# 處理進入聊天室 def login(sock, name, address):if name in user or "管理" in name: #名字里不能帶管理sock.sendto(b"FAIL", address)else:sock.sendto(b"OK", address)# 告知其他人msg = "歡迎 %s 進入聊天室" % namefor key, value in user.items():sock.sendto(msg.encode(), value)user[name] = address # 存儲用戶# print(user) # 測試# 處理聊天 def chat(sock, name, content):msg = "%s : %s" % (name, content)for key, value in user.items():# 不是本人就發(fā)送if key != name:sock.sendto(msg.encode(), value)# 處理退出 def exit(sock, name):if name in user:del user[name] # 刪除該用戶# 通知其他用戶msg = "%s 退出聊天室" % namefor key, value in user.items():sock.sendto(msg.encode(), value)def handle(sock):# 不斷接收請求，分情況討論while True:request, addr = sock.recvfrom(1024)tmp = request.decode().split(" ", 2)# 分情況討論if tmp[0] == "LOGIN":# tmp ->[LOGIN,name]login(sock, tmp[1], addr)elif tmp[0] == "CHAT":# tmp ->[CHAT,name,content]chat(sock, tmp[1], tmp[2])elif tmp[0] == "EXIT":# tmp ->[EXIT,name]exit(sock, tmp[1])# 程序入口函數(shù) def main():# 創(chuàng)建udpsock = socket(AF_INET, SOCK_DGRAM)sock.bind(ADDR)# 接收請求，分類處理p = Process(target=handle, args=(sock,), daemon=True)p.start()while True:content = input("管理員消息:")if not content:breakmsg = "CHAT 管理員消息 " + content # 從父進程發(fā)送到子進程，不能遍歷user{}發(fā)送，父子進程隔離sock.sendto(msg.encode(), ADDR)if __name__ == '__main__':main()################## 客戶端參考代碼 ################## from socket import * from multiprocessing import Process import sys# 服務器地址 SERVER_ADDR = ("124.71.188.218", 8888)def login(sock):while True:name = input("請輸入昵稱:")# 組織請求msg = "LOGIN " + namesock.sendto(msg.encode(), SERVER_ADDR)result, addr = sock.recvfrom(1024)if result == b"OK":print("進入聊天室")return name """ 客戶端返回自己的name，給父進程使用發(fā)送msg """else:print("該昵稱已存在")# 子進程接收函數(shù) def recv_msg(sock):while True:data, addr = sock.recvfrom(1024 * 10)# 格式處理content = "\n" + data.decode() + "\n發(fā)言："print(content, end="")# 　父進程發(fā)送函數(shù) def send_msg(sock, name):while True:try:content = input("發(fā)言：")except KeyboardInterrupt:content = "exit"# 表示退出if content == 'exit':msg = "EXIT " + namesock.sendto(msg.encode(), SERVER_ADDR)sys.exit("您已退出聊天室")msg = "CHAT %s %s" % (name, content) sock.sendto(msg.encode(), SERVER_ADDR)def main():sock = socket(AF_INET, SOCK_DGRAM)sock.bind(("0.0.0.0",55224)) # 端口不要變，udp端口會釋放改變name = login(sock) # 請求進入聊天室 接收返回的name給父進程使用# 子進程負責接收p = Process(target=recv_msg, args=(sock,), daemon=True)p.start()send_msg(sock, name) # 父進程發(fā)送消息 """客戶端返回自己的name，給父進程使用發(fā)送msg """if __name__ == '__main__':main()

線程 (Thread)

線程概述

什么是線程線程被稱為輕量級的進程，也是多任務編程方式也可以利用計算機的多cpu資源線程可以理解為進程中再開辟的分支任務線程特征一個進程中可以包含多個線程線程也是一個運行行為，消耗計算機資源一個進程中的所有線程共享這個進程的資源多個線程之間的運行同樣互不影響各自運行線程的創(chuàng)建和銷毀消耗資源遠小于進程

多線程編程 threading

#創(chuàng)建線程對象 from threading import Thread t = Thread() 功能：創(chuàng)建線程對象 參數(shù)：target 綁定線程函數(shù)args 元組 給線程函數(shù)位置傳參kwargs 字典 給線程函數(shù)鍵值傳參daemon bool值，主線程推出時該分支線程也推出 #啟動線程t.start()#等待分支線程結(jié)束t.join([timeout]) 功能：阻塞等待分支線程退出 參數(shù)：最長等待時間 """線程示例""" import threading from time import sleep import osa = 1# 線程函數(shù) def music():global aprint("a =",a)a = 10000for i in range(3):sleep(2)print(os.getpid(),"播放:黃河大合唱")# 實例化線程對象 thread = threading.Thread(target=music)# 啟動線程 線程存在 thread.start() #分支線程6576 播放: 勇氣\n6576 播放: 勇氣\n6576 播放: 勇氣\na = 1for i in range(4):sleep(1)print(os.getpid(),"播放:葫蘆娃") #主線程：6576 播放: 葫蘆娃# 阻塞等待分支線程結(jié)束 thread.join() print("a:",a) #a=10000 """ 創(chuàng)建多個線程，線程參數(shù) """ from threading import Thread from time import sleep# 含有參數(shù)的線程函數(shù) def func(sec,name):print("含有參數(shù)的線程來嘍。")sleep(sec)print("%s線程執(zhí)行結(jié)束"%name)# 循環(huán)創(chuàng)建線程 for i in range(5):t = Thread(target=func,args=(2,),kwargs={"name":"T-%d"%i})# daemon = True) # 分支線程隨主線程退出，加上的話可能打印不出結(jié)束t.start() #5個線程互不影響，搶占執(zhí)行，結(jié)束順序不定 # daemon = True 分支線程隨主線程退出，加上的話，分支線程無法完成2秒等待的打印

創(chuàng)建線程類

創(chuàng)建步驟繼承Thread類重寫`__init__`方法添加自己的屬性，使用super()加載父類屬性重寫run()方法

使用方法

實例化對象；調(diào)用start自動執(zhí)行run方法

from threading import Thread from time import sleepclass MyThread(Thread):def __init__(self,song):self.song = songsuper().__init__() # 得到父類內(nèi)容# 線程要做的事情def run(self):for i in range(3):sleep(2)print("播放:",self.song)t = MyThread("涼涼") t.start() # 運行run 隨堂練習： 現(xiàn)在有500張票，存在一個列表中 ["T1",...."T500"]，10個窗口同時賣這500張票 W1-W10使用10個線程模擬這10個窗口，同時賣票，直到所有的票都賣出為止，每出一張票 需要0.1秒，打印表示即可print("W1----T250")from threading import Thread,Lock from time import sleeplock = Lock() # 創(chuàng)建鎖# 將票準備好 ticket = ["T%d" % x for x in range(1, 501)]# 線程函數(shù) w:表示窗口 def sell(w):while ticket:print("%s --- %s"%(w,ticket.pop(0)))sleep(0.1)# 10個線程 for i in range(1,11):t = Thread(target=sell,args=("W%d"%i,))t.start()

線程同步互斥

進程資源相互隔離，線程資源共享

線程通信方法： 線程間使用全局變量進行通信

共享資源爭奪

• 共享資源：多線程都可以操作的資源稱為共享資源。對共享資源的操作代碼段稱為臨界區(qū)。
• 影響 ： 對共享資源的無序操作可能會帶來數(shù)據(jù)的混亂，或者操作錯誤。此時往往需要同步互斥機制協(xié)調(diào)操作順序。

同步、互斥機制

• 同步 ： 同步是一種協(xié)作關(guān)系，為完成操作，線程間形成一種協(xié)調(diào)，按照必要的步驟有序執(zhí)行操作。

• 互斥 ： 互斥是一種制約關(guān)系，當一個進程或者線程占有資源時會進行加鎖處理，此時其他進程線程就無法操作該資源，直到解鎖后才能操作。

線程Event

from threading import Evente = Event() #創(chuàng)建線程event對象 初始設置狀態(tài)unset：阻塞，set非阻塞e.wait([timeout]) #阻塞等待e被set 終端界面返回False阻塞，True非阻塞e.set() #設置e，使wait結(jié)束阻塞，返回Truee.clear() #使e回到未被設置狀態(tài)e.is_set() #查看當前e是否被設置

Event使用示例

""" 線程同步互斥，注釋掉阻塞，if可能提前做，導致else結(jié)果 """ from threading import Thread,Evente = Event() #　　---------------創(chuàng)建event對象 msg = None # 線程間通信def 楊子榮():print("楊子榮前來拜山頭")global msgmsg = "天王蓋地虎"e.set() #　-------------------------解除初始狀態(tài)或者下面e.wait()阻塞t = Thread(target=楊子榮) t.start()print("說對口令才是自己人") e.wait() #　------------------阻塞等待 if msg == "天王蓋地虎":print("寶塔鎮(zhèn)河妖")print("確認過眼神，你是對的人") else:print("打死他...無情啊哥哥...")

線程鎖 Lock

from threading import Locklock = Lock() 創(chuàng)建鎖對象 lock.acquire() 上鎖 返回True 如果lock已經(jīng)上鎖再調(diào)用會阻塞 lock.release() 解鎖

Lock使用示例

from threading import Thread, Locklock = Lock() #　創(chuàng)建鎖 a = b = 0def value():while True:lock.acquire() # ---------------上鎖 #注釋掉所有的鎖，將產(chǎn)生打印if a != b:print("a = %d,b = %d" % (a, b))lock.release() # ---------------解鎖t = Thread(target=value) t.start()while True:lock.acquire() # ---------------上鎖a += 1b += 1lock.release() # ---------------解鎖 隨堂練習： 使用兩個分支線程，一個線程打印1-52 這52個數(shù)字，另一個線程打印A-Z 這26個字母。要求同時執(zhí)行兩個線程，打印順序為： 12A34B....5152Zfrom threading import Thread,Locklock1 = Lock() lock2 = Lock()def print_num():for i in range(1,53,2):lock1.acquire()print(i)print(i + 1)lock2.release()def print_chr():for i in range(65,91):lock2.acquire()print(chr(i))lock1.release()t1 = Thread(target=print_num) t2 = Thread(target=print_chr)lock2.acquire() # 先把打印字母的部分鎖住t1.start() t2.start()

死鎖

什么是死鎖

死鎖是指兩個或兩個以上的線程在執(zhí)行過程中，由于競爭資源或者由于彼此通信而造成的一種阻塞的現(xiàn)象，若無外力作用，它們都將無法推進下去。此時稱系統(tǒng)處于死鎖狀態(tài)或系統(tǒng)產(chǎn)生了死鎖。

死鎖產(chǎn)生條件

• 互斥條件：指線程使用了互斥方法，使用一個資源時其他線程無法使用。

• 請求和保持條件：指線程已經(jīng)保持至少一個資源，但又提出了新的資源請求，在獲取到新的資源前不會釋放自己保持的資源。

• 不剝奪條件：不會受到線程外部的干擾，如系統(tǒng)強制終止線程等。

• 環(huán)路等待條件：指在發(fā)生死鎖時，必然存在一個線程——資源的環(huán)形鏈，如 T0正在等待一個T1占用的資源；T1正在等待T2占用的資源，……，Tn正在等待已被T0占用的資源。

• 如何避免死鎖

  • 邏輯清晰，不要同時出現(xiàn)上述死鎖產(chǎn)生的四個條件
  • 通過測試工程師進行死鎖檢測

""" 死鎖現(xiàn)象演示 """ from time import sleep from threading import Thread,Lock# 賬戶類 class Account:def __init__(self,id,balance,lock):self._id = idself._balance = balanceself.lock = lock# 取錢def withdraw(self,amount):self._balance -= amount# 存錢def deposit(self,amount):self._balance += amount# 查看余額def getBalance(self):return self._balance# 轉(zhuǎn)賬函數(shù) def transfer(from_,to,amount): """from_.lock.acquire()from_.withdraw(amount) # from_錢減少sleep(0.1) # 網(wǎng)絡延遲to.lock.acquire() #開始產(chǎn)生鎖，另一個線程鎖了開始賺轉(zhuǎn)錢to.deposit(amount) # to錢增加to.lock.release()from_.lock.release() """from_.lock.acquire()from_.withdraw(amount) # from_錢減少from_.lock.release() # 使用后解鎖，不會其他線程產(chǎn)生死鎖sleep(0.1) # 網(wǎng)絡延遲to.lock.acquire()to.deposit(amount) # to錢增加to.lock.release()if __name__ == '__main__':tom = Account("Tom",5000,Lock())abby = Account("abby",8000,Lock())t1 = Thread(target=transfer,args=(tom,abby,600))t2 = Thread(target=transfer,args=(abby,tom,400))t1.start()t2.start()t1.join()t2.join()print("Tom:",tom.getBalance())print("Abby:",abby.getBalance())

GIL問題

什么是GIL問題 （全局解釋器鎖）

由于python解釋器設計中加入了解釋器鎖，導致python解釋器同一時刻只能解釋執(zhí)行一個線程，大大降低了線程的執(zhí)行效率。

導致后果
因為遇到阻塞時線程會主動讓出解釋器，去解釋其他線程。所以python多線程在執(zhí)行多阻塞任務時可以提升程序效率，其他情況并不能對效率有所提升。

關(guān)于GIL問題的處理盡量使用進程完成無阻塞的并發(fā)行為不使用c作為解釋器 （可以用Java C#）結(jié)論 GIL問題與Python語言本身并沒什么關(guān)系，屬于解釋器設計的歷史問題。在無阻塞狀態(tài)下，多線程程序程序執(zhí)行效率并不高，甚至還不如單線程效率。Python多線程只適用于執(zhí)行有阻塞延遲的任務情形。

線程效率對比進程實驗

import time from threading import Threadclass Prime(Thread):@staticmethoddef is_prime(num):if num <= 1:return Falsefor i in range(2, num // 2 + 1):if num % i == 0:return Falsereturn Truedef __init__(self,begin,end):self.begin = beginself.end = endsuper().__init__()def run(self):prime = [] # 存放所有質(zhì)數(shù)for i in range(self.begin,self.end):if Prime.is_prime(i):prime.append(i)print(sum(prime))def thread_10():jobs = []for i in range(1,100001,10000):t = Prime(i,i+10000)jobs.append(t)t.start()[i.join() for i in jobs]begin = time.time() # thread_4() # 用時: 12.594112157821655 thread_10() # 用時: 12.398741960525513 print("用時:",time.time() - begin)

進程線程的區(qū)別聯(lián)系

1. 兩者都是多任務編程方式，都能使用計算機多核資源 2. 進程的創(chuàng)建刪除消耗的計算機資源比線程多 3. 進程空間獨立，數(shù)據(jù)互不干擾，有專門通信方法；線程使用全局變量通信 4. 一個進程可以有多個分支線程，兩者有包含關(guān)系 5. 多個線程共享進程資源，在共享資源操作時往往需要同步互斥處理 6. Python線程存在GIL問題，但是進程沒有。

使用場景

任務場景：一個大型服務，往往包含多個獨立的任務模塊，每個任務模塊又有多個小獨立任務構(gòu)成，此時整個項目可能有多個進程，每個進程又有多個線程。

編程語言：Java,C#之類的編程語言在執(zhí)行多任務時一般都是用線程完成，因為線程資源消耗少；而Python由于GIL問題往往使用多進程。

總結(jié)

以上是生活随笔為你收集整理的pt14多任务编程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。