Python一篇学会多进程

阅读目录

• 1. Process
• 2. Lock
• 3. Semaphore
• 4. Event
• 5. Queue
• 6. Pipe
• 7. Pool

            import
             multiprocessing 
            import
             time  
            def
             worker(interval):     n = 5
    while n > 0:         print("The time is {0}".format(time.ctime()))         time.sleep(interval)         n -= 1

if__name__ == "__main__":     p = multiprocessing.Process(target = worker, args = (3,))     p.start()     print"p.pid:", p.pid     print"p.name:", p.name     print"p.is_alive:", p.is_alive()

            import
             multiprocessing 
            import
             time  
            def
             worker_1(interval):     
            print
            "
            worker_1
            "
                 time.sleep(interval)     
            print
            "
            end worker_1
            "
            def
             worker_2(interval):     
            print
            "
            worker_2
            "
                 time.sleep(interval)     
            print
            "
            end worker_2
            "
            def
             worker_3(interval):     
            print
            "
            worker_3
            "
                 time.sleep(interval)     
            print
            "
            end worker_3
            "
            if
            __name__ == "__main__":     p1 = multiprocessing.Process(target = worker_1, args = (2,))     p2 = multiprocessing.Process(target = worker_2, args = (3,))     p3 = multiprocessing.Process(target = worker_3, args = (4,))      p1.start()     p2.start()     p3.start()      print("The number of CPU is:" + str(multiprocessing.cpu_count()))     for p in multiprocessing.active_children():         print("child   p.name:" + p.name + "\tp.id" + str(p.pid))     print"END!!!!!!!!!!!!!!!!!"

            import
             multiprocessing 
            import
             time  
            class
             ClockProcess(multiprocessing.Process):     
            def
            __init__
            (self, interval):         multiprocessing.Process.
            __init__
            
                (self)         self.interval = interval      def run(self):         n = 5
        while n > 0:             print("the time is {0}".format(time.ctime()))             time.sleep(self.interval)             n -= 1

if__name__ == ‘__main__‘:     p = ClockProcess(3)     p.start()      

            import
             multiprocessing 
            import
             time  
            def
             worker(interval):     
            print("work start:{0}".format(time.ctime()));     time.sleep(interval)     print("work end:{0}".format(time.ctime()));  if__name__ == "__main__":     p = multiprocessing.Process(target = worker, args = (3,))     p.start()     print"end!"

            import
             multiprocessing 
            import
             time  
            def
             worker(interval):     
            print("work start:{0}".format(time.ctime()));     time.sleep(interval)     print("work end:{0}".format(time.ctime()));  if__name__ == "__main__":     p = multiprocessing.Process(target = worker, args = (3,))     p.daemon = True     p.start()     print"end!"

            import
             multiprocessing 
            import
             time  
            def
             worker(interval):     
            print("work start:{0}".format(time.ctime()));     time.sleep(interval)     print("work end:{0}".format(time.ctime()));  if__name__ == "__main__":     p = multiprocessing.Process(target = worker, args = (3,))     p.daemon = True     p.start()     p.join()     print"end!"

            import
             multiprocessing 
            import
             sys  
            def
             worker_with(lock, f):     with lock:         fs = open(f, ‘a+‘)         n = 10
        while n > 1:             fs.write("Lockd acquired via with\n")             n -= 1         fs.close()          def worker_no_with(lock, f):     lock.acquire()     try:         fs = open(f, ‘a+‘)         n = 10
        while n > 1:             fs.write("Lock acquired directly\n")             n -= 1         fs.close()     finally:         lock.release()      if__name__ == "__main__":     lock = multiprocessing.Lock()     f = "file.txt"     w = multiprocessing.Process(target = worker_with, args=(lock, f))     nw = multiprocessing.Process(target = worker_no_with, args=(lock, f))     w.start()     nw.start()     print"end"

            import
             multiprocessing 
            import
             time  
            def
             worker(s, i):     s.acquire()     
            print(multiprocessing.current_process().name + "acquire");     time.sleep(i)     print(multiprocessing.current_process().name + "release\n");     s.release()  if__name__ == "__main__":     s = multiprocessing.Semaphore(2)     for i in range(5):         p = multiprocessing.Process(target = worker, args=(s, i*2))         p.start()

            import
             multiprocessing 
            import
             time  
            def
             wait_for_event(e):     
            print("wait_for_event: starting")     e.wait()     print("wairt_for_event: e.is_set()->" + str(e.is_set()))  def wait_for_event_timeout(e, t):     print("wait_for_event_timeout:starting")     e.wait(t)     print("wait_for_event_timeout:e.is_set->" + str(e.is_set()))  if__name__ == "__main__":     e = multiprocessing.Event()     w1 = multiprocessing.Process(name = "block",             target = wait_for_event,             args = (e,))      w2 = multiprocessing.Process(name = "non-block",             target = wait_for_event_timeout,             args = (e, 2))     w1.start()     w2.start()      time.sleep(3)      e.set()     print("main: event is set")

            import
             multiprocessing  
            def
             writer_proc(q):           
            try
            :                  q.put(1, block = False)      except:                  passdef reader_proc(q):           try:                  print q.get(block = False)      except:                  passif__name__ == "__main__":     q = multiprocessing.Queue()     writer = multiprocessing.Process(target=writer_proc, args=(q,))       writer.start()         reader = multiprocessing.Process(target=reader_proc, args=(q,))       reader.start()        reader.join()       writer.join()

                import
                 multiprocessing 
                import
                 time  
                def
                 proc1(pipe):     
                while
                 True:         
                for i in xrange(10000):             print"send: %s" %(i)             pipe.send(i)             time.sleep(1)  def proc2(pipe):     while True:         print"proc2 rev:", pipe.recv()         time.sleep(1)  def proc3(pipe):     while True:         print"PROC3 rev:", pipe.recv()         time.sleep(1)  if__name__ == "__main__":     pipe = multiprocessing.Pipe()     p1 = multiprocessing.Process(target=proc1, args=(pipe[0],))     p2 = multiprocessing.Process(target=proc2, args=(pipe[1],))     #p3 = multiprocessing.Process(target=proc3, args=(pipe[1],))     p1.start()     p2.start()     #p3.start()     p1.join()     p2.join()     #p3.join()

结果

            #
            coding: utf-8
            import
             multiprocessing 
            import
             time  
            def
             func(msg):     
            print
            "
            msg:
            "
            , msg     time.sleep(3)     print"end"if__name__ == "__main__":     pool = multiprocessing.Pool(processes = 3)     for i in xrange(4):         msg = "hello %d" %(i)         pool.apply_async(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去print"Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"     pool.close()     pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print"Sub-process(es) done."

            #
            coding: utf-8
            import
             multiprocessing 
            import
             time  
            def
             func(msg):     
            print
            "
            msg:
            "
            , msg     time.sleep(3)     print"end"if__name__ == "__main__":     pool = multiprocessing.Pool(processes = 3)     for i in xrange(4):         msg = "hello %d" %(i)         pool.apply(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去print"Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"     pool.close()     pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print"Sub-process(es) done."

            import
             multiprocessing 
            import
             time  
            def
             func(msg):     
            print
            "
            msg:
            "
            , msg     time.sleep(3)     print"end"return"done" + msg  if__name__ == "__main__":     pool = multiprocessing.Pool(processes=4)     result = []     for i in xrange(3):         msg = "hello %d" %(i)         result.append(pool.apply_async(func, (msg, )))     pool.close()     pool.join()     for res in result:         print":::", res.get()     print"Sub-process(es) done."

            #
            coding: utf-8
            import
             multiprocessing 
            import
             os, time, random  
            def
             Lee():     
            print
            "
            \nRun task Lee-%s
            " %(os.getpid()) #os.getpid()获取当前的进程的ID
    start = time.time()     time.sleep(random.random() * 10) #random.random()随机生成0-1之间的小数
    end = time.time()     print‘Task Lee, runs %0.2f seconds.‘ %(end - start)  def Marlon():     print"\nRun task Marlon-%s" %(os.getpid())     start = time.time()     time.sleep(random.random() * 40)     end=time.time()     print‘Task Marlon runs %0.2f seconds.‘ %(end - start)  def Allen():     print"\nRun task Allen-%s" %(os.getpid())     start = time.time()     time.sleep(random.random() * 30)     end = time.time()     print‘Task Allen runs %0.2f seconds.‘ %(end - start)  def Frank():     print"\nRun task Frank-%s" %(os.getpid())     start = time.time()     time.sleep(random.random() * 20)     end = time.time()     print‘Task Frank runs %0.2f seconds.‘ %(end - start)          if__name__==‘__main__‘:     function_list=  [Lee, Marlon, Allen, Frank]      print"parent process %s" %(os.getpid())      pool=multiprocessing.Pool(4)     for func in function_list:         pool.apply_async(func)     #Pool执行函数，apply执行函数,当有一个进程执行完毕后，会添加一个新的进程到pool中print‘Waiting for all subprocesses done...‘     pool.close()     pool.join()    #调用join之前，一定要先调用close() 函数，否则会出错, close()执行后不会有新的进程加入到pool,join函数等待素有子进程结束print‘All subprocesses done.‘