6、并发编程六

# 常用参数
group : 通过源码来看，该源码无实际用途；
target : 表示该进程所调用的对象，即子进程要执行的任务；
args : 表示调用对象的位置参数，元组形式；
kwargs : 表示调用对象的字典，字典形式；
name : 为子进程的名称；
# 常用方法
start() : 启动子进程，并调用该子进程中的run()方法；
run() : 进程启动时运行的方法，它会在子进程中调用target指定的函数，同时，自定义类的类中一定要实现该方法；
terminate() : 强制终止子进程，不会进行任何清理操作，如果创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果该进程还保存了一个锁那么也将不会被释放，进而导致死锁；
is_alive() : 如果进程仍然运行，返回True；
join([timeout]) : 主线程等待子进程终止(强调：是主线程处于等的状态，而子进程是处于运行的状态)，timeout是可选的超时时间；
# 常用属性
daemon : 默认值为False，如果设为True，代表子进程为后台运行的守护进程，当子进程的父进程终止时，子进程也随之终止，需要注意的是，开启守护进程的子进程内，不能创建自己的新进程，必须在子进程start()之前设置；
name : 进程的名称；
pid : 进程的pid；
exitcode : 进程在运行时为None、如果为–N，表示被信号N结束(了解即可)；
authkey : 进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功(了解即可)；

def calc():
    sum=0
    for i in range(90000000):
        sum+=1

process_list=[]
start=datetime.datetime.now()
for _ in range(4):
    process_obj=multiprocessing.Process(target=calc)
    process_list.append(process_obj)
    process_obj.start()
for process in process_list:
    process.join()
print((datetime.datetime.now()-start).total_seconds())
# 5.92691

class Calc(multiprocessing.Process):
    def __init__(self,sum):
        super().__init__()
        self.sum=sum
    def run(self):
        print(os.getpid())
        for i in range(90000000):
            self.sum+=1
        print(self.sum)

p1=Calc(0)
p2=Calc(0)
p3=Calc(0)
p4=Calc(0)
p1.start()
p2.start()
p3.start()
p4.start()

Queue.qsize()：返回当前队列包含的消息数量；
Queue.empty()：如果队列为空，返回True，反之False；
Queue.full()：如果队列满了，返回True,反之False；
Queue.get(block, timeout)：获取队列中的⼀条消息，然后将其从列队中移除，block默认值为True。如果block使⽤默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为⽌，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出Queue.Empty异常；如果block值为False，消息列队如果为空，则会⽴刻抛出Queue.Empty异常；
Queue.get_nowait()：相当Queue.get(False)；
Queue.put(item, block, timeout)：将item消息写⼊队列，block默认值为True，如果block使⽤默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写⼊，此时程序将被阻塞（停在写⼊状态），直到消息列队腾出空间为⽌，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出Queue.Full异常；如果block值为False，消息列队如果没有空间可写⼊，则会⽴刻抛出Queue.Full异常；
Queue.put_nowait(item)：相当于Queue.put(item, False)；

def write_task(queue,count):  # 队列生产者
    for i in range(count):
        if queue.full():
            break
        queue.put(i)
def read_task(queue):  # 队列消费者
    while True:
        print(queue.get()) # 如果队列中没有数据，那么就阻塞

queue=multiprocessing.Queue()
pw=multiprocessing.Process(target=write_task,args=(queue,5))
pr=multiprocessing.Process(target=read_task,args=(queue,))
pw.start()
pr.start()
# 0
# 1
# 2
# 3
# 4

def worker(lock,queue):
    while True:
        lock.acquire()
        if not queue.full():
            time.sleep(0.001)
            queue.put(os.getpid())
            lock.release()
        else:
            lock.release()
            break
Lock = multiprocessing.Lock()
process_list=[]
queue=multiprocessing.Queue(maxsize=5)
for _ in range(5):
    process=multiprocessing.Process(target=worker,args=(Lock,queue))
    process.start()
    process_list.append(process)
for process in  process_list:
    process.join()

while not queue.empty():
    print(queue.get())
# 26785
# 26786
# 26787
# 26785
# 26788

# 主要参数
numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值；
initializer：是每个工作进程启动时要执行的可调用对象，默认为None；
initargs：是要传给initializer的参数组；
# 主要方法
apply(func [, args [, kwargs]]) : 在一个池工作进程中执行func(*args,**kwargs)，然后返回结果。需要强调的是，此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()；
apply_async(func, args=(), kwds={}, callback=None,error_callback=None) : 在一个池工作进程中执行func(*args,**kwargs)，此方法的结果是AsyncResult类的实例，callback是可调用对象，当进程正确完成之后，会执行callback中的函数，该函数接受一个参数，callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。error_callback如果进程中出现错误，那么会调用error_callback这个函数，同样接受一个参数，该参数为字符串异常类型描述；
close() : 正常关闭进程池，调用close()方法后，不会有新的进程启动，它等待正在执行的任务完成之后，关闭进程池；
terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数；
jion() : 等待所有工作进程退出。此方法只能在close（）或teminate()之后调用；
# AsyncResul对象方法
get() : 返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发；
ready() : 如果调用完成，返回True；
successful() : 如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常；
wait([timeout]) : 等待结果变为可用；

def calc():
    sum=0
    for i in range(90000000):
        sum+=1
    return sum

for _ in range(4):
    pool=multiprocessing.Pool(processes=4,)
    result=pool.apply(calc,)
    print(result)
    pool.close()
# 90000000
# 90000000
# 90000000
# 90000000

def calc():
    sum=0
    for i in range(9000000):
        sum+=1
    print(sum)
pool=multiprocessing.Pool(processes=4,)

def handler():
    pool.apply(calc,)
    pool.close()
    pool.join()
for i in range(4):
    threading.Thread(target=handler,).start()
# 9000000
# 9000000
# 9000000
# 9000000

def calc():
    sum=0
    for i in range(90000000):
        sum+=1
    print(sum)
pool=multiprocessing.Pool(processes=4,)
for _ in range(4):
    result=pool.apply_async(calc,)
    print(result)
pool.close()
pool.join()

def calc():
    sum=0
    for i in range(90000000):
        sum+=1
    return sum
pool=multiprocessing.Pool(processes=4,)
for _ in range(4):
    pool.apply_async(calc,callback=lambda func_result:print(func_result))
pool.close()
pool.join()
# 90000000
# 90000000
# 90000000
# 90000000

ThreadPoolExecutor(max_workers=1)：池中至多创建max_workers个线程的池来同时异步执行，返回Executor实例；
ProcessPoolExecutor(max_workers=1)：池中至多创建max_workers个进程的池来同时异步执行，返回Executor实例；
wait(fs, timeout=None, return_when=ALL_COMPLETED)：接受一个元素为future对象的可迭代对象，它判断这个可迭代对象里面的future对象里面的线程是否已经执行完成，有三种选择，即return_when参数，具体看源码；
as_completed(fs, timeout=None)：接受一个元素为future对象的可迭代对象，一个一个地等待，future对象里面的线程完成，它返回一个生成器，在没有一个线程或者进程执行完成前，处于阻塞状态，完成一个返回一个；

submit(fn,*args,**kwargs)：提交线程需要执行的函数及其参数，返回Future类的实例；
shutdown(wait=True)：清理进程池，wait为True表示池清理干净才退出；
# Future方法
done()：如果调用被成功取消或者执行完成，则返回True，需要注意的是，任务异常终止也算finished；
cancelled()：如果调用被成功的取消，则返回True；
running()：如果正在运行且不可取消，则返回True；
cancel()：尝试取消调用，只能取消pending状态的进程或者线程，取消成功返回True，否则返回False；
result(timeout=None)：取返回结果，timeout为None时会一直等待返回，timeout如果设置等待时，时间到还未拿到返回结果抛出concurrent.futures.TimeoutError异常，需要注意的是，虽然result能返回执行结果，但是它是会阻塞的；
exception(timeout=None)：取返回结果，timeout为None时会一直等待返回，timeout如果设置等待时，时间到还未拿到返回结果抛出concurrent.futures.TimeoutError异常，同时，如果内部异常，exception也可以拿到异常信息；

def worker(n):
    print("enter thread")
    time.sleep(n)
    print(threading.enumerate())
    print("finished")

executor=ThreadPoolExecutor(max_workers=3)
for _ in range(10):
    future=executor.submit(worker,2)

time.sleep(1)
print(threading.enumerate())
# enter thread
# enter thread
# enter thread
# [<_MainThread(MainThread, started 140734813474240)>, 
# <Thread(Thread-1, started daemon 123145343614976)>, 
# <Thread(Thread-2, started daemon 123145348870144)>,
# <Thread(Thread-3, started daemon 123145354125312)>]

def worker(n):
    time.sleep(n)

executor=ThreadPoolExecutor(max_workers=3)
future_list=[]
for _ in range(3):
    future=(executor.submit(worker,2))
    future_list.append(future)
while True:
    states=[state.done() for state in future_list]
    print(states)
    if all(states):
        print("done...")
        break
    time.sleep(1)
# [False, False, False]
# [False, False, False]
# [True, True, True]
# done...

def worker(n):
    time.sleep(n)

executor=ThreadPoolExecutor(max_workers=3)
future_list=[]
for _ in range(3):
    future=(executor.submit(worker,2))
    future_list.append(future)
done,not_done=wait(future_list,return_when=ALL_COMPLETED)
print(done) # {<Future at 0x106455b00 state=finished returned NoneType>, <Future at 0x1060eb3c8 state=finished returned NoneType>, <Future at 0x1060ade80 state=finished returned NoneType>}
print(not_done) # set()

def worker(n):
    time.sleep(n*3)

executor=ThreadPoolExecutor(max_workers=3)
future_list=[]
for i in range(3):
    future=(executor.submit(worker,i))
    future_list.append(future)

for completed in as_completed(future_list):
    print(completed.done())
# True
# True
# True

def worker(n):
    time.sleep(n*3)

with ProcessPoolExecutor(max_workers=3) as executor:
    future_list=[]
    for i in range(3):
        future=(executor.submit(worker,i))
        future_list.append(future)

    for completed in as_completed(future_list):
        print(completed.done())
print(threading.enumerate())
# True
# True
# True
# [<_MainThread(MainThread, started 140734964358592)>]   # 最后仅剩下一个主进程的主线程