Python爬虫 multiprocessing库应用详解_multiprocessing 爬虫

Python爬虫（十）

学习Python爬虫过程中的心得体会以及知识点的整理，方便我自己查找，也希望可以和大家一起交流。

—— multiprocessing库应用详解 ——

multiprocessing库对应的是进程。
进程和线程的问题点击 详细教程查看。

1. Process模块

Process模块用来创建子进程，可以实现多进程的创建，启动，关闭等操作。

1.1 构造方法

Process([group [, target [, name [, args [, kwargs]]]]])

p = multiprocessing.Process(target = worker, args = (3,))
1

• group: 线程组，目前还没有实现，库引用中提示必须是None；
• target: 要执行的方法；
• name: 进程名；
• args/kwargs: 要传入方法的参数。

1.2 其他方法

• is_alive()：返回进程是否在运行。

  print("p.is_alive:", p.is_alive())
  1

• start()：进程准备运行，等待调度。

  p.start()
  1

• run()：如果实例进程时未制定传入target，start()执行默认run()方法。

• terminate()：不管任务是否完成，立即停止工作进程。

• daemon：将父进程设置为守护进程，当父进程结束时，子进程也结束。

  p.daemon = True
  1

  具体使用方法见multiprocessing库实践。

• name：进程名字。

  print("p.name:", p.name)
  1

• pid：进程号。

  print("p.pid:", p.pid)
  1

※最简单的使用进程的方式：

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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

结果如图：

2. Pool模块

Pool模块是用来创建管理进程池的，当子进程非常多且需要控制子进程数量时可以使用此模块。

2.1 构造方法

Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

pool = Pool(processes=10)
1

• processes ：使用的工作进程的数量，如果processes是None那么使用 os.cpu_count()返回的数量。

• initializer： 如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。

• maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。

• context: 用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context。

※最简单的使用进程池的方式：

在windows系统下使用Python3的IDLE进行编译会出现子进程进行不了的问题，针对这个问题官方回复是：

Well, IDLE is a strange thing. In order to “capture” everything what you write using print statements orsys.stdout.write, IDLE “overrides” sys.stdout and replaces it with an object that passes everything back to IDLE so it can print it. I guess when you are starting a new process from multiprocessing, this hackery is not inherited by the child process, therefore you don’t see anything in IDLE. But I’m just guessing here, I don’t have a Windows machine at the moment to check it. – Tamás May 6 '10 at 9:10

也就是说由于Windows安全机制以及IDLE设计的问题，这个没办法搞定，只能在命令行模式下运行正常。所以这个部分我们将就在命令模式下运行。

非阻塞方式：

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 range(8):
        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.")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

结果如图：

阻塞方式：

在系统任务边界比较容易划分，任务比较容易切割成多段子任务的情况下，非阻塞模式往往是最佳选择。因为任务明确， 不需要子进程和主进程之间反复频繁的通信 ，各子进程只需要完成相关的计算，然后返回给主进程一个简单的结果就行了。例如上面的例子就属于这种情况。

可如果系统任务比较复杂，计算边界难以把握，任务难以划分的情况下，非阻塞模式并不是最佳选择，阻塞模式往往性能更好。例如：有一个计算某文件夹下所有文件的大小之和的程序，该程序需要遍历文件夹下的所有子文件和子目录，你无法预计该文件夹下有多深的目录层次，因此也无法具体划分子任务。

2.2 其他方法

• apply_async(func[, args[, kwds[, callback]]]) ：非阻塞的。

• apply(func[, args[, kwds]])：阻塞的。

• close()：关闭pool，使其不在接受新的任务。

• terminate()：关闭pool，结束工作进程，不在处理未完成的任务。

3. Queue模块

Queue模块用来控制进程安全。可以使用Queue实现多进程之间的数据传递。

put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。
如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。
如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。

get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。
如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。
如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常。

示例代码：

import multiprocessing
import time

def writer_proc(q):
    try:
        q.put(1, block = False)
    except:
        pass

def reader_proc(q):
    time.sleep(3)
    try:
        print (q.get(block = False))
    except:
        pass

if __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()
    print("aaa")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

结果如图：

4. Pipe模块

Pipe模块用来管道操作。返回(conn1, conn2)代表一个管道的两个端。Pipe方法有duplex参数。

如果duplex参数为True(默认值)，那么这个管道是全双工模式，也就是说conn1和conn2均可收发。

duplex为False，conn1只负责接受消息，conn2只负责发送消息。send和recv方法分别是发送和接受消息的方法。

例如，在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError。