概念

程序

• 程序是系統進行資源分配和排程的基本單位，是作業系統結構的基礎。簡單可以理解為正在執行的程式的例項，都會有一個pid作為標識。

多工

• 以作業系統為例，在上面可以同時執行QQ word 微信等多個程式，這就是多工
• 之前的單核CPU也可以執行多工，讓任務交替執行，因為CPU執行效率很快，所以對使用者是無感知的
• 真正的多工只能在多核CPU上完成，但如果任務數量大於CPU核心數量，作業系統也會自動把很多工輪流排程到每個核心上執行

建立程序

fork建立程序

fork建立子程序

• python中的os模組封裝了常見的系統呼叫，其中包含fork，通過fork可以輕鬆的建立子程序

import os 
import time 

pid = os.fork()
print('pid is %s' % pid)
if pid < 0:
    print('fork呼叫失敗')
elif pid == 0:
    # while True:
    #     print('this is 子程序')
    #     time.sleep(1)
    print('this is child process(%s) and parent is (%s)' % (os.getpid(), os.getppid()))
else:
    print('this is parent process(%s) and my child is (%s)'
 
 % (os.getpid(), pid))

print('父子程序都可以執行這裡的程式碼')

輸出結果為：

pid is 6661
this is parent process(6660) and my child is (6661)
父子程序都可以執行這裡的程式碼
pid is 0
this is child process(6661) and parent is (6660)
父子程序都可以執行這裡的程式碼

• 在unix和Linux中，fork這個系統函式比較特殊，呼叫一次，返回兩次。因為作業系統自動把當前程序複製一份，然後分別在父程序和子程序中返回
• 當程式執行到fork時，作業系統會建立一個新的程序(子程序),然後複製父程序的所有資訊到子程序中
• 父程序和子程序都會從fork函式中得到一個返回值，在子程序中該值為0，父程序中則是子程序的ID

多程序修改全域性變數

• 多程序中，每個程序的變數(包括全域性變數)都維繫一份，互不影響

import os
import time

rac = os.fork()
num = 1
if rac == 0:
    time.sleep(1)
    num += 1
    print('---parent num = %d' % num)
else:
    num += 1
    time.sleep(1)
    print('---child num = %d' % num)

print('---end num = %d' % num)

輸出結果為：

---child num = 2
---end num = 2
---parent num = 2
---end num = 2

多次fork

• 每次fork都會建立一個子程序，並且後續程式碼每個程序都會執行到。

import os
import time

rac = os.fork()
if rac == 0:
    print('---demo1---')
else:
    print('---demo2---')

rac2 = os.fork()
if rac2 == 0:
    print('---demo3---')
else:
    print('---demo4---')

print('---end---')

輸出結果為：

---demo2---
---demo4---
---end---
---demo3---
---end---
---demo1---
---demo4---
---end---
---demo3---
---end---

• 如上所示，demo1 demo2各執行一次，demo3 demo4各執行兩次，end累計執行四次

父子程序執行順序

• 執行順序沒有規律，取決於作業系統的排程演算法

multiprocessing建立程序

上面提到的fork，只能在linux unix上呼叫，在Windows上是無法執行的，但python作為一個跨平臺的語言，自然而然也提供了一個跨平臺多程序的模組–multiprocessing

通過process建立程序

• Process代表一個程序物件
• 建立子程序時，只需要在建構函式內傳入目標函式以及引數，呼叫start方法即可
• join()方法是讓主程序等待子程序結束後再執行後續程式碼

import os
from multiprocessing import Process
from time import sleep


def run_proc(name):
    print('子程序執行中，name = %s，pid = %d ...' % (name, os.getpid()))


"""
建立子程序時，只需要傳入一個執行函式和函式的引數，建立一個Process例項，用start方法啟動
join方法可以等待子程序結束後再往下執行，通常用於程序間的同步
"""
if __name__ == '__main__':
    print('父程序 %d.' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('子程序將要執行')
    p.start()
    p.join()
    print('子程序執行完畢')

輸出結果為：

父程序 6784.
子程序將要執行
子程序執行中，name = test，pid = 6785 ...
子程序執行完畢

通過Process子類建立程序

• 建立程序還可以使用類的方式，自定義一個類，繼承Process即可

import time
from multiprocessing import Process


class MyProcess(Process):
    def __init__(self, time, ):
        Process.__init__(self)
        self.time = time

    def run(self):
        # while True:
        print('---run---')
        # time.sleep(self.time)


if __name__ == '__main__':
    p = MyProcess(1)
    p.start()
    p.join()
    print('--main end--')

輸出結果為：

---run---
--main end--

程序池Pool

• 當建立的子程序數量不多時，可以通過Process來建立，如果要建立的子程序數量太多，那麼就不適用這種方式了，需要使用程序池來解決。
• 初始化一個Pool時，可以指定一個最大程序數，如果超過這個數，那麼請求就會等待，直到池中有程序結束，才會建立新的程序來執行

from multiprocessing import Pool
import time, os, random


def worker(msg):
    t_start = time.time()
    print('%s 開始執行，進城號為 %d' % (msg, os.getpid()))
    # random.random()隨機生成0-1的浮點數
    time.sleep(random.random() * 2)
    t_stop = time.time()
    print(msg, " 執行完畢，耗時%0.2f" % (t_stop - t_start))


# 建立一個程序池，裡面最多三個程序
pool = Pool(3)
for i in range(0, 10):
    # 向程序池中新增人物
    # 如果人物的數量超過程序池大小的話，會等待程序池中有空閒的時候，自動新增進去
    pool.apply_async(worker, args=(i,))

print('---start---')
# 關閉程序池 關閉後po不再接受新的請求
pool.close()
# 等待po中所有子程序執行完成，必須放在close語句之後
# 如果沒有join，會導致程序中的任務不會執行
pool.join()
print('---end---')

輸出結果為

---start---
0 開始執行，進城號為 6804
1 開始執行，進城號為 6805
2 開始執行，進城號為 6803
0  執行完畢，耗時1.23
3 開始執行，進城號為 6804
2  執行完畢，耗時1.29
4 開始執行，進城號為 6803
1  執行完畢，耗時1.85
5 開始執行，進城號為 6805
4  執行完畢，耗時0.82
6 開始執行，進城號為 6803
3  執行完畢，耗時0.90
7 開始執行，進城號為 6804
6  執行完畢，耗時0.54
8 開始執行，進城號為 6803
5  執行完畢，耗時0.89
9 開始執行，進城號為 6805
7  執行完畢，耗時0.70
9  執行完畢，耗時0.92
8  執行完畢，耗時1.77
---end---

• apply_async 非同步非阻塞
• apply 阻塞，上個程序結束才會執行下一個程序
• close 關閉pool 不再接收請求
• join 主程序阻塞，等待子程序執行結束，必須在closez以後

程序間通訊

程序中的Queue

• 程序間有時需要互相通訊，作業系統提供了多種方式進行通訊，Queue就是其中的一種
• queue就是一個訊息佇列，提供了get() put() qsize() empty() full()等方法對佇列進行操作
• 下面以queue為例，建立兩個程序，分別用來儲存資料

from multiprocessing import Queue, Process
import random, time


# queue，實現多程序之間的資料傳遞，其實就是個訊息佇列

def write(q):
    for value in ['A', 'B', 'C']:
        print("put %s to queue" % value)
        q.put(value)
        time.sleep(random.random())


def read(q):
    while True:
        if not q.empty():
            value = q.get()
            print("Get value is %s" % value)
            time.sleep(random.random())
        else:
            break


if '__main__' == __name__:
    q = Queue()
    qw = Process(target=write, args=(q,))
    qr = Process(target=read, args=(q,))
    qw.start()
    qw.join()
    qr.start()
    qr.join()
    print('---end---')
    print(random.randint(2, 4))
    print(random.random())

輸出結果為：

put A to queue
put B to queue
put C to queue
Get value is A
Get value is B
Get value is C
---end---
4
0.5260624608369131

程序池中的Queue

• 如果使用程序池pool建立程序的話，就需要使用Manager().Queue()

from multiprocessing import Manager, Process, Pool
import threading
import random, time,os


# queue，實現多程序之間的資料傳遞，其實就是個訊息佇列

def write(q):
    print('---write thread is %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print("put %s to queue" % value)
        q.put(value)


def read(q):
    print('---read thread is %s' % os.getpid())
    for i in range(q.qsize()):
        print("Get value is %s" % q.get(True))


if '__main__' == __name__:
    print('---main thread is %s' % os.getpid())
    q = Manager().Queue()
    po = Pool()
    po.apply(write, args=(q,))
    po.apply(read, args=(q,))
    po.close()
    po.join()
    print('---end---')

輸出結果為：

---main thread is 6907
---write thread is 6909
put A to queue
put B to queue
put C to queue
---read thread is 6910
Get value is A
Get value is B
Get value is C
---end---

總結

• 本文主要介紹了程序概念 多工 程序池 建立程序的幾種方式以及程序間的通訊等
• fork 不建議使用，不支援跨平臺
• process
  
  • process函式形式
  • 繼承Process，實現子類
• pool
  
  • 程序池，當要建立的程序數量比較多時，建議使用該方式
• queue
  
  • 程序間通訊
  • 使用process建立的程序，使用Queue()
  • 使用程序池建立的程序通訊，使用Manager().Queue()