五、python多進程與多線程。

  1. 進程
    進程的概念是需要理解的，進程是操作系統(tǒng)中正在運行的一個程序?qū)嵗?，操作系統(tǒng)通過進程操作原語來對其進行調(diào)度。操作系統(tǒng)得到調(diào)用某個進程指令時，將硬盤上的程序調(diào)入內(nèi)存，分配空間，初始化進程堆棧，然后進程開始運行。有時候我們有同時運行多個程序的需求，如果你的電腦只能做一件事，那是一件很抓狂的事。操作系統(tǒng)通過進程調(diào)度算法調(diào)度進程運行，使計算機看起來同時運行了很多程序。
  2. python中多進程實現(xiàn)
    • fork，fork是linux下創(chuàng)建新進程的機制，通過fork父進程復制出一個相似，通過fork返回值判斷執(zhí)行子進程代碼 
       import os
 def main():
     print 'current Process {} start...'.format(os.getpid())
     pid = os.fork()
     if pid < 0:
         print 'fork error!'
         exit(1)
     elif pid == 0:
         print 'child Process {} starting... , and my parent process is {}'.format(os.getpid(), os.getppid())
 
     else:
         print 'I({}) created the child({})'.format(os.getpid(), pid)
 if __name__ == '__main__':
         main()
      得到如下輸出 
      current Process 5351 start...
I(5351) created the child(5352)
child Process 5352 starting... , and my parent process is 5351
    • 使用multiprocess模塊創(chuàng)建子進程，模塊提供一個Process對象描述進程，創(chuàng)建進程時，只需要傳入一個可調(diào)用的函數(shù)，以及函數(shù)運行時的參數(shù)即可 
       import os
 import multiprocessing
 
 
 def run_proc(name):
     print 'child process {}({}) running...'.format(name, os.getpid())
 
 def main():
     print 'main process starting... {}'.format(os.getpid())
     processes = []
     for i in range(5):
         p = multiprocessing.Process(target=run_proc, args=(str(i),))
         processes.append(p)
         print 'process {} will start'
         p.start()
 
     for p in processes:
         p.join()
     print 'processes end'
 if __name__ == '__main__':
     main()
      • 使用進程池限制進程個數(shù)。multiprocessing模塊中的Pool對象，用來表示進程池，Pool對象的apply_async函數(shù)用于創(chuàng)建進程，同樣的給出可調(diào)用的函數(shù)與函數(shù)運行需要的參數(shù) 
         import os
 from multiprocessing import Pool
 
 def run_proc(name):
     print 'child process {}({}) running...'.format(name, os.getpid())
 
 def main():
     print 'main process starting... {}'.format(os.getpid())
     processes = Pool(processes=3)
     for i in range(5):
         processes.apply_async(run_proc,(str(i),))
     processes.close()
     processes.join()
 
 
     print 'processes end'
 if __name__ == '__main__':
     main()
    • 進程間通信
      • 通過隊列。
        隊列，即multiprocessing模塊中的Queue對象，隊列中有某種資源，可以向隊列中放入數(shù)據(jù)，另一個進程從隊列中取出數(shù)據(jù)，當無數(shù)據(jù)可用時，消費者應該決定是阻塞等待資源還是返回一個錯誤，當隊列已滿，生產(chǎn)者應決定是阻塞等待可用空間還是返回錯誤。Queue對象有兩個主要方法，get和put，get從隊列中取出數(shù)據(jù)，put向隊列中添加數(shù)據(jù)。blocked參數(shù)決定當隊列不滿足條件時是阻塞等待還是返回錯誤，默認為True，表示阻塞等待。timeout指定了隊列阻塞的時間，如果超時，同樣返回異常 
         from multiprocessing import Queue, Process
 import os, time, random
 
 def Proc_writer(q, urls):
     print 'Process {} is writing...'.format(os.getpid())
     for url in urls:
         q.put(url)
         print 'put {} to the Queue'.format(url)
         time.sleep(random.random())
 
 def Proc_reader(q):
     print 'Process {} is reading...'.format(os.getpid())
     while True:
         url = q.get(True)
         print 'get the {} from the Queue'.format(url)
 
 def main():
     print 'main process {} is running...'.format(os.getpid())
     q = Queue()
     process_1 = Process(target=Proc_writer, args=(q,['url_1', 'url_2', 'url_3']))
     process_2 = Process(target=Proc_writer, args=(q,['url_4', 'url_5', 'url_6']))
     process_3 = Process(target=Proc_reader, args=(q,))
     process_1.start()
     process_2.start()
     process_3.start()
     process_1.join()
     process_2.join()
     process_3.terminate()
     print 'done'
 
 
 if __name__ == '__main__':
     main()
      • 通過管道
        multiprocessing模塊的Pipe方法，返回一個二元組（conn1，conn2），Pipe方法有一個duplex參數(shù)，為True時代表管道連接是全雙工的，為False時代表管道連接是單方向的，只能由conn2發(fā)送到conn1。send和recv方法用于發(fā)送與接受消息，如果沒有消息可接受，recv阻塞，如果管道關(guān)閉，recv會拋出EOFError 
         import multiprocessing
 import os, time, random
 
 def proc_send(pipe, urls):
     print 'process {} is read to send urls'.format(os.getpid())
     for url in urls:
         pipe.send(url)
         print 'process {}: send {}'.format(os.getpid(), url)
         time.sleep(random.random())
 
 def proc_recv(pipe):
     print 'process {} is ready to recv urls'.format(os.getpid())
     while True:
         print 'process {}: recv {}'.format(os.getpid(), pipe.recv())
         time.sleep(random.random())
 
 def main():
     pipe = multiprocessing.Pipe()
     process_send = multiprocessing.Process(
         target=proc_send,
         args=(pipe[0], ['url_' + str(i) for i in range(10)]))
     process_recv = multiprocessing.Process(
         target=proc_recv,
         args=(pipe[1],)
     )
     process_send.start()
     process_recv.start()
     process_send.join()
     process_recv.join()
     print 'done'
 
 if __name__ == '__main__':
     main()
    • 分布式多進程
      分布式也是一個比較重要的概念，通過將負載高的計算分攤到多臺計算機上來提高系統(tǒng)性能。使用python完成分布式計算功能是簡單的。需要用到的一個數(shù)據(jù)結(jié)構(gòu)是隊列，聯(lián)想一下操作系統(tǒng)中的生產(chǎn)者消費者模型，一些進程放入數(shù)據(jù)，一些進程取出數(shù)據(jù)。程序開始需要在服務端維護一個網(wǎng)絡隊列管理器，服務端程序注冊操作網(wǎng)絡隊列的方法，隨后使用該方法從網(wǎng)絡上獲取隊列，對該隊列的操作，對網(wǎng)絡上的其他進程是可見的。隊列的put和get方法用于放入取出數(shù)據(jù)，注意服務端和客戶端注冊的接口方法需統(tǒng)一。
      使用multiprocessing子模塊managers管理網(wǎng)絡隊列，其中的BaseManager類是一個基本的管理器，新建類繼承該類。使用該類的register方法注冊操作隊列的方法，隨后監(jiān)聽信道。如下例程
      server 
       #!/usr/bin/env python
 import Queue
 from multiprocessing.managers import BaseManager
 
 # 創(chuàng)建隊列實體
 task_queue = Queue.Queue()
 result_queue = Queue.Queue()
 
 class Queuemanager(BaseManager):
     pass
 
 # 注冊方法
 print 'register the func'
 Queuemanager.register('get_task_queue', callable=lambda:task_queue)
 Queuemanager.register('get_result_queue', callable=lambda:result_queue)
 
 # 創(chuàng)建manager對象
 print 'initialing the task manager'
 manager = Queuemanager(address=('192.168.56.1', 8000), authkey='password')
 
 # 開始監(jiān)聽
 manager.start()
 # 從網(wǎng)絡得到隊列
 print 'get the queue from network...'
 task = manager.get_task_queue()
 result = manager.get_result_queue()
 # 向隊列中放入數(shù)據(jù)等待處理
 print 'put urls to the task queue'
 for url in ['ImageUrl_' + str(i) for i in range(10)]:
     print 'put {} in task'.format(url)
     task.put(url)
 # 從隊列中取出數(shù)據(jù)，阻塞等待
 for i in range(10):
     print 'result is {}'.format(result.get())
 
 manager.shutdown()
      client 
            #!/usr/bin/env python
from multiprocessing.managers import BaseManager
import Queue


class Queuemanager(BaseManager):
    pass

Queuemanager.register('get_task_queue')
Queuemanager.register('get_result_queue')

server = '192.168.56.1'
port = 8000
key = 'password'
print 'try to connect to {}'.format(server)
manager = Queuemanager(address=(server, port), authkey=key)
manager.connect()

task = manager.get_task_queue()
result = manager.get_result_queue()

while not task.empty():
    image_url = task.get(True, timeout=10)
    print 'run task download {}'.format(image_url)
    result.put(image_url + '------>completed!')

print 'worker exit!'
  1. 線程
    線程是一個存在于進程中的概念，用于在進程中并行完成不同的工作。線程與進程的不同另做介紹
  2. python中的多線程

    • threading推薦使用的多線程模塊
      threading中的模塊對象



      threading中的常見方法


      Thread類



      初始化一個thread類來創(chuàng)建一個線程，我們可以

      • 初始化Thread類，傳入我們要運行的函數(shù)與參數(shù)
      • 初始化Thread類，傳入可調(diào)用對象，比如自定義可調(diào)用類
      • 創(chuàng)建類繼承Thread，覆蓋run函數(shù)

    • threading模塊實例
      直接使用thread類

      #!/usr/bin/env python
import threading
from time import ctime, sleep


secLoop = [6, 4]
def loop(sec, i):
    print 'loop', i, 'start at', ctime()
    sleep(sec)
    print 'loop', i, 'finished at', ctime()


def main():
    nloop = range(len(secLoop))
    threads = []
    for i in nloop:
        threads.append(threading.Thread(target=loop, args=(secLoop[i], i)))

    for i in nloop:
        threads[i].start()

    for i in nloop:
        threads[i].join()

if __name__ == '__main__':
    main()

      自定義可調(diào)用類

      import threading
from time import ctime, sleep


secLoop = [6, 4]
def loop(sec, i):
    print 'loop', i, 'start at', ctime()
    sleep(sec)
    print 'loop', i, 'finished at', ctime()

class ThreadFunc(object):
    def __init__(self, func, args):
        self.func = func
        self.args = args

    def __call__(self):
        apply(self.func,self.args)

def main():
    nloop = range(len(secLoop))
    threads = []
    for i in nloop:
        threads.append(threading.Thread(target=ThreadFunc(loop,(secLoop[i],i))))

    for i in nloop:
        threads[i].start()

    for i in nloop:
        threads[i].join()

if __name__ == '__main__':
    main()
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關(guān)閱讀更多精彩內(nèi)容

  • 一文讀懂Python多線程 1、線程和進程 計算機的核心是CPU，它承擔了所有的計算任務。它就像一座工廠，時刻在運...
    星丶雲(yún)閱讀 1,595評論 0贊 4
  • 我們前面提到了進程是由若干線程組成的，一個進程至少有一個線程。多線程優(yōu)點： 在一個進程中的多線程和主線程分享相同的...
    第八共同體閱讀 597評論 0贊 0
  • python中threading庫 另一篇文章：pthreading庫詳解 threading模塊 threadi...
    咫尺天涯var閱讀 2,487評論 0贊 0
  • 線程 引言&動機 考慮一下這個場景，我們有10000條數(shù)據(jù)需要處理，處理每條數(shù)據(jù)需要花費1秒，但讀取數(shù)據(jù)只需要0....
    不浪漫的浪漫_ea03閱讀 415評論 0贊 0
  • 引言&動機 考慮一下這個場景，我們有10000條數(shù)據(jù)需要處理，處理每條數(shù)據(jù)需要花費1秒，但讀取數(shù)據(jù)只需要0.1秒，...
    chen_000閱讀 584評論 0贊 0

友情鏈接更多精彩內(nèi)容