multiprocessing模塊

Process

進程（Process）是計算機中的程序關於某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操作系統結構的基礎。在早期面向進程設計的計算機結構中，進程是程序的基本執行實體；在當代面向線程設計的計算機結構中，進程是線程的容器。

一個進程至少包含一個線程。

Process 類用來描述一個進程對象。創建子進程的時候，只需要傳入一個執行函數和函數的參數即可完成 Process 示例的創建。

Process類通過創建子進程來是實現更多的占有cpu,提高效率

在pycharm里查看下Process的源碼:

class Process(object):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        self.name = ''
        self.daemon = False
        self.authkey = None
        self.exitcode = None
        self.ident = 0
        self.pid = 0
        self.sentinel = None

    def run(self):
        pass

    def start(self):
        pass

    def terminate(self):
        pass

    def join(self, timeout=None):
        pass

    def is_alive(self):
        return False

 

除了__init__()之外,包含的方法有run(),start(),terminate(),join(),is_alive()五個方法.

run()方法:

在研究run()方法之前我們先看下面一段代碼:

from multiprocessing import Process
import os


def func():

    print(os.getpid(), os.getppid())


if __name__ == '__main__':

    print(os.getpid(), os.getppid())

    p = Process(target=func)

    p.start()

運行這段代碼的結果是:

6864 6100
5612 6864

py文件中的子進程在調用Process類對象后由子進程變成了父進程,也就是說新開了一個子進程去執行func()函數.

如果是使用面向對象的思想解決進程,自定義的子類要繼承Process類,並且在子類中重構run()方法,切必須重構run()方法.

重構的run()方法就相當於def func()函數一樣,切套通過start()方法調用run()方法來執行run()方法,看下面一段代碼:

from multiprocessing import Process
import os


class MyProcess(Process):

    def run(self):

        print(os.getpid(), os.getppid())


if __name__ == '__main__':

    print(os.getpid(), os.getppid())

    p = MyProcess()

    p.start()

 

這段代碼執行的結果:

6184 6100
5320 6184

說明我們重構的run()方法是正確的,和面向過程的實現一樣

不過此處要注意的是:在windows操作系統下,一定要加上if __name__ == '__main__':這段代碼,因為windows的操作機制是在是在開辟新的進程是會重新加載一編前邊的額引入模塊和變量.相當於把程序重新讀了一遍,這樣會陷入開辟子進程的死循環中,且會拋出RuntimeError的異常:

RuntimeError: 
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

而linux操作系統是直接把內存中的變量直接拷貝,不在去執行一遍

start()方法:

start()方法不是運行一個程序,而是調用操作系統的命令,告訴操作系統要創建子進程

在Process類內部start()方法是調用run()方法,來實現開啟一個子進程,且必須通過start()方法來實現開啟子進程

 join()方法:

join()方法會阻塞主程序一直到目標進程/線程執行完畢才執行編程非阻塞:

from os import getpid
from multiprocessing import Process


def func(i):
    print('Process %s will be executed ' % i)
    print('Process ID is %s' % getpid())
    print('Process %s has been executed ' % i)


if __name__ == '__main__':
    process_list = []
    for i in range(10):
        p = Process(target=func, args=(i,))
        p.start()
        process_list.append(p)  # 操作系統執行子進程並不是按照這里遍歷i的順序來執行的,而是按照他內部的算法實現的

    for p in process_list:
        p.join()    # 遍歷process_list列表,檢查p進程是否執行結束

    print('The main process has been executed')

打印結果是:

Process 0 will be executed 
Process ID is 11728
Process 0 has been executed 
Process 4 will be executed 
Process ID is 11632
Process 4 has been executed 
Process 1 will be executed 
Process ID is 7568
Process 1 has been executed 
Process 3 will be executed 
Process ID is 10548
Process 3 has been executed 
Process 2 will be executed 
Process ID is 10600
Process 2 has been executed 
Process 9 will be executed 
Process ID is 11940
Process 9 has been executed 
Process 7 will be executed 
Process ID is 11980
Process 7 has been executed 
Process 6 will be executed 
Process ID is 11772
Process 6 has been executed 
Process 5 will be executed 
Process ID is 11644
Process 5 has been executed 
Process 8 will be executed 
Process ID is 11600
Process 8 has been executed 
The main process has been executed

terminate()方法:

terminate()方法是用來終止一個子進程的.即回收該子進程的資源.

is_alive()方法:

is_alive()方法是用來判斷一個進程是否還在運行:

如果運行:return True

如果沒有運行:return False

下面這段代碼整體來掩飾一遍:

from os import getpid
from time import sleep
from multiprocessing import Process


class MyProcess(Process):

    def __init__(self, args):
        super(MyProcess, self).__init__()
        self.args = args

    def run(self):
        print('Process ID is %s' % getpid())
        print('Process %s has been executed ' % self.args)

    def start(self):
        print('Process %s will be executed ' % self.args)
        super().start()

    def terminate(self):
        super().terminate()

    def join(self, timeout=None):
        super().join()

    def is_alive(self):
        try:
            result = super().is_alive()
            return result
        except:
            print("Something was wrong!")


if __name__ == '__main__':
    process_list = []
    for i in range(10):
        p = MyProcess(i)
        p.start()
        print(p.is_alive())
        process_list.append(p)
    for p in process_list:
        print(p.is_alive())
        print('There will terminated process %s' % p)
        p.terminate()
        sleep(0.2)      # 這里不延時0.2而直接打印的話,下邊的is_alive會顯示True
        print(p.is_alive())
        p.join()
        print(p.is_alive())

Process的一些常用方法:

daemon()方法:

daemon()方法是守護進程守護進程會在主進程的代碼執行完畢之后直接結束,無論守護進程是否執行完畢,使用daemon()方法注意:

• 守護進程的屬性,默認是False,如果設置成True,就表示設置這個子進程為一個守護進程
• 設置守護進程的操作應該在開啟子進程之前

守護進程的主要應用:

• 報活 主進程還活着

from time import sleep
from multiprocessing import Process


def func1():
    print('begin')
    sleep(3)
    print('wahaha')


def func2():
    print('in func2')
    sleep(4)
    print('The func2 has been executed')


if __name__ == '__main__':
    Process(target=func1).start()
    p = Process(target=func2)
    p.daemon = True
    p.start()
    sleep(1)
    print('In main process')

打印會發現func2中的最后一句沒有被打印出來

Lock類:

 現階段了解的lock主要適用於共享數據時控制數據的安全,這個鎖有且只有一把鑰匙,如果前邊的進程拿到了后邊的就要等着那倒鑰匙才能進去

lock的類中有如下兩個方法:

class Lock(object):
    def acquire(self, blocking=True, timeout=-1):
        pass

    def release(self):
        pass

acquire()方法：

相當於拿鑰匙開鎖,

release()方法：

相當於還鑰匙

import json
from multiprocessing import Process, Lock


class MyProcess(Process):

    def __init__(self, lock):
        self.lock = lock
        super(MyProcess, self).__init__()

    def run(self):
        ticket_dict = self.read()
        print("The rest of the ticket>>>%s" % ticket_dict['count'])
        if ticket_dict['count'] >= 1:
            print("Out of ticket")
            self.lock.acquire()
            self.buy()
            self.lock.release()
        else:
            print('Failure to buy tickets')

    def read(self):
        with open('db', 'r') as filehandler:
            content = json.load(filehandler)
        return content

    def buy(self):
        conrent = self.read()
        if conrent['count'] >= 1:
            print('Buy ticket successful')
            conrent['count'] -= 1
            with open('db', 'w') as fh:
                json.dump(conrent, fh)
        else:
            print('More votes short')


if __name__ == '__main__':
    lock = Lock()
    for i in range(11):
        p = MyProcess(lock)
        p.start()

Semaphore()類：

semaphore()類的實質就是鎖+計數器，一把鑰匙或者多把鑰匙對應一個鎖，用法和Lock()基本相同，唯一的區別是如果是多把鑰匙的話要設置鑰匙數，默認是1。下面是源碼：

class Semaphore(object):
    def __init__(self, value=1):
        pass

    def acquire(self, blocking=True, timeout=None):
        pass

    def release(self):
        pass

semaphore()類包含了三個方法：

__init__(self, value=1)方法：

value=1：默認是1，這里的傳的值是設置鑰匙的個數

acquire()方法：

同Lock()類的acquire()

release()方法：

同Lock()類的release()

semaphore()類對應的時間是商場里的唱吧：假設一個唱吧里只能同時有四個人，外邊公有20人要准備玩，只能排隊等：

from time import sleep
from random import randint
from multiprocessing import Semaphore, Process


def ktv(s, i):
    s.acquire()
    print('%s 進入了ktv' % i)
    sleep(randint(1, 5))
    print('%s 離開了ktv' % i)
    s.release()


if __name__ == '__main__':
    s = Semaphore(4)
    for i in range(20):
        p = Process(target=ktv, args=(s, i))
        p.start()

打印會發現離開一個進一個

Event()類:

 event()類主要是用於進程間的通信,通過其他進程間的傳遞的信號去控制其他進程,下面是event()類的源碼:

class Event(object):
    def is_set(self):
        return False

    def set(self):
        pass

    def clear(self):
        pass

    def wait(self, timeout=None):
        pass

Enent提供了四個方法:

is_set(self)方法:

is_set()方法是用來查看標志的

在事件的創建之初 默認是False

set(self)方法:

將標志設置為True

clear(self)方法:

將標志設置為False

wait(self, timeout=None)方法:

等待：阻塞、如果這個標志是False 那么就阻塞。非阻塞，如果這個標志是True 那么就非阻塞。

timeout：如果是None，將會永遠阻塞。如果設置了時間：在設置的時間內，如果標志變成True，則直接pass，如果沒有，則繼續阻塞

Event()典型事件就是交通燈控制車輛是否通過，通過函數和面向對象模擬練習下這個事件：

from time import sleep
from random import randrange,choice
from multiprocessing import Process, Event


class MyObj(object):

    def traffic_light(self, e):
        """
        默認紅燈開始：is_set = False，
        sleep（2）是亮燈時間，然后切換另一個燈，然后設置is_set的狀態，
        紅燈亮：普通車等待，救護車等0.5秒，如果0.5秒后沒有變燈，則闖燈
        綠燈亮：所有車輛都通過
        :param e: Event類的實例化對象
        :return:
        """

        print('\033[1;31m The red light is on \033[0m')
        while 1:
            sleep(2)
            if e.is_set():
                print('\033[1;31m The red light is on \033[0m')
                e.clear()
            else:
                print('\033[1;32m The green light is on \033[0m')
                e.set()

    def car(self, e, id):
        if not e.is_set():
            print('%s car wait' % id)
            e.wait()
        print('%s car pass' % id)

    def ambulance(self, e, id):
        if e.is_set():
            e.wait(timeout=0.5)
        print('%s ambulance pass' % id)


if __name__ == '__main__':
    myobj = MyObj()
    e = Event()
    p = Process(target=myobj.traffic_light, args=(e,))
    p.start()
    car_list = [myobj.ambulance, myobj.car]
    for i in range(20):
        p = Process(target=choice(car_list), args=(e, i))
        p.start()
        sleep(randrange(0, 3, 2))

函數模擬：

from time import sleep
from random import randrange, choice
from multiprocessing import Event, Process


def traffic_light(e):
    print('\033[1;31m The red light is on\033[0m')
    while 1:
        sleep(2)
        if e.is_set():
            print('\033[1;31m The red light is on\033[0m')
            e.clear()
        else:
            print('\033[1;32m The green light is on\033[0m')
            e.set()


def car(id, e):
    if not e.is_set():
        print('%s car wait' % id)
        e.wait()
    print('%s car pass' % id)


def ambulance(id, e):
    if not e.is_set():
        e.wait(timeout=0.5)
    print('%s ambulance pass' % id)


if __name__ == '__main__':
    e = Event()
    p = Process(target=traffic_light, args=(e,))
    p.start()
    car_list = [car, ambulance]
    for i in range(20):
        p = Process(target=choice(car_list), args=(i, e))
        p.start()
        sleep(randrange(0, 3, 2))