一. multiprocessing类
    6. 管道
    进程间通信(ipc)方式二: 管道会导致数据不安全的情况, 后面我们会说到为什么会带来数据不安全的问题
    创建管道的类:
    Pipe([duplex]): 在进程之间创建一条管道, 并返回元祖(conn1, conn2), 其中conn1和conn2表示管道两端的连接对象, 强调一点: 必须在产生Process对象之前产生管道
    参数介绍:
    dumplex: 默认管道是双全工的, 如果将duplex改成False, conn1只能用于接收, conn2只能用于发送
    主要方法:
    conn1.recv(): 接收conn2.send(obj)发送的对象, 如果没有消息可接收, recv方法会一直阻塞, 如果连接的另外一端已关闭, 那么recv方法会抛出EOFError
    conn1.send(): 通过连接发送对象, obj是序列化兼容的任意对象
    其他方法: 
    conn1.close(): 关闭连接, 如果conn1被垃圾回收, 将自动调用此方法
    conn1.fileno(): 返回连接使用的整数文件描述符
    conn1.poll([timeout]): 如果连接上的数据可用, 返回True, timeout指定等待的最长时限, 如果省略此参数, 方法将立即返回结果, 如果将timeout改成None, 操作将无限期地等待数据到达 
    conn1.recv_bytes([maxlength]): 接收c.send_bytes()方法发送的一条完整的字节消息, maxlength指定要接收的最大字节数, 如果进入的消息超过了这个最大值, 将引发EOFError异常
    conn1.send_bytes(buffer[, offset[, size]]): 通过连接发送字节数据缓存区, buffer是支持缓冲区接口的任意对象, offset是缓冲区中的字节偏移量, 而size是要发送的字节数, 结果数据以单条消息发送, 然后调用c.recv_bytes()函数接收
    conn1.recv_bytes_info(buffer[, offset]): 接收一条完整的字节消息, 并把它保存在buffer对象中, 该对象支持可写入的缓冲区接口(即bytearray对象或类似的对象). offset指定缓冲区中放置消息处的字节位移, 返回值是收到的字节数, 如果消息长度大于可用的缓冲区大小, 将引发BufferTooShort异常
    应该特别注意管道端点的正确管理问题, 如果是生产者或消费者中都没有使用管道的某个端点, 就应该将他关闭, 这也说明了为何在生产者中关闭了管道的输出端, 在消费者中关闭了管道的输入端. 如果忘记这些步骤, 程序可能在消费者中的recv()操作上挂起(就是阻塞). 管道是有操作系统进行引用计数的, 如果在所有进程中关闭管道的相同一端就会生成EOFError异常, 因此, 在生产者中关闭管道不会有任何效果, 除非消费者也关闭了相同的管道端点
    from multiprocessing import Process, Pipe
    def f(parent_conn, child_conn):
    # parent_conn.close()    # 不写close将不会引发EOFError
    while 1:
        try:
            print(child_conn.recv())
        except EOFError:
            child_conn.close()
            break
    if __name__ == "__main__":
        parent_conn, child_conn = Pipe()
        p = Process(target=f, args=(parent_conn, child_conn,))
        p.start()
        child_conn.close()
        parent_conn.send("hello")
        parent_conn.close()
        p.join()
    主进程将管道两端都传递给了子进程, 子进程和主进程共用管道的两种报错情况, 都是在recv接收的时候报错的: (1). 主进程和子进程中的管道的相同一端都关闭了, 出现EOFError (2). 如果你管道的一端在主进程和子进程中都关闭了, 但是你还用这个关闭的一端去接收消息, 那么就会出现OSError.
    所以关闭管道的时候就容易出现问题, 需要将所有只用这个管道的进程中的两端全部关闭才行, 当然也可以通过捕获异常(try: except EOFError:)来处理
    虽然我们在主进程和子进程中都打印了一下conn一端的对象, 发现两个不在同一个地址, 但是子进程中的管道 和主进程中的管道还是可以通信的, 因为管道是同一套, 系统能够记录
    我们的目标是关闭所有的管道, 那么主进程和子进程进行通信的时候, 可以给子进程传管道的一端就够了, 并且用我们之前学到的, 信息发送完之后, 再发送衣蛾结束信号None, 那么你收到的消息为None的时候直接结束接收或者说结束循环, 就不用每次都关闭各个进程中的管道了
    from multiprocessing import Process, Pipe
    def consumer(p, name):
        produce, consume = p
        produce.close()
        while 1:
            try:
                baozi = consume.recv()
                print("%s 收到包子: %s" % (name, baozi))
            except EOFError:
                break

def producer(seq, p):
        produce, consume = p
        consume.close()
        for i in seq:
            produce.send(i)

if __name__ == "__main__":
        produce, consume = Pipe()
        c1 = Process(target=consumer, args=((produce, consume), "c1"))
        c1.start()

seq = (i for i in range(10))
        producer(seq, (produce, consume))

produce.close()
        consume.close()

c1.join()
        print("主进程结束")
    # c1 收到包子: 0
    # c1 收到包子: 1
    # c1 收到包子: 2
    # c1 收到包子: 3
    # c1 收到包子: 4
    # c1 收到包子: 5
    # c1 收到包子: 6
    # c1 收到包子: 7
    # c1 收到包子: 8
    # c1 收到包子: 9
    # 主进程结束
    由于Pipe方法返回的两个连接对象表示管道的两端, 每个连接对象都有send和recv方法, 注意, 如果两个进程试图同时从管道的同一端读取或写入数据, 那么管道中的数据可能会损坏,当然, 在使用管道的不同端部不存在损坏风险
    from multiprocessing import Process, Pipe, Lock

def consumer(p, name, lock):
        priduce, consume = p
        produce.close()
        while 1:
            lock.acquire()
            baozi = consume.recv()
            lock.release()
            if baozi:
                print("%s 收到包子: %s" % (name, baozi))
            else:
                consume.close()
                break
        
    def producer(p, n):
        produce, consume = p
        consume.close()
        for i in range(n):
            produce.send(i)
        produce.send(None)
        produce.send(None)
        produce.close()
    
    if __name__ == "__main__":
        produce, consume = Pipe()
        lock = Lock()
        c1 = Process(target=consumer, args=((produce, consume), "c1", lock))
        c2 = Process(target=consumer, args=((produce, consume), "c2", lock))
        p1 = Process(target=producer, args=((produce, consume), 10))
        c1.start()
        c2.start()
        p1.start()

produce.close()
        consume.close()

c1.join()
        c2.join()
        p1.join()
        print("主程序结束")
    管道通常用于双工通信, 通常利用在客户端/服务端中使用的请求/响应模型, 或者远程过程调用, 就可以使用管道编写与进程交互的程序, 像前面讲网络通信的时候, 我们使用了一个叫subprocess的模块, 里面有个参数是pipe通道, 执行系统命令, 并通过管道获取结果
    7. 数据共享
    基于消息传递的并发编程是大势所趋
    即使是使用线程, 推荐做法也是讲程序设计成为大量独立的线程集合
    通过消息队列交换数据, 这样极大地减少了对使用锁定和其他同步手段的需求,还可以扩展到分布式系统在哪个
    进程之间应尽量避免通信, 以后我们会尝试使用数据库来解决进程之间的数据共享问题
    进程之间数据共享的模块之一Manager模块：
    进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的
    虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此
    多进程共同去处理共享数据的时候，就和我们多进程同时去操作一个文件中的数据是一样的，不加锁就会出现错误的结果，进程不安全的，所以也需要加锁
    from multiprocessing import Manager, Process, Lock
    def work(d, lock):
        with lock:      # 不加锁而操作共享的数据, 肯定会出现数据错乱
            d["count"] -= 1
    if __name__ == '__main__':
        lock = Lock()
        with Manager() as m:
            dic = m.dict({"count": 100})
            p_1 = []
            for i in range(50):
                p = Process(target=work, args=(dic, lock))
                p_1.append(p)
                p.start()
            for p in p_1:
                p.join()
            print(dic)
    总结一下: 进程之间的通信: 队列, 管道, 数据共享也算
    下面要讲的信号量和事件也相当于锁，也是全局的，所有进程都能拿到这些锁的状态，进程之间这些锁啊信号量啊事件啊等等的通信，其实底层还是socekt，只不过是基于文件的socket通信，而不是跟上面的数据共享啊空间共享啊之类的机制，我们之前学的是基于网络的socket通信，还记得socket的两个家族吗，一个文件的一个网络的，所以将来如果说这些锁之类的报错，可能你看到的就是类似于socket的错误，简单知道一下就可以啦~~~
    工作中常用的是锁，信号量和事件不常用，但是信号量和事件面试的时候会问到，你能知道就行啦~~~
    8. 信号量
    互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。
    实现：
    信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。
    信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念
    from multiprocessing import Process, Semaphore
    import time, random
    def go_ktv(sem, user):
        sem.acquire()
        print("%s 占到一间ktv小屋"% user)
        time.sleep(random.randint(1, 3))    # 模拟每个人在ktv中呆的时间不长
        sem.release()
    if __name__ == "__main__":
        sem = Semaphore(4)
        p_l = []
        for i in range(13):
            p = Process(target=go_ktv, args=(sem, "user%s"%i,))
            p.start()
            p_l.append(p)
        for i in p_l:
            i.join()
        print("主进程结束")
    9. 事件
    python线程的事件用于主线程控制其他线程的执行, 事件主要提供了三个方法: set, wait, clear
    事件处理的机制: 全局定义了一个flag, 如果flag值为false, 那么当程序执行event.wait方法时,
    就会阻塞, 如果flag值为true, 那么event.wait方法就不会阻塞
    clear: 将flag设置为flase
    set: 将flag设置为true
二. 进程池
    首先，创建进程需要消耗时间，销毁进程（空间，变量，文件信息等等的内容）也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，维护一个很大的进程列表的同时，调度的时候，还需要进行切换并且记录每个进程的执行节点，也就是记录上下文（各种变量等等乱七八糟的东西，虽然你看不到，但是操作系统都要做），这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。
    在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果
    1. multiprocess pool 模块
    创建进程池的类:如果指定numprocess为3, 则进程池会从无到有创建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(): 关闭进程池, 组织进一步操作, 如果所有操作持续挂起, 他们将在工作进程终止前完成
    pjoin(): 等待所有工作进程退出, 此方法只能在close()或terminate()之后调用
    方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法:
    obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发异常。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
    obj.ready():如果调用完成，返回True
    obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
    obj.wait([timeout]):等待结果变为可用。
    obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
    2. 进程池的简单应用及与多线程的效率对比
    import time
    from multiprocessing import Process, Pool
    def f1(n):
        # time.sleep(2)
        for i in range(10):
            n += i
    if __name__ == '__main__':
        # 多进程
        s = time.time()
        p_list = []
        for i in range(100):
            p = Process(target=f1, args=(i,))
            p_list.append(p)
            p.start()
        [pp.join() for pp in p_list]
        e = time.time()
        print("多进程用时: %s" % (e - s))
        # 进程池
        s = time.time()
        pool = Pool(4)
        pool.map(f1, range(100))
        e = time.time()
        print("进程池用时: %s" % (e-s))
    # 多进程用时: 12.567207336425781
    # 进程池用时: 0.6605911254882812
    有一点, map是异步执行的, 并且自带close和join
    一般约定俗称的是进程池中的进程数量为cpu的数量, 工作中要看具体情况来考量
    3. 同步和异步两种执行方式:
    (1). 进程的同步调用
    import os, time
    from multiprocessing import Pool
    def work(n):
        print("%s run" % os.getpid())
        time.sleep(1)
        return n**2
    if __name__ == '__main__':
        # 进程池中从无到有创建三个进程, 以后一直是这三个进程在执行任务
        s = time.time()
        p = Pool(4)
        res_l = []
        for i in range(10):
            # 同步调用, 直到本次任务执行完毕拿到res, 等待任务work执行的过程中可能有阻塞也可能没有阻塞, 但不管该任务是否存在阻塞, 同步调用都会在原地等着
            res = p.apply(work, args=(i,))
            res_l.append(res)
        print(res_l)
        e = time.time()
        print("用时:", e-s)
    (2). 进程的异步调用
    import os, time, random
    from multiprocessing import Pool
    def work(n):
        print("%s run" % os.getpid())
        # time.sleep(random.randint(1, 3))
        time.sleep(1)
        return n**2
    if __name__ == '__main__':
        s = time.time()
        p = Pool(4)
        res_l =[]
        for i in range(10):
            res = p.apply_async(work, args=(i,))
            res_l.append(res)
        # 结束进程池接收任务, 确保没有新任务再提交过来
        p.close()
        # 感知进程池中的任务已经执行结束, 只有当没有新任务添加进来的时候, 才能感知到任务结束了, 所以在join之前必须加上close方法
        p.join()
        for res in res_l:
            print(res.get())
            # 使用get来获取apply_async的结果, 如果是apply, 则没有get方法, 因为apply是同步执行, 立刻获取到结果, 也根本无需get
        e = time.time()
        print("用时:", e-s)
    # 异步运行, 根据进程池中有的进程数, 每次最多4个子进程在异步运行, 并且可以执行不同的任务, 传送任意的参数了,
    # 返回结果后, 将结果放入列表, 归还进程, 之后再执行新的任务
    # 需要注意的是, 进程池中的4个进程不会同时开启或者同时结束,
    # 而是执行完一个就释放一个进程, 这个进程就去接收新的任务
    # 异步apply_async用法: 如果使用异步提交的任务, 主进程需要使用join, 等待进程池内任务都处理完, 然后可以用get收集结果, 否则, 主进程结束, 进程池可能还没来得及执行, 也就跟着一起结束了,
    (3). 详解apply_async和apply
    #一：使用进程池（异步调用,apply_async）
    #coding: utf-8
    from multiprocessing import Process,Pool
    import time
    def func(msg):
        print( "msg:", msg)
        time.sleep(1)
        return msg
    if __name__ == "__main__":
        pool = Pool(processes = 3)
        res_l=[]
        for i in range(10):
            msg = "hello %d" %(i)
            res=pool.apply_async(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
            res_l.append(res)
            # s = res.get() #如果直接用res这个结果对象调用get方法获取结果的话，这个程序就变成了同步，因为get方法直接就在这里等着你创建的进程的结果，第一个进程创建了，并且去执行了，那么get就会等着第一个进程的结果，没有结果就一直等着，那么主进程的for循环是无法继续的，所以你会发现变成了同步的效果
        print("==============================>") #没有后面的join，或get，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了
        pool.close() #关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
        pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
        print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果
        for i in res_l:
            print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
    #二：使用进程池（同步调用,apply）
    #coding: utf-8
    from multiprocessing import Process,Pool
    import time
    
    def func(msg):
    print( "msg:", msg)
    time.sleep(0.1)
    return msg
    if __name__ == "__main__":
        pool = Pool(processes = 3)
        res_l=[]
        for i in range(10):
            msg = "hello %d" %(i)
            res=pool.apply(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
            res_l.append(res) #同步执行，即执行完一个拿到结果，再去执行另外一个
        print("==============================>")
        pool.close()
        pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
    
        print(res_l) #看到的就是最终的结果组成的列表
        for i in res_l: #apply是同步的，所以直接得到结果，没有get()方法
            print(i)
    (4). 进程池版的socket并发聊天代码示例
    (5). 回调函数
    需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数，这是进程池特有的，普通进程没有这个机制，但是我们也可以通过进程通信来拿到返回值，进程池的这个回调也是进程通信的机制完成的。
    我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果
    import os, time
    from multiprocessing import Pool
    
    def func1(n):
        print("func1>>>", os.getpid())
        print("func1")
        return n*n
    
    def func2(nn):
        print("func2>>>", os.getpid())
        print("func2")
        # print(nn)
        # time.sleep(0.5)
    
    if __name__ == '__main__':
        print("主进程:", os.getpid())
        p = Pool(4)
        # args里面的10给了func1, func1的返回值最为回调函数的参数给了callback对应的函数, 不是直接给回调函数直接传递参数, 只能是任务函数func1的返回值
        # p.apply_async(func1, args=(10,), callback=func2)
    
        # 如果是多个进程来执行任务，那么当所有子进程将结果给了回调函数之后，回调函数又是在主进程上执行的，那么就会出现打印结果是同步的效果。我们上面func2里面注销的时间模块打开看看
        for i in range(10, 20):
            p.apply_async(func1, args=(i,), callback=func2)
        p.close()
        p.join()
    
    # 主进程: 9312     # 回调函数是在主进程中完成的
    # func1>>> 9888
    # func1
    # func2>>> 9312
    # func2
    # 100
    回调函数在写的时候注意一点，回调函数的形参执行有一个，如果你的执行函数有多个返回值，那么也可以被回调函数的这一个形参接收，接收的是一个元祖，包含着你执行函数的所有返回值。