进程

multiprocess

Process —— 进程 在python中创建一个进程的模块

    start

    daemon 守护进程

    join 等待子进程执行结束

锁 Lock

acquire release

锁是一个同步控制的工具

如果同一时刻有多个进程同时执行一段代码，

那么在内存中的数据是不会发生冲突的

但是，如果涉及到文件，数据库就会发生资源冲突的问题

我们就需要用锁来把这段代码锁起来

任意一个进程执行了acquire之后，

其他所有的进程都会在这里阻塞，等待一个release

信号量 semaphore

锁 + 计数器

同一时间只能有指定个数的进程执行同一段代码

事件 Event

set clear is_set   控制对象的状态

wait  根据状态不同执行效果也不同

    状态是True ---> pass

    状态是False --> 阻塞

一般wait是和set clear放在不同的进程中

set/clear负责控制状态

wait负责感知状态

我可以在一个进程中控制另外一个或多个进程的运行情况

IPC通信

队列 Queue

管道 PIPE

from multiprocessing import Process,Queue

q = Queue()

print(q.empty())

from multiprocessing import Process,Queue

q = Queue()

print(q.full())

from multiprocessing import Process,Queue

q = Queue(10)  # 创建一个只能放10个value的队列

for i in range(10):

    q.put(i)  # 增加一个value

print(q.qsize())  # 返回队列中目前项目的正确数量

print(q.full())  # 如果q已满，返回为True

q.put(111)  # 再增加一个值

print(q.empty())

import time

from multiprocessing import Process,Queue

def wahaha(q):

    print(q.get())

    q.put(2)  # 增加数字2

if __name__ == '__main__':

    q = Queue()

    p = Process(target=wahaha,args=[q,])

    p.start()

    q.put(1)  # 增加数字1

    time.sleep(0.1)

    print(q.get())

from multiprocessing import Process,Queue

def producer(q,name,food):

    for i in range(5):

        print('{}生产了{}{}'.format(name,food,i))

if __name__ == '__main__':

    q = Queue()

    Process(target=producer,args=[q,'康师傅','红烧牛肉']).start()

    Process(target=producer,args=[q,'郑师傅','红烧鱼块']).start()

import time

import random

from multiprocessing import Process,Queue

def producer(q,name,food):

    for i in range(5):

        time.sleep(random.random())  # 模拟生产时间

        print('{}生产了{}{}'.format(name,food,i))

        q.put('{}{}'.format(food,i))  # 放入队列

def consumer(q,name):

    for i in range(10):

        food = q.get()  # 获取队列

        time.sleep(random.random())  # 模拟吃的时间

        print('{}吃了{}'.format(name,food))

if __name__ == '__main__':

    q = Queue()

    Process(target=producer,args=[q,'康师傅','红烧牛肉']).start()

    Process(target=producer,args=[q,'郑师傅','红烧鱼块']).start()

    Process(target=consumer,args=[q,'xiao']).start()

import time

import random

from multiprocessing import Process,Queue

def producer(q,name,food):

    for i in range(5):

        time.sleep(random.random())  # 模拟生产时间

        print('{}生产了{}{}'.format(name,food,i))

        q.put('{}{}'.format(food,i))  # 放入队列

def consumer(q,name):

    for i in range(5):  # 修改为5,因为有2个人

        food = q.get()  # 获取队列

        time.sleep(random.random())  # 模拟吃的时间

        print('{}吃了{}'.format(name,food))

if __name__ == '__main__':

    q = Queue()

    Process(target=producer,args=[q,'康师傅','红烧牛肉']).start()

    Process(target=producer,args=[q,'郑师傅','红烧鱼块']).start()

    Process(target=consumer,args=[q,'xiao']).start()

    Process(target=consumer, args=[q,'lin']).start()

import time

import random

from multiprocessing import Process,Queue

def producer(q,name,food):

    for i in range(5):

        time.sleep(random.random())  # 模拟生产时间

        print('{}生产了{}{}'.format(name,food,i))

        q.put('{}{}'.format(food,i))  # 放入队列

def consumer(q,name):

    while True:

        food = q.get()  # 获取队列

        if food == 'done':break  # 当获取的值为done时,结束循环

        time.sleep(random.random())  # 模拟吃的时间

        print('{}吃了{}'.format(name,food))

if __name__ == '__main__':

    q = Queue()  #创建队列对象，如果不提供maxsize，则队列数无限制

    p1 = Process(target=producer,args=[q,'康师傅','红烧牛肉'])

    p2 = Process(target=producer,args=[q,'郑师傅','红烧鱼块'])

    p1.start()  # 启动进程

    p2.start()

    Process(target=consumer,args=[q,'xiao']).start()

    Process(target=consumer, args=[q,'lin']).start()

    p1.join()  # 保证子进程结束后再向下执行

    p2.join()

    q.put('done')  # 向队列添加一个值done

    q.put('done')

JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：

q.task_done()

使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。

q.join()

生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task_done()方法为止。

下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

方法介绍

import time

import random

from multiprocessing import Process,JoinableQueue

def producer(q,name,food):

    for i in range(5):

        time.sleep(random.random())  # 模拟生产时间

        print('{}生产了{}{}'.format(name,food,i))

        q.put('{}{}'.format(food,i))

        q.join()  # 等到所有的数据都被task_done才结束

def consumer(q,name):

    while True:

        food = q.get()  # 获取队列

        time.sleep(random.random())  # 模拟吃的时间

        print('{}吃了{}'.format(name,food))

        q.task_done()  # 向q.join()发送一次信号,证明一个数据已经被取走了

if __name__ == '__main__':

    q = JoinableQueue()  #创建可连接的共享进程队列

    # 生产者们:即厨师们

    p1 = Process(target=producer,args=[q,'康师傅','红烧牛肉'])

    p2 = Process(target=producer,args=[q,'郑师傅','红烧鱼块'])

    p1.start()  # 启动进程

    p2.start()

    # 消费者们:即吃货们

    c1 = Process(target=consumer,args=[q,'xiao'])

    c2 = Process(target=consumer, args=[q,'lin'])

    c1.daemon = True  # 设置守护进程

    c2.daemon = True

    c1.start()  # 启动进程

    c2.start()

    p1.join()  # 保证子进程结束后再向下执行

    p2.join()

producer

    put

    生产完全部的数据就没有其他工作了

    在生产数据方 ： 允许执行q.join

    join会发起一个阻塞，直到所有当前队列中的数据都被消费

consumer

    get 获取到数据

    处理数据

    q.task_done()  告诉q，刚刚从q获取的数据已经处理完了

consumer每完成一个任务就会给q发送一个taskdone

producer在所有的数据都生产完之后会执行q.join()

producer会等待consumer消费完数据才结束

主进程中对producer进程进行join

主进程中的代码会等待producer执行完才结束

producer结束就意味着主进程代码的结束

consumer作为守护进程结束

结束顺序：

consumer中queue中的所有数据被消费

producer join结束

主进程的代码结束

consumer结束

主进程结束

#创建管道的类：

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

def f(conn):

    conn.send("Hello The_Third_Wave")

    conn.close()

if __name__ == '__main__':

    parent_conn, child_conn = Pipe()

    p = Process(target=f, args=(child_conn,))

    p.start()

    print(parent_conn.recv())

    p.join()

pipe初使用

pipe初使用

from multiprocessing import Pipe

left,right = Pipe()

left.send('1234')

print(right.recv())

from multiprocessing import Process, Pipe

def f(parent_conn,child_conn):

    parent_conn.close() #不写close将不会引发EOFError

    while True:

        try:

            print(child_conn.recv())

        except EOFError:

            child_conn.close()

            break

if __name__ == '__main__':

    # 在进程之间创建一条管道，并返回元组（conn1,conn2），其中conn1和conn2是表示管道两端的Connection对象

    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.send('hello')

    parent_conn.send('hello')

    parent_conn.close()

    p.join() # 等待子进程结束

from multiprocessing import Process,Pipe

def consumer(p,name):

    produce, consume=p

    produce.close()

    while True:

        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('主进程')

pipe实现生产者消费者模型

from multiprocessing import Process,Pipe,Lock

def consumer(p,name,lock):

    produce, consume=p

    produce.close()

    while True:

        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('主进程')

多个消费之之间的竞争问题带来的数据不安全问题

进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的

虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.

Manager模块介绍

from multiprocessing import Manager,Process

def func(dic):

    print(dic)

if __name__ == '__main__':

    m = Manager()  # 创建一个server进程

    dic = m.dict({'count':100})  #这是一个特殊的字典

    p = Process(target=func,args=[dic,])

    p.start()

    p.join()

from multiprocessing import Manager,Process

def func(dic):

    dic['count'] = dic['count'] -1

    print(dic)

if __name__ == '__main__':

    m = Manager()  # 创建一个server进程

    dic = m.dict({'count':100})  #这是一个特殊的字典

    p = Process(target=func,args=[dic,])

    p.start()

    p.join()

from multiprocessing import Manager,Process

def func(dic):

    dic['count'] = dic['count'] -1  # 每次减1

if __name__ == '__main__':

    m = Manager()  # 创建一个server进程

    dic = m.dict({'count':100})  #这是一个特殊的字典

    p_lst = []  # 定义一个空列表

    for i in range(100):  # 启动100个进程

        p = Process(target=func,args=[dic,])

        p_lst.append(p)  # 进程追加到列表中

        p.start()  # 启动进程

    for p in p_lst:p.join()  # 等待100个进程全部结束

    print(dic)  # 打印dic的值

from multiprocessing import Manager,Process,Lock

def func(dic,lock):

    lock.acquire()  #取得锁

    dic['count'] = dic['count'] -1  # 每次减1

    lock.release()  #释放锁

if __name__ == '__main__':

    m = Manager()  # 创建一个server进程

    lock = Lock()  #创建锁

    dic = m.dict({'count':100})  #这是一个特殊的字典

    p_lst = []  # 定义一个空列表

    for i in range(100):  # 启动100个进程

        p = Process(target=func,args=[dic,lock])

        p_lst.append(p)  # 进程追加到列表中

        p.start()  # 启动进程

    for p in p_lst:p.join()  # 等待100个进程全部结束

    print(dic)  # 打印dic的值

from multiprocessing import Manager,Process,Lock

def func(dic,lock):

    with lock:  # 上下文管理 ：必须有一个开始动作 和 一个结束动作的时候

        dic['count'] = dic['count'] -1  # 每次减1

if __name__ == '__main__':

    m = Manager()  # 创建一个server进程

    lock = Lock()  #创建锁

    dic = m.dict({'count':100})  #这是一个特殊的字典

    p_lst = []  # 定义一个空列表

    for i in range(100):  # 启动100个进程

        p = Process(target=func,args=[dic,lock])

        p_lst.append(p)  # 进程追加到列表中

        p.start()  # 启动进程

    for p in p_lst:p.join()  # 等待100个进程全部结束

    print(dic)  # 打印dic的值

Pool([numprocess  [,initializer [, initargs]]]):创建进程池

1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值

2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None

3 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()之后调用

主要方法

方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法

obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。

obj.ready():如果调用完成，返回True

obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常

obj.wait([timeout]):等待结果变为可用。

obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数

其他方法(了解)

import time

from multiprocessing import Pool

def fc(i):

    time.sleep(0.5)

    print('func%s'%i)

if __name__ == '__main__':

    p = Pool(5)

    p.apply(func=fc,args=(1,))

import time

from multiprocessing import Pool

def fc(i):

    time.sleep(0.5)

    print('func%s'%i)

if __name__ == '__main__':

    p = Pool(5)

    for i in range(5):

        p.apply(func=fc,args=(1,))  # 同步调用

        #ret = p.apply_async(func=fc, args=(1,))  # 异步调用

import time

import random

from multiprocessing import Pool

def fc(i):

    print('func%s' % i)

    time.sleep(random.randint(1, 3))

    return i ** 2

if __name__ == '__main__':

    p = Pool(5) # 创建拥有5个进程数量的进程池

    ret_1 = []

    for i in range(5):

        #p.apply(func=fc,args=(1,))  # 同步调用

        ret = p.apply_async(func=fc, args=(i,))  # 异步调用

        ret_1.append(ret)

    for ret in ret_1:print(ret.get())  # 打印返回结果

import os,time

from multiprocessing import Pool

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执行的过程中可能有阻塞也可能没有阻塞

                                    # 但不管该任务是否存在阻塞，同步调用都会在原地等着

    print(res_l)

进程池的同步调用

import os

import time

import random

from multiprocessing import Pool

def work(n):

    print('%s run' %os.getpid())

    time.sleep(random.random())

    return n**2

if __name__ == '__main__':

    p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务

    res_l=[]

    for i in range(10):

        res=p.apply_async(work,args=(i,)) # 异步运行，根据进程池中有的进程数，每次最多3个子进程在异步执行

                                          # 返回结果之后，将结果放入列表，归还进程，之后再执行新的任务

                                          # 需要注意的是，进程池中的三个进程不会同时开启或者同时结束

                                          # 而是执行完一个就释放一个进程，这个进程就去接收新的任务。

        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

进程池的异步调用

#Pool内的进程数默认是cpu核数，假设为4（查看方法os.cpu_count()）

#开启6个客户端，会发现2个客户端处于等待状态

#在每个进程内查看pid，会发现pid使用为4个，即多个客户端公用4个进程

from socket import *

from multiprocessing import Pool

import os

server=socket(AF_INET,SOCK_STREAM)

server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)

server.bind(('127.0.0.1',8080))

server.listen(5)

def talk(conn):

    print('进程pid: %s' %os.getpid())

    while True:

        try:

            msg=conn.recv(1024)

            if not msg:break

            conn.send(msg.upper())

        except Exception:

            break

if __name__ == '__main__':

    p=Pool(4)

    while True:

        conn,*_=server.accept()

        p.apply_async(talk,args=(conn,))

        # p.apply(talk,args=(conn,client_addr)) #同步的话，则同一时间只有一个客户端能访问

server：进程池版socket并发聊天

from socket import *

client=socket(AF_INET,SOCK_STREAM)

client.connect(('127.0.0.1',8080))

while True:

    msg=input('>>: ').strip()

    if not msg:continue

    client.send(msg.encode('utf-8'))

    msg=client.recv(1024)

    print(msg.decode('utf-8'))

client

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。

import os

from urllib.request import urlopen

from multiprocessing import Pool

def get_url(url):

    print('-->',url,os.getpid(),'get_url进程')

    ret = urlopen(url)  # 打开url

    content = ret.read()  # 读取网页内容

    return content

def call(url):  # 回调函数

    #分析

    print(url,os.getpid(),'回调函数')

if __name__ == '__main__':

    print(os.getpid(),'主进程')  # 主进程id

    l = [

        'http://www.baidu.com',

        'http://www.sina.com',

        'http://www.sohu.com',

        'http://www.sogou.com',

        'http://www.qq.com',

        'http://www.bilibili.com',

    ]

    p = Pool(5)

    ret_l = []

    for url in l:

        ret = p.apply_async(func=get_url,args=[url,],callback=call)  # 异步

        ret_l.append(ret)  # 将进程追加到列表中

    for ret in ret_l:ret.get()  # 获取进程返回值

import os

from urllib.request import urlopen

from multiprocessing import Pool

def get_url(url):

    print('-->',url,os.getpid(),'get_url进程')

    ret = urlopen(url)  # 打开url

    content = ret.read()  # 读取网页内容

    return url

def call(url):  # 回调函数

    #分析

    print(url,os.getpid(),'回调函数')

if __name__ == '__main__':

    print(os.getpid(),'主进程')  # 主进程id

    l = [

        'http://www.baidu.com',

        'http://www.sina.com',

        'http://www.sohu.com',

        'http://www.sogou.com',

        'http://www.qq.com',

        'http://www.bilibili.com',

    ]

    p = Pool(5)

    ret_l = []

    for url in l:

        ret = p.apply_async(func=get_url,args=[url,],callback=call)  # 异步

        ret_l.append(ret)  # 将进程追加到列表中

    for ret in ret_l:ret.get()  # 获取进程返回值

from multiprocessing import Pool

import requests

import json

import os

def get_page(url):

    print('<进程%s> get %s' %(os.getpid(),url))

    respone=requests.get(url)

    if respone.status_code == 200:

        return {'url':url,'text':respone.text}

def pasrse_page(res):

    print('<进程%s> parse %s' %(os.getpid(),res['url']))

    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))

    with open('db.txt','a') as f:

        f.write(parse_res)

if __name__ == '__main__':

    urls=[

        'https://www.baidu.com',

        'https://www.python.org',

        'https://www.openstack.org',

        'https://help.github.com/',

        'http://www.sina.com.cn/'

    ]

    p=Pool(3)

    res_l=[]

    for url in urls:

        res=p.apply_async(get_page,args=(url,),callback=pasrse_page)

        res_l.append(res)

    p.close()

    p.join()

    print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了

'''

打印结果:

<进程3388> get https://www.baidu.com

<进程3389> get https://www.python.org

<进程3390> get https://www.openstack.org

<进程3388> get https://help.github.com/

<进程3387> parse https://www.baidu.com

<进程3389> get http://www.sina.com.cn/

<进程3387> parse https://www.python.org

<进程3387> parse https://help.github.com/

<进程3387> parse http://www.sina.com.cn/

<进程3387> parse https://www.openstack.org

[{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]

'''

使用多进程请求多个url来减少网络等待浪费的时间

import re

from urllib.request import urlopen

from multiprocessing import Pool

def get_page(url,pattern):

    response=urlopen(url).read().decode('utf-8')

    return pattern,response

def parse_page(info):

    pattern,page_content=info

    res=re.findall(pattern,page_content)

    for item in res:

        dic={

            'index':item[0].strip(),

            'title':item[1].strip(),

            'actor':item[2].strip(),

            'time':item[3].strip(),

        }

        print(dic)

if __name__ == '__main__':

    regex = r'<dd>.*?<.*?class="board-index.*?>(\d+)</i>.*?title="(.*?)".*?class="movie-item-info".*?<p class="star">(.*?)</p>.*?<p class="releasetime">(.*?)</p>'

    pattern1=re.compile(regex,re.S)

    url_dic={

        'http://maoyan.com/board/7':pattern1,

    }

    p=Pool()

    res_l=[]

    for url,pattern in url_dic.items():

        res=p.apply_async(get_page,args=(url,pattern),callback=parse_page)

        res_l.append(res)

    for i in res_l:

        i.get()

爬虫实例

from multiprocessing import Pool

import time,random,os

def work(n):

    time.sleep(1)

    return n**2

if __name__ == '__main__':

    p=Pool()

    res_l=[]

    for i in range(10):

        res=p.apply_async(work,args=(i,))

        res_l.append(res)

    p.close()

    p.join() #等待进程池中所有进程执行完毕

    nums=[]

    for res in res_l:

        nums.append(res.get()) #拿到所有结果

    print(nums) #主进程拿到所有的处理结果,可以在主进程中进行统一进行处理

无需回调函数

import time

import random

from multiprocessing import Process,Queue

def consumer(q,name):

    while True:

        food = q.get()

        if food == 'done':break

        time.sleep(random.random())

        print('%s吃了%s'%(name,food))

def producer(q,name,food):

    for i in range(10):

        time.sleep(random.random())

        print('%s生产了%s%s'%(name,food,i))

        q.put('%s%s'%(food,i))

if __name__ == '__main__':

    q = Queue()

    p1 = Process(target=producer,args=[q,'Egon','泔水'])

    p2 = Process(target=producer,args=[q,'Yuan','骨头鱼刺'])

    p1.start()

    p2.start()

    Process(target=consumer,args=[q,'alex']).start()

    Process(target=consumer,args=[q,'wusir']).start()

    p1.join()

    p2.join()

    q.put('done')

    q.put('done')