multiprocessing 模塊

multiprocessing模塊

進程對象

• 建立
  • p = Process(target=foo, args=(param,))
• 屬性
  • p.daemon: True爲守護進程, 守護進程內沒法再開啓子進程,不然拋出異常：AssertionError: daemonic processes are not allowed to have children, 父進程結束則本身也馬上結束; False則爲非守護進程, 自身進程運行與父進程是否結束無關; p.daemon = True | False 必須在p.start()以前調用
  • p.name: 進程名
  • p.pid: 進程pid; 若是當前進程爲父進程, 則p.pid與os.getpid()的結果同樣
  • p.exitcode: 爲None表示進程正在運行, 爲-n表示因爲某一個信號結束了
  • p.start(): 啓動一個進程, 內部會調用p.run()方法
  • p.join(): 調用p.join()語句的進程須要等待p進程結束才能繼續執行, p.join()中會調用wait函數
  • p.terminate(): 強制終止p進程, 可是不會當即終止, 因此若是在p.terminate()後緊接着是p.is_alive()則返回True, 可是若是緊接着會後面再來一個p.is_alive()就會返回False了, 第一個p.is_alive()會催促p結束
  • p.is_alive(): p進程是否還在運行

• 殭屍進程與進程

  • 殭屍進程
    • 父進程還在執行, 可是子進程結束了, 父進程沒有調用wait或者waitpid函數回收子進程的資源致使子進程死亡了可是仍然佔用着進程資源
    • 對系統有害, 會形成資源浪費
      • 解決
        1. 若是已經產生了殭屍進程: 殺掉父進程, 讓子進程成爲孤兒進程從而交給init進程管理便可
        2. 預防殭屍進程: p爲進程對象, 父進程p調用join()方法
  • 孤兒進程
    • 父進程結束, 可是子進程還在執行
    • 孤兒進程對系統無害, 孤兒進程會被init進程管理
• 注意: Windows與類Unix系統建立子進程的方式不一樣
  • 類Unix
    • 類Unix系統採用fork()系統調用函數建立子進程, 字如其名, 子進程就是父進程的一個副本, 拷貝的過程當中cs:ip的指針指向指令的哪一個位置也是一致的(這多虧了虛擬內存), 也就是說, 複製出來的子進程不是從頭開始執行的, 是從父進程調用fork()函數語句的下一條指令開始執行的, 隨着技術的發展, fork()如今採用的是CoW實現
  • Windows

    • Windows上建立子進程的函數爲CreateProcess(), 也是字如其名, 是建立一個進程而不是複製一個進程。CreateProcess()函數的API是

      BOOL CreateProcessA(
              LPCSTR                lpApplicationName, // 進程要執行的.exe文件名
              LPSTR                 lpCommandLine, // 執行的.exe的命令行參數
              LPSECURITY_ATTRIBUTES lpProcessAttributes,
              LPSECURITY_ATTRIBUTES lpThreadAttributes,
              BOOL                  bInheritHandles,
              DWORD                 dwCreationFlags,
              LPVOID                lpEnvironment,
              LPCSTR                lpCurrentDirectory,
              LPSTARTUPINFOA        lpStartupInfo,
              LPPROCESS_INFORMATION lpProcessInformation
          );

    • 咱們主要看API的第1和第2個參數, 很明顯, CreateProcess()API能夠建立一個與當前父進程徹底不一樣的子進程, 由於它接受一個.exe文件的路徑, 該路徑能夠是任何一個.exe文件, 將該.exe文件加載到內存中CPU從頭開始執行代碼, 若是要實現與類Unix中fork()函數相似的功能, CreateProcess()的第一個參數應該爲父進程的.exe文件的位置, 這樣就建立出來一個與父進程同樣的子進程了, 可是剛纔說了是相似, 確定有不一樣, 類Unix中fork出來的子進程的入口是父進程fork語句的下一條指令, 而CreateProcess是從頭開始執行子進程

  • Windows底層採用CreateProcess函數建立子進程在Python中出現的問題

    • 在main.py中
      ```py

      import time
      import multiprocessing
      from multiprocessing import Process

      def foo():
      time.sleep(3)
      print('this is foo function')

      p = Process(target=foo)
      p.start()

      print('Finish')
      ```
    • 在命令行執行python3 main.py, 報錯: 常見了無限個進程
    • 分析
      • Windows 底層調用CreateProcess函數建立子進程, 建立的子進程會從頭開始執行程序, 對於main.py, 建立一個子進程, 就會再走一遍import time..., 這個時候確定還會遇到Process(), 沒有辦法, 在CreateProcess一次, 一次類推, 一直建立子進程
      • 解決方案
        • 依據Python中主進程的__name__ == __main__而子進程__name__ != __main__規避
        • 代碼

        import time
        import multiprocessing
        from multiprocessing import Process
        
        def foo():
            time.sleep(3)
            print('this is foo function')
        
        if __name__ == '__main__':
            p = Process(target=foo)
            p.start()
            print('Finish')

進程共享

• 通常來講進程中定義的數據是不會共享的, 父進程的數據與子進程中的數據無關, 對於一個通常的全局變量也是不共享的; 在這樣勢必會致使程序運行效率低下, 在Windows中在不一樣進程中打印id(an_obj), 顯示出來的id是不一樣的, 由於Windows中的進程實質上不是fork出來的而是CreateProcessAPI產生的, 子進程要從頭開始走一遍, 具體內容在上文注意中提到, 可是在類Unix中打印出來的id是同樣的, 由於是fork出來的;

進程同步

• 進程之間的數據是不共享的, 可是文件系統, 屏幕等是共享的, 能夠共同訪問一個文件, 一個屏幕(終端), 因此會產生這些資源的競爭, 爲此咱們須要控制他們的競爭關係
• 爲了控制資源的訪問, 誕生了進程鎖, 這裏很特別, 咱們知道兩個進程(A與B)之間的資源是獨立的, 可是multiprocessing中的對象(Lock, Queue等)在兩個進程中內部會有複雜的映射, 目的就是要達到資源共享
  • 在main程序中定義了multiprocessing.Lock(), 在main中fork出A和B兩個子進程執行一個同一段代碼(代碼同樣, 可是不是同一段代碼, 是複製出來的兩份獨立的代碼), lock做爲參數傳入. A中lock.acquire()時, 按道理來講, 進程A與B是獨立的, A中調用了acquire()應該不會影響B, 可是Python內部作了複雜的映射, 當A中lock.acquire()時也會對B中同一段代碼上鎖; 由於lock在內核空間

IPC

• Queue(建議多使用Queue)
  1. Queue內部有鎖機制, 而且支持數據共享
  2. q.put方法用以插入數據到隊列中，put方法還有兩個可選參數：blocked和timeout。若是blocked爲True(默認值)，而且timeout爲正值，該方法會阻塞timeout指定的時間，直到該隊列有剩餘的空間。若是超時，會拋出Queue.Full異常。若是blocked爲False，但該Queue已滿，會當即拋出Queue.Full異常。
  3. q.get方法能夠從隊列讀取而且刪除一個元素。一樣，get方法有兩個可選參數：blocked和timeout。若是blocked爲True（默認值），而且timeout爲正值，那麼在等待時間內沒有取到任何元素，會拋出Queue.Empty異常。若是blocked爲False，有兩種狀況存在，若是Queue有一個值可用，則當即返回該值，不然，若是隊列爲空，則當即拋出Queue.Empty異常.
  4. q.empty():調用此方法時q爲空則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中又加入了項目。
  5. q.full()：調用此方法時q已滿則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中的項目被取走。
  6. q.qsize():返回隊列中目前項目的正確數量，結果也不可靠，理由同q.empty()和q.full()同樣
• Pipe
  1. 與Linux C語言中的Pipe有一些不一樣, Python中的Pipe更加高級, C中的Pipe只能是父子進程之間進行數據交互, 可是Python中的Pipe除了父子之間還能夠是其餘的進程之間
  2. Pipe中一端recv或者send時, 若是全部其餘端口都被close(全部涉及到該Pipe的進程)了, 纔會拋出EOFError異常
  3. send(obj)和recv()方法使用pickle模塊對對象進行序列化
  4. Pipe編程中的close(), send(), recv()等操做必定要在fork完了全部的進程執行, 不然會產生不少意想不到的錯誤
• Manager(共享數據, 內部沒有鎖, 須要本身加鎖)
  1. 就是C中的mmap
  2. 通常配合lock與with使用, 由於Manger本身不會加鎖

  lock = Lock()
  with Manager() as m:
      d = m.dict({'data': 100})
      p = Process(target=foo, args=(d, lock))
      p.start()

• Semaphore(信號量)
  1. 信號量規定了一個資源最多最多能夠被多少個進程訪問, 多出來的會被阻塞
  2. Semaphore(size)
  3. acquire() 鎖資源
  4. release() 釋放資源鎖
• Pool
  1. Pool(3), 建立一個有3個進程的進程池, 從無到有, 最多有3個, 以後就一直是3個
  2. apply 與 apply_async
     • apply
       • 同步執行進程, 會阻塞當前主進程
       • 會當即返回進程執行的結果
     • apply_async
       • 異步執行進程, 不會當即返回結果, 不會阻塞
       • 返回的對象爲ApplyResult, 獲取結果只須要調用對象的get方法, 可是調用get方法, 須要調用p.close(), 再調用p.join()方法以後才能取結果, 不然主程序結束了, 進程池中的任務還沒來得及所有執行完也都跟着主進程一塊兒結束了, 這與咱們直接使用Process不一樣, 要獲取結果get, 建議將全部的進程都執行完再批量查看結果