Python之进程

>什么是进程

在一个正在运行的程序中，代码和调度的资源称为进程，进程是操作系统分配资源的基本单元
之前有提到过，多任务不仅线程可以完成，进程也可以

>进程的状态

现实情况里，我们电脑的任务数通常是大于CPU核数，这样就会导致一些任务正在执行中，而一些任务在等待执行，这样就会导致不同的状态

image

就绪态：运行的条件已经满足，正在等待CPU执行

运行态：CPU正在执行

等待态：等待一些满足的条件，条件满足后进入就绪态

>进程的创建multiprocessing

multiprocessing模块提供了一个Process类来代表一个进程对象

-使用主进程创建一个子进程

# -*- coding:utf-8 -*-
import multiprocessing
import time

def run_proc():
    """子进程要执行的代码"""
    while True:
        print("这是子进程...1")
        time.sleep(2)

if __name__=='__main__':
    p = multiprocessing.Process(target=run_proc)
    p.start()
    while True:
        print("这是主进程...2")
        time.sleep(2)

执行结果

这是主进程...2
这是子进程...1
这是主进程...2
这是子进程...1
这是主进程...2
.
.

小结

• 进程的创建和执行和线程传入参数的方式有些类似，都是调用start()方法启动

>进程中的PID

# -*- coding:utf-8 -*-
from multiprocessing import Process
import os
import time

def run_proc():
    """子进程要执行的代码"""
    print('子进程运行中，pid=%d...' % os.getpid())  # os.getpid获取当前进程的进程号
    print('子进程将要结束...')

if __name__ == '__main__':
    print('父进程pid: %d' % os.getpid())  # os.getpid获取当前进程的进程号
    p = Process(target=run_proc)
    p.start()

执行结果

父进程pid: 15128
子进程运行中，pid=4868...
子进程将要结束...

>Process类的属性和方法

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

• target：如果传递了函数的引用，可以任务这个子进程就执行这里的代码
• args：给target指定的函数传递的参数，以元组的方式传递
• kwargs：给target指定的函数传递命名参数
• name：给进程设定一个名字，可以不设定
• group：指定进程组，大多数情况下用不到

Process创建的实例对象的常用方法

• start()：启动子进程实例（创建子进程）
• is_alive()：判断进程子进程是否还在活着
• join([timeout])：是否等待子进程执行结束，或等待多少秒
• terminate()：不管任务是否完成，立即终止子进程

Process创建的实例对象的常用属性

• name：当前进程的别名，默认为Process-N，N为从1开始递增的整数
• pid：当前进程的pid（进程号）

>给子进程指定的函数传递参数

# -*- coding:utf-8 -*-
from multiprocessing import Process
import os
from time import sleep


def run_proc(name, age, **kwargs):
    for i in range(10):
        print('子进程运行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
        print(kwargs)
        sleep(0.2)

if __name__=='__main__':
    p = Process(target=run_proc, args=('test',18), kwargs={"m":20})
    p.start()
    sleep(1)  # 1秒中之后，立即结束子进程
    p.terminate()
    p.join()

执行结果

子进程运行中，name= test,age=18 ,pid=14828...
{'m': 20}
子进程运行中，name= test,age=18 ,pid=14828...
{'m': 20}
子进程运行中，name= test,age=18 ,pid=14828...
{'m': 20}
子进程运行中，name= test,age=18 ,pid=14828...
{'m': 20}
子进程运行中，name= test,age=18 ,pid=14828...
{'m': 20}

>进程间不共享全局变量

import multiprocessing
import time
import os

nums = [11,22]

def sub_process1():
    print("its sub_process1,pid:%s;nums:%s" % (os.getpid(),nums))
    nums.append(33)
    time.sleep(1)
    print("its sub_process1,pid:%s;nums:%s" % (os.getpid(),nums))

def sub_process2():
    print("its sub_process2,pid:%s;nums:%s" % (os.getpid(),nums))

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=sub_process1)
    p1.start()
    p1.join()

    p2 = multiprocessing.Process(target=sub_process2)
    p2.start()

执行结果

its sub_process1,pid:5028;nums:[11, 22]
its sub_process1,pid:5028;nums:[11, 22, 33]
its sub_process2,pid:6400;nums:[11, 22]

>进程、线程

进程是系统进行资源分配和调度的一个独立单位

线程是进程的一个实体,是CPU调度和分配的基本单位,它是比进程更小的独立运行的基本单位.线程基本不拥有系统资源,只拥有一些必不可少的资源(程序计数器,寄存器,栈等),线程可与同属一个进程的其它线程共享进程所拥有的全部资源

二者区别

一个程序至少有一个进程,一个进程至少有一个线程

线程的划分尺度(资源)小于进程,使得线程的并发性高

进程在运行过程中独享内存,而线程是共享的,极大的提高了程序的运行效率

小结

线程的资源开销小,但是资源的管理和保护不如进程

>进程间通信Queue

Queue()实例化时,如果没有指定最大可接收消息的最大参数或者为负值,那么默认没有上限(直到内存的上限)

• Queue.qsize(): 返回当前队列包含消息的数量
• Queue.empty(): 判断队列是否为空,是为True,否为False
• Queue.full(): 判断队列是否满了,是为True,否为False
• Queue.get([block[,timeout]]): 获取一条队列的消息,然后将其从队列中删除,black默认为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(): 相当于Queue.put(block=False)

>Queue操作

from multiprocessing import Queue

q = Queue(3) #在实例化时传入消息队列的最大数量
q.put("test1")
q.put("test2")
print(q.full()) #这里队列未满,返回False
q.put("test3")
print(q.full()) #这里队列已经满了,返回True

try:
    q.put("test4", timeout=2) #在写入时,如果队列是满的,就等待两秒钟,如还不可以,抛出异常
except:
    print("当前消息队列的数量为%s" % q.qsize())

try:
    q.put("test4", block=False) # 在写入时,如果队列是满的,直接抛出异常
except:
    print("当前消息队列的数量为%s" % q.qsize())

# 在写入之前,可以判断下队列是否是满的
if not q.full():
    q.put_nowait("test4")

# 同理,在取数之前,判断队列是否为空
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

执行结果

False
True
当前消息队列的数量为3
当前消息队列的数量为3
test1
test2
test3

>两个进程分别读写Queue

from multiprocessing import Process,Queue
import time
import random

def write(q):
    nums = [11,22,33]
    for num in nums:
        if not q.full():
            print("write into num:%s" % num)
            q.put(num)
            time.sleep(random.random())
def read(q):
    while True:
        if not q.empty():
            num = q.get(True)
            print("read data num:%s" % num)
            time.sleep(random.random())
        else:
            break

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=write, args=(q,))
    p2 = Process(target=read, args=(q,))

    p1.start()
    p1.join()

    p2.start()
    p2.join()

    print("所有的数据都读写完毕了...")

执行结果

write into num:11
write into num:22
write into num:33
read data num:11
read data num:22
read data num:33
所有的数据都读写完毕了...

>multiprocess Pool(在linux环境可以正常执行,windows有异常)

为什么要引用进程池,当我们需要创建少量的进程时,可以手动的去创建,可是需要的创建的进程数量多时,就可以用到multiprocess中的Pool方法

我们在初始化Pool时,可以制定一个参数(最大的进程数),当有新的请求到Pool时,如果Pool池还没满,就会添加,如果满了,就会等到Pool池中有进程结束,才会用之前的进程来执行这个请求

multiprocess.Pool方法解析

• apply_async(func[,args[,kwargs]]): 使用非阻塞方式调用func(并行执行,阻塞的情况是等待上一个进程结束才能执行下一个请求),args为传入的参数,kwargs为关键字参数列表
• close(): 关闭Pool,使其不再结束新的任务
• terminate(): 不管任务是否完成,都立即退出
• join(): 主进程阻塞,等待子进程,在close()和terminate()后执行

# -*- coding:utf-8 -*-
from multiprocessing import Pool
import os, time, random

def work(msg):
    t_start = time.time()
    print("执行开始,进程的ID号为:%s" % os.getpid())
    time.sleep(random.random()*2)
    t_send = time.time()
    print("执行耗费时间为:%s" % (t_start - t_send))

po = Pool(3)

for i in range(0, 10):
    po.apply_async(work, (i,))

print("programmer starting....")
po.close()
po.join()
print("programmer ending...")

执行结果

programmer starting....
执行开始,进程的ID号为:24832
执行开始,进程的ID号为:24831
执行开始,进程的ID号为:24833
执行耗费时间为:-0.21535086631774902
执行开始,进程的ID号为:24832
执行耗费时间为:-1.3048121929168701
执行开始,进程的ID号为:24833
执行耗费时间为:-1.4840171337127686
执行开始,进程的ID号为:24831
执行耗费时间为:-1.6602394580841064
执行开始,进程的ID号为:24832
执行耗费时间为:-0.48267197608947754
执行开始,进程的ID号为:24831
执行耗费时间为:-0.277604341506958
执行开始,进程的ID号为:24831
执行耗费时间为:-0.2472696304321289
执行开始,进程的ID号为:24831
执行耗费时间为:-1.3967657089233398
执行耗费时间为:-0.2590181827545166
执行耗费时间为:-1.4253907203674316
programmer ending...

>进程池中的进程通信(Queue)

使用进程池Pool时,不能使用multiprocess.Queue,应该使用multiprocess.Manager()中的Queue
实战演示

# -*- coding:utf-8 -*-
from multiprocessing import Manager,Pool
import os,time,random

def reader(q):
    print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
    for i in range(q.qsize()):
        print("reader从Queue获取到消息：%s" % q.get(True))

def writer(q):
    print("writer启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))
    for i in "itcast":
        q.put(i)

if __name__=="__main__":
    print("(%s) start" % os.getpid())
    q = Manager().Queue()
    po = Pool()
    po.apply_async(writer, (q,))

    time.sleep(1)  # 先让上面的任务向Queue存入数据，然后再让下面的任务开始从中取数据

    po.apply_async(reader, (q,))
    po.close()
    po.join()
    print("(%s) End" % os.getpid())

执行结果

(7740) start
writer启动(7448),父进程为(7740)
reader启动(8096),父进程为(7740)
reader从Queue获取到消息：i
reader从Queue获取到消息：t
reader从Queue获取到消息：c
reader从Queue获取到消息：a
reader从Queue获取到消息：s
reader从Queue获取到消息：t
(7740) End