python併發編程之多進程

時間 2019-11-13

原文原文鏈接

一 multiprocessing模塊介紹python

python中的多線程沒法利用多核優點，若是想要充分地使用多核CPU的資源（os.cpu_count()查看），在python中大部分狀況須要使用多進程。Python提供了multiprocessing。
multiprocessing模塊用來開啓子進程，並在子進程中執行咱們定製的任務（好比函數），該模塊與多線程模塊threading的編程接口相似。數據庫

　 multiprocessing模塊的功能衆多：支持子進程、通訊和共享數據、執行不一樣形式的同步，提供了Process、Queue、Pipe、Lock等組件。編程

須要再次強調的一點是：與線程不一樣，進程沒有任何共享狀態，進程修改的數據，改動僅限於該進程內。json

二 Process類的介紹

建立進程的類：windows

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化獲得的對象，表示一個子進程中的任務（還沒有啓動）

強調：
1. 須要使用關鍵字的方式來指定參數
2. args指定的爲傳給target函數的位置參數，是一個元組形式，必須有逗號

參數介紹：數組

group參數未使用，值始終爲None

target表示調用對象，即子進程要執行的任務

args表示調用對象的位置參數元組，args=(1,2,)

kwargs表示調用對象的字典,kwargs={'name':'abc'}

name爲子進程的名稱

方法介紹：安全

p.start()：啓動進程，並調用該子進程中的p.run() 
p.run():進程啓動時運行的方法，正是它去調用target指定的函數，咱們自定義類的類中必定要實現該方法  

p.terminate():強制終止進程p，不會進行任何清理操做，若是p建立了子進程，該子進程就成了殭屍進程，使用該方法須要特別當心這種狀況。若是p還保存了一個鎖那麼也將不會被釋放，進而致使死鎖
p.is_alive():若是p仍然運行，返回True

p.join([timeout]):主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，須要強調的是，p.join只能join住start開啓的進程，而不能join住run開啓的進程

屬性介紹：服務器

p.daemon：默認值爲False，若是設爲True，表明p爲後臺運行的守護進程，當p的父進程終止時，p也隨之終止，而且設定爲True後，p不能建立本身的新進程，必須在p.start()以前設置

p.name:進程的名稱

p.pid：進程的pid

p.exitcode:進程在運行時爲None、若是爲–N，表示被信號N結束(瞭解便可)

p.authkey:進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是爲涉及網絡鏈接的底層進程間通訊提供安全性，這類鏈接只有在具備相同的身份驗證鍵時才能成功（瞭解便可）

三 Process類的使用

注意：在windows中Process()必須放到# if __name__ == '__main__':下網絡

建立並開啓子進程的兩種方式多線程

from multiprocessing import Process
import time
import random

def task(name):
    print("start %s"%name)
    time.sleep(random.randint(3))
    print("end %s"name)
    
if __name__ == '__main__':
    p = Process(target=task,args=("abc",))
    p.start()
    
    print("主線程")

方法一

from multiprocessing import Process
import time
import random

class MyProcess(Process)
    def __init__(self,name):
        super().__init__()
        self.name = name
    def run():
        print("start %s"%self.name)
        time.sleep(random.randint(3))
        print("end %s"self.name)
    
if __name__ == '__main__':
    p = MyProcess("abc")
    p.start()
    
    print("主線程")

方法二

進程直接的內存空間是隔離的

from multiprocessing import Process
import os

n = 100
def func():
    global n
    n += 10
    print("%s:%d"%(os.getpid(),n))


if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    p.join()
    print("%s:%d"%(os.getpid(),n))
    print("主")


運行結果
17424:110
15080:100
主

View Code

Process對象的join方法

#沒有加join
from multiprocessing import Process
import time
import os

def func():
    time.sleep(1)
    print("start:%s"%(os.getpid()))


if __name__ == '__main__':
    p = Process(target=func)
    p.start()

    print("主:%s"%os.getpid())

運行結果
主:18064
start:7064  #主進程先運行完畢，不會等待子進程

#加join後

from multiprocessing import Process
import time
import os

def func():
    time.sleep(1)
    print("start:%s"%(os.getpid()))

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    p.join()
    print("主:%s"%os.getpid())

運行結果
start:16988
主:1716   #主進程會堵塞等待子進程運行完畢後，才繼續運行

View Code

from multiprocessing import Process
import time
import os

def func(name):
    time.sleep(1)
    print("start:%s"%name)


if __name__ == '__main__':
    start_time = time.time()
    p = Process(target=func,args=("p",))
    p2 = Process(target=func,args=("p2",))
    p.start()
    p2.start()
    p.join()
    p2.join()
    end_time = time.time()
    print("運行時間[%s]"%(end_time-start_time))
    print("主:%s"%os.getpid())

運行結果
start:p
start:p2
運行時間[1.2825298309326172]  #耗時是運行時間最長的進程，而不是之和
主:12056

#join是讓主進程進行堵塞等待，對於其餘的進程是不影響的
上述代碼中同時開啓了進程p和p2在進程p運行的時候,進程p2也是在運行的，等待的只是主進程。

join將程序變成串行嗎？

Process對象的其餘方法或屬性（瞭解）

from multiprocessing import Process
import time
import random


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

    def run(self):
        print("start %s" % self.name)
        time.sleep(random.randint(3))


if __name__ == '__main__':
    p = MyProcess("abc")
    p.start()
    p.terminate()  # 關閉進程,不會當即關閉,因此is_alive馬上查看的結果可能仍是存活
    print(p.is_alive())  # 結果爲True
    time.sleep(0.2)
    print("主線程")
    print(p.is_alive()) #結果爲False

View Code

守護進程

主進程建立守護進程

　　其一：守護進程會在主進程代碼執行結束後就終止

　　其二：守護進程內沒法再開啓子進程,不然拋出異常：AssertionError: daemonic processes are not allowed to have children

注意：進程之間是互相獨立的，主進程代碼運行結束，守護進程隨即終止

from multiprocessing import Process
import time
import random


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

    def run(self):
        print("start %s" % self.name)
        time.sleep(random.randint(3))


if __name__ == '__main__':
    p = MyProcess("abc")
    p.daemon = True  #設置要在p.start()以前
    p.start()
    print("主線程")

運行結果
主進程

#將p設置爲子進程以後，主進程一旦運行完畢，其守護進程無論是否運行完畢，都會被終止。

View Code

#主進程代碼運行完畢,守護進程就會結束
from multiprocessing import Process
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
print("main-------") #打印該行則主進程代碼結束,則守護進程p1應該被終止,可能會有p1任務執行的打印信息123,由於主進程打印main----時,p1也執行了,可是隨即被終止

View Code

進程同步(鎖)

進程之間數據不共享,可是共享同一套文件系統,因此訪問同一個文件,或同一個打印終端,是沒有問題的,

而共享帶來的是競爭，競爭帶來的結果就是錯亂，如何控制，就是加鎖處理

一、多個進程共享同一打印終端

from multiprocessing import Process
import time
import random


def foo(name):
    print("%s start print"%name)
    time.sleep(0.2)
    print("%s end print" % name)

if __name__ == '__main__':
    l = []
    for i in range(3):
        p = Process(target=foo,args=(i,))
        l.append(p)

    for i in l:
        i.start()
    print("主")

運行結果
主
start print
start print
start print
end print
end print
end print

#進程間競爭打印終端，形成數據穿插

沒有加鎖，形成數據的不安全

from multiprocessing import Process,Lock
import time
import random


def foo(name,lock):
    lock.acquire()
    print("%s start print"%name)
    time.sleep(0.2)
    print("%s end print" % name)
    lock.release()

if __name__ == '__main__':
    l = []
    lock = Lock()
    for i in range(3):
        p = Process(target=foo,args=(i,lock,))
        l.append(p)

    for i in l:
        i.start()
    print("主")

運行結果
主
0 start print
0 end print
1 start print
1 end print
2 start print
2 end print

#程序變成串行運行，效率下降，但卻保證了數據的安全

加了鎖，形成了程序的串行，效率下降，保證了數據的安全

二、多個進程共享同一文件

文件當數據庫，模擬搶票

#文件db的內容爲：{"count":1}
#注意必定要用雙引號，否則json沒法識別
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩餘票數%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模擬讀數據的網絡延遲
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模擬寫數據的網絡延遲
        json.dump(dic,open('db.txt','w'))
        print('\033[43m購票成功\033[0m')

def task(lock):
    search()
    get()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模擬併發100個客戶端搶票
        p=Process(target=task,args=(lock,))
        p.start()

併發運行，效率高，但競爭寫同一文件，數據寫入錯亂

併發運行，效率高

#文件db的內容爲：{"count":1}
#注意必定要用雙引號，否則json沒法識別
from multiprocessing import Process,Lock
import time,json,random
def search():
    dic=json.load(open('db.txt'))
    print('\033[43m剩餘票數%s\033[0m' %dic['count'])

def get():
    dic=json.load(open('db.txt'))
    time.sleep(0.1) #模擬讀數據的網絡延遲
    if dic['count'] >0:
        dic['count']-=1
        time.sleep(0.2) #模擬寫數據的網絡延遲
        json.dump(dic,open('db.txt','w'))
        print('\033[43m購票成功\033[0m')

def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(100): #模擬併發100個客戶端搶票
        p=Process(target=task,args=(lock,))
        p.start()

加鎖：購票行爲由併發變成了串行，犧牲了運行效率，但保證了數據安全

加鎖串行

總結

#加鎖能夠保證多個進程修改同一塊數據時，同一時間只能有一個任務能夠進行修改，即串行的修改，沒錯，速度是慢了，但犧牲了速度卻保證了數據安全。
雖然能夠用文件共享數據實現進程間通訊，但問題是：
1.效率低（共享數據基於文件，而文件是硬盤上的數據）
2.須要本身加鎖處理



#所以咱們最好找尋一種解決方案可以兼顧：一、效率高（多個進程共享一塊內存的數據）二、幫咱們處理好鎖問題。這就是mutiprocessing模塊爲咱們提供的基於消息的IPC通訊機制：隊列和管道。
隊列和管道都是將數據存放於內存中
隊列又是基於（管道+鎖）實現的，可讓咱們從複雜的鎖問題中解脫出來，
咱們應該儘可能避免使用共享數據，儘量使用消息傳遞和隊列，避免處理複雜的同步和鎖問題，並且在進程數目增多時，每每能夠得到更好的可獲展性。

View Code

隊列

進程彼此之間互相隔離，要實現進程間通訊（IPC），multiprocessing模塊支持兩種形式：隊列和管道，這兩種方式都是使用消息傳遞的。

建立隊列的類（底層就是以管道和鎖定的方式實現）：

1 Queue([maxsize]):建立共享的進程隊列，Queue是多進程安全的隊列，可使用Queue實現多進程之間的數據傳遞。

參數介紹：

　1 maxsize是隊列中容許最大項數，省略則無大小限制。

方法介紹：

　主要方法

q.put方法用以插入數據到隊列中，put方法還有兩個可選參數：blocked和timeout。若是blocked爲True（默認值），而且timeout爲正值，該方法會阻塞timeout指定的時間，直到該隊列有剩餘的空間。若是超時，會拋出Queue.Full異常。若是blocked爲False，但該Queue已滿，會當即拋出Queue.Full異常。
q.get方法能夠從隊列讀取而且刪除一個元素。一樣，get方法有兩個可選參數：blocked和timeout。若是blocked爲True（默認值），而且timeout爲正值，那麼在等待時間內沒有取到任何元素，會拋出Queue.Empty異常。若是blocked爲False，有兩種狀況存在，若是Queue有一個值可用，則當即返回該值，不然，若是隊列爲空，則當即拋出Queue.Empty異常.
 
q.get_nowait():同q.get(False)
q.put_nowait():同q.put(False)

q.empty():調用此方法時q爲空則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中又加入了項目。
q.full()：調用此方法時q已滿則返回True，該結果不可靠，好比在返回True的過程當中，若是隊列中的項目被取走。
q.qsize():返回隊列中目前項目的正確數量，結果也不可靠，理由同q.empty()和q.full()同樣

View Code

　其餘方法

1 q.cancel_join_thread():不會在進程退出時自動鏈接後臺線程。能夠防止join_thread()方法阻塞
2 q.close():關閉隊列，防止隊列中加入更多數據。調用此方法，後臺線程將繼續寫入那些已經入隊列但還沒有寫入的數據，但將在此方法完成時立刻關閉。若是q被垃圾收集，將調用此方法。關閉隊列不會在隊列使用者中產生任何類型的數據結束信號或異常。例如，若是某個使用者正在被阻塞在get()操做上，關閉生產者中的隊列不會致使get()方法返回錯誤。
3 q.join_thread()：鏈接隊列的後臺線程。此方法用於在調用q.close()方法以後，等待全部隊列項被消耗。默認狀況下，此方法由不是q的原始建立者的全部進程調用。調用q.cancel_join_thread方法能夠禁止這種行爲

View Code

應用

'''
multiprocessing模塊支持進程間通訊的兩種主要形式:管道和隊列
都是基於消息傳遞實現的,可是隊列接口
'''

from multiprocessing import Process,Queue
import time
q=Queue(3)


#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
print(q.full()) #滿了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了

View Code

管道

進程間通訊（IPC）方式二：管道（不推薦使用，瞭解便可）

#建立管道的類：
Pipe([duplex]):在進程之間建立一條管道，並返回元組（conn1,conn2）,其中conn1，conn2表示管道兩端的鏈接對象，強調一點：必須在產生Process對象以前產生管道
#參數介紹：
dumplex:默認管道是全雙工的，若是將duplex射成False，conn1只能用於接收，conn2只能用於發送。
#主要方法：
    conn1.recv():接收conn2.send(obj)發送的對象。若是沒有消息可接收，recv方法會一直阻塞。若是鏈接的另一端已經關閉，那麼recv方法會拋出EOFError。
    conn1.send(obj):經過鏈接發送對象。obj是與序列化兼容的任意對象
 #其餘方法：
conn1.close():關閉鏈接。若是conn1被垃圾回收，將自動調用此方法
conn1.fileno():返回鏈接使用的整數文件描述符
conn1.poll([timeout]):若是鏈接上的數據可用，返回True。timeout指定等待的最長時限。若是省略此參數，方法將當即返回結果。若是將timeout射成None，操做將無限期地等待數據到達。
 
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法發送的一條完整的字節消息。maxlength指定要接收的最大字節數。若是進入的消息，超過了這個最大值，將引起IOError異常，而且在鏈接上沒法進行進一步讀取。若是鏈接的另一端已經關閉，不再存在任何數據，將引起EOFError異常。
conn.send_bytes(buffer [, offset [, size]])：經過鏈接發送字節數據緩衝區，buffer是支持緩衝區接口的任意對象，offset是緩衝區中的字節偏移量，而size是要發送字節數。結果數據以單條消息的形式發出，而後調用c.recv_bytes()函數進行接收    
 
conn1.recv_bytes_into(buffer [, offset]):接收一條完整的字節消息，並把它保存在buffer對象中，該對象支持可寫入的緩衝區接口（即bytearray對象或相似的對象）。offset指定緩衝區中放置消息處的字節位移。返回值是收到的字節數。若是消息長度大於可用的緩衝區空間，將引起BufferTooShort異常。

介紹

from multiprocessing import Process,Pipe

import time,os
def consumer(p,name):
    left,right=p
    left.close()
    while True:
        try:
            baozi=right.recv()
            print('%s 收到包子:%s' %(name,baozi))
        except EOFError:
            right.close()
            break
def producer(seq,p):
    left,right=p
    right.close()
    for i in seq:
        left.send(i)
        # time.sleep(1)
    else:
        left.close()
if __name__ == '__main__':
    left,right=Pipe()

    c1=Process(target=consumer,args=((left,right),'c1'))
    c1.start()


    seq=(i for i in range(10))
    producer(seq,(left,right))

    right.close()
    left.close()

View Code

注意：生產者和消費者都沒有使用管道的某個端點，就應該將其關閉，如在生產者中關閉管道的右端，在消費者中關閉管道的左端。若是忘記執行這些步驟，程序可能在消費者中的recv()操做上掛起。管道是由操做系統進行引用計數的,必須在全部進程中關閉管道後才能生產EOFError異常。所以在生產者中關閉管道不會有任何效果，付費消費者中也關閉了相同的管道端點。

from multiprocessing import Process,Pipe

import time,os
def adder(p,name):
    server,client=p
    client.close()
    while True:
        try:
            x,y=server.recv()
        except EOFError:
            server.close()
            break
        res=x+y
        server.send(res)
    print('server done')
if __name__ == '__main__':
    server,client=Pipe()

    c1=Process(target=adder,args=((server,client),'c1'))
    c1.start()

    server.close()

    client.send((10,20))
    print(client.recv())
    client.close()

    c1.join()
    print('主進程')
#注意：send()和recv()方法使用pickle模塊對對象進行序列化。

管道能夠用於雙向通訊，利用一般在客戶端/服務器中使用的請求／響應模型或遠程過程調用，就可使用管道編寫與進程交互的程序

例

共享數據

展望將來，基於消息傳遞的併發編程是大勢所趨

即使是使用線程，推薦作法也是將程序設計爲大量獨立的線程集合

經過消息隊列交換數據。這樣極大地減小了對使用鎖定和其餘同步手段的需求，

還能夠擴展到分佈式系統中

進程間通訊應該儘可能避免使用本節所講的共享數據的方式

Value、Array是經過共享內存的方式共享數據
Manager是經過共享進程的方式共享數據

Value\Array

from multiprocessing import Process,Lock
import multiprocessing
#Value/Array
def func1(a,arr):
    a.value=3.14
    for i in range(len(arr)):
        arr[i]=-arr[i]
if __name__ == '__main__':
    num=multiprocessing.Value('d',1.0)#num=0
    arr=multiprocessing.Array('i',range(10))#arr=range(10)
    p=multiprocessing.Process(target=func1,args=(num,arr))
    p.start()
    p.join()
    print (num.value)
    print (arr[:])

#執行結果
3.14
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

View Code

Manager管理的共享數據類型有：Value、Array、dict、list、Lock、Semaphore等等，同時Manager還能夠共享類的實例對象。
實例代碼：

from multiprocessing import Process,Manager
def func1(shareList,shareValue,shareDict,lock):
    with lock:
        shareValue.value+=1
        shareDict[1]='1'
        shareDict[2]='2'
        for i in range(len(shareList)):
            shareList[i]+=1

if __name__ == '__main__':
    manager=Manager()
    list1=manager.list([1,2,3,4,5])
    dict1=manager.dict()
    array1=manager.Array('i',range(10))
    value1=manager.Value('i',1)
    lock=manager.Lock()
    proc=[Process(target=func1,args=(list1,value1,dict1,lock)) for i in range(20)]
    for p in proc:
        p.start()
    for p in proc:
        p.join()
    print (list1)
    print (dict1)
    print (array1)
    print (value1)


#運行結果
[21, 22, 23, 24, 25]
{1: '1', 2: '2'}
array('i', [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
Value('i', 21)

View Code

進程池

在利用Python進行系統管理的時候，特別是同時操做多個文件目錄，或者遠程控制多臺主機，並行操做能夠節約大量的時間。多進程是實現併發的手段之一，須要注意的問題是：

很明顯須要併發執行的任務一般要遠大於核數
一個操做系統不可能無限開啓進程，一般有幾個核就開幾個進程
進程開啓過多，效率反而會降低（開啓進程是須要佔用系統資源的，並且開啓多餘核數目的進程也沒法作到並行）

例如當被操做對象數目不大時，能夠直接利用multiprocessing中的Process動態成生多個進程，十幾個還好，但若是是上百個，上千個。。。手動的去限制進程數量卻又太過繁瑣，此時能夠發揮進程池的功效。

咱們就能夠經過維護一個進程池來控制進程數目，好比httpd的進程模式，規定最小進程數和最大進程數...
ps：對於遠程過程調用的高級應用程序而言，應該使用進程池，Pool能夠提供指定數量的進程，供用戶調用，當有新的請求提交到pool中時，若是池尚未滿，那麼就會建立一個新的進程用來執行該請求；但若是池中的進程數已經達到規定最大值，那麼該請求就會等待，直到池中有進程結束，就重用進程池中的進程。

建立進程池的類：若是指定numprocess爲3，則進程池會從無到有建立三個進程，而後自始至終使用這三個進程去執行全部任務，不會開啓其餘進程

　　Pool([numprocess [,initializer [, initargs]]]):建立進程池

參數介紹　　

　　numprocess:要建立的進程數，若是省略，將默認使用cpu_count()的值
　　initializer：是每一個工做進程啓動時要執行的可調用對象，默認爲None
　　initargs：是要傳給initializer的參數組

方法介紹

p.apply(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。須要強調的是：此操做並不會在全部池工做進程中並執行func函數。若是要經過不一樣參數併發地執行func函數，必須從不一樣線程調用p.apply()函數或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變爲可用時，將理解傳遞給callback。callback禁止執行任何阻塞操做，不然將接收其餘異步操做中的結果。
   
p.close():關閉進程池，防止進一步操做。若是全部操做持續掛起，它們將在工做進程終止前完成
P.jion():等待全部工做進程退出。此方法只能在close（）或teminate()以後調用

View Code

應用

from multiprocessing import Pool
import os,time
def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3) #進程池中從無到有建立三個進程,之後一直是這三個進程在執行任務
    res_l=[]
    for i in range(10):
        res=p.apply(work,args=(i,)) #同步調用，直到本次任務執行完畢拿到res，等待任務work執行的過程當中可能有阻塞也可能沒有阻塞，但無論該任務是否存在阻塞，同步調用都會在原地等着，只是等的過程當中如果任務發生了阻塞就會被奪走cpu的執行權限
        res_l.append(res)
    print(res_l)

同步調用apply

apply同步調用

from multiprocessing import Pool
import os,time
def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3) #進程池中從無到有建立三個進程,之後一直是這三個進程在執行任務
    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,)) #同步運行,阻塞、直到本次任務執行完畢拿到res
        res_l.append(res)

    #異步apply_async用法：若是使用異步提交的任務，主進程須要使用jion，等待進程池內任務都處理完，而後能夠用get收集結果，不然，主進程結束，進程池可能還沒來得及執行，也就跟着一塊兒結束了
    p.close()
    p.join()
    for res in res_l:
        print(res.get()) #使用get來獲取apply_aync的結果,若是是apply,則沒有get方法,由於apply是同步執行,馬上獲取結果,也根本無需get

異步調用apply_async