摘要: 简介 进程是运行的程序,每个进程有自己的系统状态,包含了内存、打开文件列表、程序计数器(跟踪执行的指令)、存储函数本地调用变量的堆栈。 使用os或subprocess可以创建新进程,比如:os.fork(), subprocess.Popen()。
简介
进程是运行的程序,每个进程有自己的系统状态,包含了内存、打开文件列表、程序计数器(跟踪执行的指令)、存储函数本地调用变量的堆栈。
使用os或subprocess可以创建新进程,比如:os.fork(), subprocess.Popen()。子进程和父进程是相互独立执行的。
interprocess communication (IPC)进程间的通信: 最常见的形式是基于消息传递(message passing)。message是原始字节的缓存,通过I/O channel比如网络socket和管道,使用原语比如send() and recv()来发送接收消息。次常用的有内存映射区:memory-mapped regions,见mmap模块,实际上是共享内存。
线程有自己的控制流和执行堆栈,但是共享系统资源和数据。
并发的难点:同步和数据共享。解决的方法一般是使用互斥锁。
python的并发程序设计
多数系统上,Python支持消息传递和基于线程的并发程序设计。global interpreter lock (the GIL)机制实际每个时间单元只允许单个线程执行,哪怕有多个CPU。如果瓶颈在I/O,使用多线程效果不错;如果在cpu,效果则会更差。还不如使用子进程和消息传递。线程数一多经常出现以下怪异的问题,比如100个线程工作良好,1000个线程就可能出问题了,这种情况一般需要使用异步事件处理系统,比如中央事件循环可能使用select模块监控I/O资源和分发异步到大量的I/O 处理器。asyncore和流行的第三方的Twisted (http://twistedmatrix/com)可以实现这点。
消息传递在python使用很广,甚至在线程中。它难于出错,减少了锁和同步原语的使用。可以扩展至网络和分布式系统。Python的高级特性比如协程序(coroutines)也使用消息传递抽象。
multiprocessing支持子进程、通信和共享数据、执行不同形式的同步。
multiprocessing
Process类
这个类表示子进程中运行的任务:Process([group [, target [, name [, args [, kwargs]]]]]),构造函数中必须使用关键字参数,target表示可调用对象,args表示调用对象的位置参数元组。kwargs表示调用对象的字典。Name为别名。Group实质上不使用。
方法有:is_alive()、.join([timeout])、run()、start()、terminate()。
属性有:authkey、daemon(要通过start()设置)、exitcode(进程在运行时为None、如果为–N,表示被信号N结束)、name、pid。
Process类中,注意daemon是父进程终止后自动终止,且自己不能产生新进程,必须在start()之前设置。
创建函数并将其作为单个进程。
将进程定义为类:
注意,要在命令行才能执行,用IDE是不行的。
进程通信
multiprocessing支持管道和队列,都是用消息传递来实现的,队列接口和线程中的队列类似。
Queue([maxsize]):默认不限制大小,队列实质是用管道和锁来实现的。支持线程会给底层管道传送数据。
方法有:cancel_join_thread()、close()、empty()、full()、get([block [, timeout]])、get_nowait()(等同于get(False))、join_thread()、put(item [, block [, timeout]])、put_nowait(item)(等同于put(item, False))、qsize()、JoinableQueue([maxsize])、task_done()、join()
下例使用队列进行通信:
JoinableQueue创建连接的进程队列。队列和普通队列基本一样,不过消费者在处理完毕之后可以通知生产者(q.task_done())。使用共享信号和条件变量实现。join()由生产者使用,等待所有成员都收到task_done。
这里控制多进程的关键在于队列get()之后,使用task_done()指示该元素处理完毕;进程启动之前设置了daemon为True;对队列使用join()。
这种方法可以启动多个进程,如下:
下面有个应用实例(复制链接到浏览器哦):
https://bitbucket.org/china-testing/small_python_daily_tools/src/87d81739633482abdd3a2d0d11f62f6edd989555/db/mysql/check_transfer.py?at=default&fileviewer=file-view-default
在某些程序中,生产者需要告知消费者没有更多项目了,消费者可以关闭了。这时需要使用哨兵(sentinel)。
注意:每个消费者都需要一个:sentinel,可以使用for语句来实现
实际使用中不局限于使用None,使用其他特殊符号等也是可以的。上面程序从表面看比使用JoinableQueue要复杂,实现的效果又是一样的。实际上这种场景应用更广泛,在consumer比较耗时的情况下,JoinableQueue如果锁住整个函数则互相等待的时间太长,如果不锁,后面几次执行可能丢失数据。
管道
使用管道:Pipe([duplex]),返回值:元组(conn1, conn2)。conn1和conn2为Connection对象,代表管道的末端。管道默认是双向的,如果设置duplex为False,conn1只能接收,conn2只能发送。
Connection对象的方法和属性如下:
close()、fileno()、poll([timeout])、recv()、recv_bytes([maxlength])、recv_bytes_into(buffer [, offset])、send(obj)、send_bytes(buffer [, offset [, size]])
下面例子实现和之前类似的功能:
管道还可以用于双向通信,比如下例的C/S模式:
send()和recv()使用pickle序列化对象。更高级的程序需要使用远程过程调用,需要使用到进程池。
进程池
Pool类在简单的情况下可用于管理固定数量的消费者。进程池的功能和列表解析及函数式编程中的map-reduce类似。
执行结果:
$ python3 multiprocessing_pool.py
Built-in: [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98, 100, 102, 104, 106, 108, 110, 112, 114, 116, 118, 120, 122, 124, 126, 128, 130, 132, 134, 136, 138, 140, 142, 144, 146, 148, 150, 152, 154, 156, 158, 160, 162, 164, 166, 168, 170, 172, 174, 176, 178, 180, 182, 184, 186, 188, 190, 192, 194, 196, 198]
3.790855407714844e-05
Starting ForkPoolWorker-1
Starting ForkPoolWorker-2
Starting ForkPoolWorker-3
Starting ForkPoolWorker-4
Starting ForkPoolWorker-5
Starting ForkPoolWorker-6
Starting ForkPoolWorker-7
Starting ForkPoolWorker-8
Starting ForkPoolWorker-9
Starting ForkPoolWorker-10
Starting ForkPoolWorker-11
Starting ForkPoolWorker-12
Starting ForkPoolWorker-13
Starting ForkPoolWorker-14
Starting ForkPoolWorker-15
Starting ForkPoolWorker-16
Pool : [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98, 100, 102, 104, 106, 108, 110, 112, 114, 116, 118, 120, 122, 124, 126, 128, 130, 132, 134, 136, 138, 140, 142, 144, 146, 148, 150, 152, 154, 156, 158, 160, 162, 164, 166, 168, 170, 172, 174, 176, 178, 180, 182, 184, 186, 188, 190, 192, 194, 196, 198]
0.2203056812286377
上面例子先计算map的时间,然后用进程池的map,计算出时间。在列表数比较少的情况下,多进程的执行时间更短。列表数比较多的情况下,多进程的执行时间更长,可见python内置的map是效率比较高的。
如果消费者函数有内存泄露,可以在执行任务之后重启,设定maxtasksperchild参数即可。
执行结果:
$ python3 multiprocessing_pool2.py
Built-in: [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98, 100, 102, 104, 106, 108, 110, 112, 114, 116, 118, 120, 122, 124, 126, 128, 130, 132, 134, 136, 138, 140, 142, 144, 146, 148, 150, 152, 154, 156, 158, 160, 162, 164, 166, 168, 170, 172, 174, 176, 178, 180, 182, 184, 186, 188, 190, 192, 194, 196, 198]
3.600120544433594e-05
Starting ForkPoolWorker-1
Starting ForkPoolWorker-3
Starting ForkPoolWorker-2
Starting ForkPoolWorker-4
Starting ForkPoolWorker-5
Starting ForkPoolWorker-6
Starting ForkPoolWorker-7
Starting ForkPoolWorker-8
Starting ForkPoolWorker-9
Starting ForkPoolWorker-10
Starting ForkPoolWorker-11
Starting ForkPoolWorker-12
Starting ForkPoolWorker-13
Starting ForkPoolWorker-14
Starting ForkPoolWorker-15
Starting ForkPoolWorker-16
Starting ForkPoolWorker-17
Starting ForkPoolWorker-18
Starting ForkPoolWorker-19
Starting ForkPoolWorker-20
Starting ForkPoolWorker-21
Starting ForkPoolWorker-22
Starting ForkPoolWorker-23
Starting ForkPoolWorker-24
Starting ForkPoolWorker-25
Starting ForkPoolWorker-26
Starting ForkPoolWorker-27
Starting ForkPoolWorker-28
Starting ForkPoolWorker-29
Starting ForkPoolWorker-30
Starting ForkPoolWorker-31
Starting ForkPoolWorker-32
Pool : [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98, 100, 102, 104, 106, 108, 110, 112, 114, 116, 118, 120, 122, 124, 126, 128, 130, 132, 134, 136, 138, 140, 142, 144, 146, 148, 150, 152, 154, 156, 158, 160, 162, 164, 166, 168, 170, 172, 174, 176, 178, 180, 182, 184, 186, 188, 190, 192, 194, 196, 198]
0.23842501640319824
从结果看,进程数有所增加。(注意,进程数似乎比预期的要少)
Pool([numprocess [,initializer [, initargs]]])
numprocess的默认值是cpu_count()。方法有:apply(func [, args [, kwargs]]),apply_async(func [, args [, kwargs [, callback]]]),close(),join(),imap(func, iterable [, chunksize]),imap_unordered(func, iterable [, chunksize]]),map(func, iterable [, chunksize]),map_async(func, iterable [, chunksize [, callback]]),terminate().
返回结果AsyncResult的方法:get([timeout])、ready()、sucessful()、wait([timeout])、wait([timeout])
以下代码生成指定目录的文件名和SHA512对应表的字典。
共享数据和同步
共享内存通过mmap实现。共享内存中创建的是ctypes对象,不需要管道中的序列化。
Value(typecode, arg1, ... argN, lock),RawValue(typecode, arg1, ..., argN),Array(typecode, initializer, lock),RawArray(typecode, initializer)
原语有: Lock,Rlock,Semaphore,BoundedSemaphore,Event,Condition.