多任務(wù)編程

cpu輪詢(xún)機(jī)制 ： cpu都在多個(gè)任務(wù)之間快速的切換執(zhí)行，切換速度在微秒級(jí)別，其實(shí)cpu同時(shí)只執(zhí)行一個(gè)任務(wù)，但是因?yàn)榍袚Q太快了，從應(yīng)用層看好像所有任務(wù)同時(shí)在執(zhí)行。

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

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

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

實(shí)現(xiàn)多任務(wù)編程的方法 ： 多進(jìn)程編程，多線(xiàn)程編程

多任務(wù)意義

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

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

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

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

進(jìn)程（Process）

進(jìn)程概述：程序在計(jì)算機(jī)中的一次執(zhí)行過(guò)程。

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

• 進(jìn)程是一個(gè)動(dòng)態(tài)的過(guò)程描述，占有計(jì)算機(jī)運(yùn)行資源，有一定的生命周期。

• 進(jìn)程狀態(tài)

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

查看進(jìn)程信息

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

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

多進(jìn)程編程 multiprocessing

創(chuàng)建流程

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

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

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

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

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

含有參數(shù)的進(jìn)程函數(shù)練習(xí)

""" 進(jìn)程創(chuàng)建示例02 ： 含有參數(shù)的進(jìn)程函數(shù) """ from multiprocessing import Process from time import sleep#　含有參數(shù)的進(jìn)程函數(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)鍵字傳參，或者同時(shí)使用 p = Process(target=worker,args = (2,),kwargs={"name":"Tom"},daemon=True) # 子進(jìn)程伴隨父進(jìn)程結(jié)束 p.start() sleep(3)

進(jìn)程處理細(xì)節(jié)

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

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

大文件拆分

""" 有一個(gè)大文件，將其拆分成上下兩個(gè)部分 （按照字節(jié)大小）， 要求兩個(gè)部分拆分要同步進(jìn)行,不用合并 plus ： 假設(shè)文件很大不要一次read讀取全部 os.path.getsize() """ import os from multiprocessing import Process# 復(fù)制上半部分 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()# 復(fù)制下半部分 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)一個(gè)意思 酷print("文件拆分完成")if __name__ == '__main__':main()

孤兒進(jìn)程和僵尸進(jìn)程

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

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

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

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

創(chuàng)建進(jìn)程類(lèi)

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

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

自定義進(jìn)程類(lèi)練習(xí)

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

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

#判斷一個(gè)數(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() # 用時(shí)： 12.56395149230957 print("用時(shí)：",time.time() - begin)

練習(xí)

""" 1. 求100000以?xún)?nèi)質(zhì)數(shù)之和，并且計(jì)算這個(gè)求和過(guò)程的時(shí)間 2. 將100000分成4份，創(chuàng)建4個(gè)進(jìn)程，每個(gè)進(jìn)程求其中一份的 質(zhì)數(shù)之和，統(tǒng)計(jì)4個(gè)進(jìn)程執(zhí)行完的時(shí)間在無(wú)阻塞的任務(wù)執(zhí)行中，并不是創(chuàng)建的進(jìn)程越多越好，而是受到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): #利用步長(zhǎng)分割,10進(jìn)程 p = Prime(i,i + 10000)#for i in range(1,100001,25000): #利用步長(zhǎng)分割 4進(jìn)程 # p = Prime(i,i + 25000) jobs.append(p)p.start()[i.join() for i in jobs]print("用時(shí):",time.time()-b)

進(jìn)程間通信

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

常用進(jìn)程間通信方法：消息隊(duì)列，套接字等。

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

實(shí)現(xiàn)方法

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

練習(xí)

""" 有一個(gè)目錄中有若干普通文件，將該目錄復(fù)制一份到當(dāng)前程序所在位置 要求： 目標(biāo)文件夾中每個(gè)文件復(fù)制都采用一個(gè)獨(dú)立的進(jìn)程完成當(dāng)所有文件復(fù)制完成之后，按復(fù)制完成順序打印所有文件名 思路提示：子進(jìn)程負(fù)責(zé)拷貝文件父進(jìn)程做文件是否重名判斷，接收用戶(hù)輸入指令創(chuàng)建文件夾 ： os.mkdir(dir) os.listdir() """from multiprocessing import Process, Queue import osold = "/home/FTP/" new = "./FTP/" q = Queue() # 消息隊(duì)列，用于傳遞文件名def copy():while True:filename = q.get() # 從消息隊(duì)列獲取名字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) # 當(dāng)前目錄FTP下的文件for file in os.listdir(old):if file in new_files:print("已存在%s 1.替換 2.跳過(guò)" % file)cmd = input("請(qǐng)選擇:")if cmd == "1":q.put(file) # 選擇1 則也放入消息隊(duì)列else:q.put(file) # 要拷貝的文件名放入消息隊(duì)列q.put("##") # 子進(jìn)程結(jié)束標(biāo)志def main():p = Process(target=copy)p.start()select_file() # 文件篩選if __name__ == '__main__':main()

群聊聊天室框架搭建設(shè)計(jì)步驟

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

網(wǎng)絡(luò)通信搭建

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

login功能添加

"""server端""" from socket import *# 服務(wù)地址 HOST = "0.0.0.0" PORT = 8888 ADDR = (HOST, PORT)# 建立存儲(chǔ)容器 {name:address} user = {}# 處理用戶(hù)進(jìn)入 def do_login(sock, name, addr):if name in user:sock.sendto(b"FAIL", addr)else:sock.sendto(b"OK", addr)msg = "歡迎 %s 進(jìn)入聊天室" % namefor key, value in user.items():sock.sendto(msg.encode(), value)user[name] = addr # 增加用戶(hù)def do_chat():passdef do_exit():pass# 入口函數(shù)，搭建網(wǎng)絡(luò)模型 def main():sock = socket(AF_INET, SOCK_DGRAM) # UDPsock.bind(ADDR)# 總體接收請(qǐng)求，分情況討論 總分while True:request, addr = sock.recvfrom(1024)tmp = request.decode().split(' ') # 簡(jiǎn)單的解析 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() """ 客戶(hù)端 """ from socket import * from multiprocessing import Process# 服務(wù)器地址 ADDR = ("127.0.0.1",8888)def do_login(sock):while True:name = input("請(qǐng)輸入昵稱(chēng):")msg = "LOGIN " + name # 請(qǐng)求sock.sendto(msg.encode(),ADDR)result,addr = sock.recvfrom(1024)if result == b'OK':print("進(jìn)入聊天室成功")breakelse:print("該昵稱(chēng)已存在")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()

其他功能代碼

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

線(xiàn)程 (Thread)

線(xiàn)程概述

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

多線(xiàn)程編程 threading

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

創(chuàng)建線(xiàn)程類(lèi)

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

使用方法

實(shí)例化對(duì)象；調(diào)用start自動(dòng)執(zhí)行run方法

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

線(xiàn)程同步互斥

進(jìn)程資源相互隔離，線(xiàn)程資源共享

線(xiàn)程通信方法： 線(xiàn)程間使用全局變量進(jìn)行通信

共享資源爭(zhēng)奪

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

同步、互斥機(jī)制

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

• 互斥 ： 互斥是一種制約關(guān)系，當(dāng)一個(gè)進(jìn)程或者線(xiàn)程占有資源時(shí)會(huì)進(jìn)行加鎖處理，此時(shí)其他進(jìn)程線(xiàn)程就無(wú)法操作該資源，直到解鎖后才能操作。

線(xiàn)程Event

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

Event使用示例

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

線(xiàn)程鎖 Lock

from threading import Locklock = Lock() 創(chuàng)建鎖對(duì)象 lock.acquire() 上鎖 返回True 如果lock已經(jīng)上鎖再調(diào)用會(huì)阻塞 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() # ---------------解鎖 隨堂練習(xí)： 使用兩個(gè)分支線(xiàn)程，一個(gè)線(xiàn)程打印1-52 這52個(gè)數(shù)字，另一個(gè)線(xiàn)程打印A-Z 這26個(gè)字母。要求同時(shí)執(zhí)行兩個(gè)線(xiàn)程，打印順序?yàn)?#xff1a; 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()

死鎖

什么是死鎖

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

死鎖產(chǎn)生條件

• 互斥條件：指線(xiàn)程使用了互斥方法，使用一個(gè)資源時(shí)其他線(xiàn)程無(wú)法使用。

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

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

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

• 如何避免死鎖

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

""" 死鎖現(xiàn)象演示 """ from time import sleep from threading import Thread,Lock# 賬戶(hù)類(lèi) class Account:def __init__(self,id,balance,lock):self._id = idself._balance = balanceself.lock = lock# 取錢(qián)def withdraw(self,amount):self._balance -= amount# 存錢(qián)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_錢(qián)減少sleep(0.1) # 網(wǎng)絡(luò)延遲to.lock.acquire() #開(kāi)始產(chǎn)生鎖，另一個(gè)線(xiàn)程鎖了開(kāi)始賺轉(zhuǎn)錢(qián)to.deposit(amount) # to錢(qián)增加to.lock.release()from_.lock.release() """from_.lock.acquire()from_.withdraw(amount) # from_錢(qián)減少from_.lock.release() # 使用后解鎖，不會(huì)其他線(xiàn)程產(chǎn)生死鎖sleep(0.1) # 網(wǎng)絡(luò)延遲to.lock.acquire()to.deposit(amount) # to錢(qián)增加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問(wèn)題

什么是GIL問(wèn)題 （全局解釋器鎖）

由于python解釋器設(shè)計(jì)中加入了解釋器鎖，導(dǎo)致python解釋器同一時(shí)刻只能解釋執(zhí)行一個(gè)線(xiàn)程，大大降低了線(xiàn)程的執(zhí)行效率。

導(dǎo)致后果
因?yàn)橛龅阶枞麜r(shí)線(xiàn)程會(huì)主動(dòng)讓出解釋器，去解釋其他線(xiàn)程。所以python多線(xiàn)程在執(zhí)行多阻塞任務(wù)時(shí)可以提升程序效率，其他情況并不能對(duì)效率有所提升。

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

線(xiàn)程效率對(duì)比進(jìn)程實(shí)驗(yàn)

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() # 用時(shí): 12.594112157821655 thread_10() # 用時(shí): 12.398741960525513 print("用時(shí):",time.time() - begin)

進(jìn)程線(xiàn)程的區(qū)別聯(lián)系

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

使用場(chǎng)景

任務(wù)場(chǎng)景：一個(gè)大型服務(wù)，往往包含多個(gè)獨(dú)立的任務(wù)模塊，每個(gè)任務(wù)模塊又有多個(gè)小獨(dú)立任務(wù)構(gòu)成，此時(shí)整個(gè)項(xiàng)目可能有多個(gè)進(jìn)程，每個(gè)進(jìn)程又有多個(gè)線(xiàn)程。

編程語(yǔ)言：Java,C#之類(lèi)的編程語(yǔ)言在執(zhí)行多任務(wù)時(shí)一般都是用線(xiàn)程完成，因?yàn)榫€(xiàn)程資源消耗少；而Python由于GIL問(wèn)題往往使用多進(jìn)程。

總結(jié)

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

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