Python之進程

>什麼是進程

在一個正在運行的程序中，代碼和調度的資源稱爲進程，進程是操做系統分配資源的基本單元 以前有提到過，多任務不只線程能夠完成，進程也能夠

>進程的狀態

現實狀況裏，咱們電腦的任務數一般是大於CPU核數，這樣就會致使一些任務正在執行中，而一些任務在等待執行，這樣就會致使不一樣的狀態

就緒態：運行的條件已經知足，正在等待CPU執行

運行態：CPU正在執行

等待態：等待一些知足的條件，條件知足後進入就緒態

>進程的建立multiprocessing

multiprocessing模塊提供了一個Process類來表明一個進程對象

-使用主進程建立一個子進程

# -*- coding:utf-8 -*-
import multiprocessing
import time

def run_proc():
    """子進程要執行的代碼"""
    while True:
        print("這是子進程...1")
        time.sleep(2)

if __name__=='__main__':
    p = multiprocessing.Process(target=run_proc)
    p.start()
    while True:
        print("這是主進程...2")
        time.sleep(2)
複製代碼

執行結果

這是主進程...2
這是子進程...1
這是主進程...2
這是子進程...1
這是主進程...2
.
.
複製代碼

小結

• 進程的建立和執行和線程傳入參數的方式有些相似，都是調用start()方法啓動

>進程中的PID

# -*- coding:utf-8 -*-
from multiprocessing import Process
import os
import time

def run_proc():
    """子進程要執行的代碼"""
    print('子進程運行中，pid=%d...' % os.getpid())  # os.getpid獲取當前進程的進程號
    print('子進程將要結束...')

if __name__ == '__main__':
    print('父進程pid: %d' % os.getpid())  # os.getpid獲取當前進程的進程號
    p = Process(target=run_proc)
    p.start()
複製代碼

執行結果

父進程pid: 15128
子進程運行中，pid=4868...
子進程將要結束...
複製代碼

>Process類的屬性和方法

Process([group [, target [, name [, args [, kwargs]]]]])

• target：若是傳遞了函數的引用，能夠任務這個子進程就執行這裏的代碼
• args：給target指定的函數傳遞的參數，以元組的方式傳遞
• kwargs：給target指定的函數傳遞命名參數
• name：給進程設定一個名字，能夠不設定
• group：指定進程組，大多數狀況下用不到

Process建立的實例對象的經常使用方法

• start()：啓動子進程實例（建立子進程）
• is_alive()：判斷進程子進程是否還在活着
• join([timeout])：是否等待子進程執行結束，或等待多少秒
• terminate()：無論任務是否完成，當即終止子進程

Process建立的實例對象的經常使用屬性

• name：當前進程的別名，默認爲Process-N，N爲從1開始遞增的整數
• pid：當前進程的pid（進程號）

>給子進程指定的函數傳遞參數

# -*- coding:utf-8 -*-
from multiprocessing import Process
import os
from time import sleep


def run_proc(name, age, **kwargs):
    for i in range(10):
        print('子進程運行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
        print(kwargs)
        sleep(0.2)

if __name__=='__main__':
    p = Process(target=run_proc, args=('test',18), kwargs={"m":20})
    p.start()
    sleep(1)  # 1秒中以後，當即結束子進程
    p.terminate()
    p.join()
複製代碼

執行結果

子進程運行中，name= test,age=18 ,pid=14828...
{'m': 20}
子進程運行中，name= test,age=18 ,pid=14828...
{'m': 20}
子進程運行中，name= test,age=18 ,pid=14828...
{'m': 20}
子進程運行中，name= test,age=18 ,pid=14828...
{'m': 20}
子進程運行中，name= test,age=18 ,pid=14828...
{'m': 20}
複製代碼

>進程間不共享全局變量

import multiprocessing
import time
import os

nums = [11,22]

def sub_process1():
	print("its sub_process1,pid:%s;nums:%s" % (os.getpid(),nums))
	nums.append(33)
	time.sleep(1)
	print("its sub_process1,pid:%s;nums:%s" % (os.getpid(),nums))

def sub_process2():
	print("its sub_process2,pid:%s;nums:%s" % (os.getpid(),nums))

if __name__ == '__main__':
	p1 = multiprocessing.Process(target=sub_process1)
	p1.start()
	p1.join()

	p2 = multiprocessing.Process(target=sub_process2)
	p2.start()
複製代碼

執行結果

its sub_process1,pid:5028;nums:[11, 22]
its sub_process1,pid:5028;nums:[11, 22, 33]
its sub_process2,pid:6400;nums:[11, 22]
複製代碼

>進程、線程

進程是系統進行資源分配和調度的一個獨立單位

線程是進程的一個實體,是CPU調度和分配的基本單位,它是比進程更小的獨立運行的基本單位.線程基本不擁有系統資源,只擁有一些必不可少的資源(程序計數器,寄存器,棧等),線程可與同屬一個進程的其它線程共享進程所擁有的所有資源

兩者區別

一個程序至少有一個進程,一個進程至少有一個線程

線程的劃分尺度(資源)小於進程,使得線程的併發性高

進程在運行過程當中獨享內存,而線程是共享的,極大的提升了程序的運行效率

小結

線程的資源開銷小,可是資源的管理和保護不如進程

>進程間通訊Queue

Queue()實例化時,若是沒有指定最大可接收消息的最大參數或者爲負值,那麼默認沒有上限(直到內存的上限)

• Queue.qsize(): 返回當前隊列包含消息的數量
• Queue.empty(): 判斷隊列是否爲空,是爲True,否爲False
• Queue.full(): 判斷隊列是否滿了,是爲True,否爲False
• Queue.get([block[,timeout]]): 獲取一條隊列的消息,而後將其從隊列中刪除,black默認爲True
  • 若是block爲默認值,沒有設置timeout,消息隊列又爲空,那麼此時程序將會阻塞(停在讀取狀態),直到隊列中有值,從消息隊列中讀取到值.若是設置了timeout,等待timeout的時間就會報出(Queue.Empty)異常
  • 若是block爲False,消息隊列爲空,則會馬上拋出(Queue.Empty)異常
• Queue.get_nowait(): 至關於Queue.get(False)
• Queue.put(item,[block[,timeout]]): 將item寫入隊列,block默認爲True
  • 若是block爲默認值,沒有設置timeout,在寫入時沒有空間了,那麼程序會阻塞(中止在寫入狀態),直到有空間寫入爲止.若是設置了timeout,在寫入時沒有空間時,會等待timeout時間,就會拋出Queue.Full異常
  • 若是block爲False,在向消息隊列寫入時沒有空間了,直接拋出Queue.Full異常
• Queue.put_nowait(): 至關於Queue.put(block=False)

>Queue操做

from multiprocessing import Queue

q = Queue(3) #在實例化時傳入消息隊列的最大數量
q.put("test1")
q.put("test2")
print(q.full()) #這裏隊列未滿,返回False
q.put("test3")
print(q.full()) #這裏隊列已經滿了,返回True

try:
	q.put("test4", timeout=2) #在寫入時,若是隊列是滿的,就等待兩秒鐘,如還不能夠,拋出異常
except:
	print("當前消息隊列的數量爲%s" % q.qsize())

try:
	q.put("test4", block=False) # 在寫入時,若是隊列是滿的,直接拋出異常
except:
	print("當前消息隊列的數量爲%s" % q.qsize())

# 在寫入以前,能夠判斷下隊列是不是滿的
if not q.full():
	q.put_nowait("test4")

# 同理,在取數以前,判斷隊列是否爲空
if not q.empty():
	for i in range(q.qsize()):
		print(q.get_nowait())
複製代碼

執行結果

False
True
當前消息隊列的數量爲3
當前消息隊列的數量爲3
test1
test2
test3
複製代碼

>兩個進程分別讀寫Queue

from multiprocessing import Process,Queue
import time
import random

def write(q):
	nums = [11,22,33]
	for num in nums:
		if not q.full():
			print("write into num:%s" % num)
			q.put(num)
			time.sleep(random.random())
def read(q):
	while True:
		if not q.empty():
			num = q.get(True)
			print("read data num:%s" % num)
			time.sleep(random.random())
		else:
			break

if __name__ == '__main__':
	q = Queue()
	p1 = Process(target=write, args=(q,))
	p2 = Process(target=read, args=(q,))

	p1.start()
	p1.join()

	p2.start()
	p2.join()

	print("全部的數據都讀寫完畢了...")
複製代碼

執行結果

write into num:11
write into num:22
write into num:33
read data num:11
read data num:22
read data num:33
全部的數據都讀寫完畢了...
複製代碼

>multiprocess Pool(在linux環境能夠正常執行,windows有異常)

爲何要引用進程池,當咱們須要建立少許的進程時,能夠手動的去建立,但是須要的建立的進程數量多時,就能夠用到multiprocess中的Pool方法

咱們在初始化Pool時,能夠制定一個參數(最大的進程數),當有新的請求到Pool時,若是Pool池還沒滿,就會添加,若是滿了,就會等到Pool池中有進程結束,纔會用以前的進程來執行這個請求

multiprocess.Pool方法解析

• apply_async(func[,args[,kwargs]]): 使用非阻塞方式調用func(並行執行,阻塞的狀況是等待上一個進程結束才能執行下一個請求),args爲傳入的參數,kwargs爲關鍵字參數列表
• close(): 關閉Pool,使其再也不結束新的任務
• terminate(): 無論任務是否完成,都當即退出
• join(): 主進程阻塞,等待子進程,在close()和terminate()後執行

# -*- coding:utf-8 -*-
from multiprocessing import Pool
import os, time, random

def work(msg):
	t_start = time.time()
	print("執行開始,進程的ID號爲:%s" % os.getpid())
	time.sleep(random.random()*2)
	t_send = time.time()
	print("執行耗費時間爲:%s" % (t_start - t_send))

po = Pool(3)

for i in range(0, 10):
	po.apply_async(work, (i,))

print("programmer starting....")
po.close()
po.join()
print("programmer ending...")
複製代碼

執行結果

programmer starting....
執行開始,進程的ID號爲:24832
執行開始,進程的ID號爲:24831
執行開始,進程的ID號爲:24833
執行耗費時間爲:-0.21535086631774902
執行開始,進程的ID號爲:24832
執行耗費時間爲:-1.3048121929168701
執行開始,進程的ID號爲:24833
執行耗費時間爲:-1.4840171337127686
執行開始,進程的ID號爲:24831
執行耗費時間爲:-1.6602394580841064
執行開始,進程的ID號爲:24832
執行耗費時間爲:-0.48267197608947754
執行開始,進程的ID號爲:24831
執行耗費時間爲:-0.277604341506958
執行開始,進程的ID號爲:24831
執行耗費時間爲:-0.2472696304321289
執行開始,進程的ID號爲:24831
執行耗費時間爲:-1.3967657089233398
執行耗費時間爲:-0.2590181827545166
執行耗費時間爲:-1.4253907203674316
programmer ending...
複製代碼

>進程池中的進程通訊(Queue)

使用進程池Pool時,不能使用multiprocess.Queue,應該使用multiprocess.Manager()中的Queue 實戰演示

# -*- coding:utf-8 -*-
from multiprocessing import Manager,Pool
import os,time,random

def reader(q):
    print("reader啓動(%s),父進程爲(%s)" % (os.getpid(), os.getppid()))
    for i in range(q.qsize()):
        print("reader從Queue獲取到消息：%s" % q.get(True))

def writer(q):
    print("writer啓動(%s),父進程爲(%s)" % (os.getpid(), os.getppid()))
    for i in "itcast":
        q.put(i)

if __name__=="__main__":
    print("(%s) start" % os.getpid())
    q = Manager().Queue()
    po = Pool()
    po.apply_async(writer, (q,))

    time.sleep(1)  # 先讓上面的任務向Queue存入數據，而後再讓下面的任務開始從中取數據

    po.apply_async(reader, (q,))
    po.close()
    po.join()
    print("(%s) End" % os.getpid())
複製代碼

執行結果

(7740) start
writer啓動(7448),父進程爲(7740)
reader啓動(8096),父進程爲(7740)
reader從Queue獲取到消息：i
reader從Queue獲取到消息：t
reader從Queue獲取到消息：c
reader從Queue獲取到消息：a
reader從Queue獲取到消息：s
reader從Queue獲取到消息：t
(7740) End
複製代碼