多任务——进程

多任务

多任务：操作系统可以同时运行多个任务

并行：任务数小于cpu核数，任务一起执行

并发：任务数多于cpu核数，通过操作系统的各种任务调度算法，实现多个任务一起执行

 

进程

进程：操作系统分配资源和调度的单位，是一个执行中的程序，是资源分配最小单位

进程的创建：multiprocessing模块就是跨平台版本的多进程模块，提供了一个Process类来创建进程对象

from multiprocessing import Process

进程pid

import multiprocessing
import os
import time


def run_proc():
    # os.getpid() 取得当前进程的id
    # os.getppid() :取得当前进程的父进程的id
    while True:
        time.sleep(1)
        print("--run_proc--子进程的父进程的id=%d   子进程的id= %d" % (os.getppid(), os.getpid()))


if __name__ == '__main__':
    print("--main--主进程的父进程id=%d   主进程的id= %d" % (os.getppid(), os.getpid()))

    # 在主进程中创建一个子进程p1   ，p1的父进程就是主进程
    p1 = multiprocessing.Process(target=run_proc)
    p1.start()

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（进程号）

q=multiprocessing.Queue(maxsize=3)
q.maxsize:设置队列的最大值
q.put(xxx):向队列存放数据
q.get():向队列获取数据
q.full()：判断队列是否满了
q.empty()：判断队列是否为空

进程池Pool

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

from multiprocessing import Pool
import os, time, random

def worker(msg):
    t_start = time.time()
    print("%s开始执行,进程号为%d" % (msg,os.getpid()))
    # random.random()随机生成0~1之间的浮点数
    time.sleep(random.random()*2) 
    t_stop = time.time()
    print(msg,"执行完毕，耗时%0.2f" % (t_stop-t_start))

po = Pool(3)  # 定义一个进程池，最大进程数3
for i in range(0,10):
    # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))
    # 每次循环将会用空闲出来的子进程去调用目标
    po.apply_async(worker,(i,))

print("----start----")
po.close()  # 关闭进程池，关闭后po不再接收新的请求
po.join()  # 等待po中所有子进程执行完成，必须放在close语句之后
print("-----end-----")

multiprocessing.Pool常用函数解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
• close()：关闭Pool，使其不再接受新的任务；
• terminate()：不管任务是否完成，立即终止；
• join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

进程池中的Queue

如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否则会得到一条如下的错误信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

进程池中的进程之间的通信，代码如下：

# 修改import中的Queue为Manager
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()  # 使用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())

进程池：节省内存空间，节省创建多个进程消耗的时间
解释：

pool=multiprocessing.Pool()  # 进程池默认最大为cpu的核数

pool.apply_async(func=子进程执行的函数,args=向func指向函数传递的可变参数,kwds=传递的关键字参数,callback=要回调的函数,)

关闭进程池pool.close（）

终止进程池中的子进程的任务执行：pool.terminate()

阻塞主进程，等进程池中的所有子进程执行完毕pool.join（）

进程锁Lock

使用进程锁购查票代码如下：

import json
import time
from multiprocessing import Process, Lock


# 查票
def show(i):
    with open('ticket')as f:
        dic = json.load(f)
    print('余票:%s' % dic['ticket'])


def buy_titck(i, lock):
    lock.acquire()  # 拿到钥匙
    with open('ticket')as f:
        dic = json.load(f)
        # 网络延迟
        time.sleep(0.1)
    if dic['ticket'] > 0:
        dic['ticket'] -= 1
        print('33[32m%s买到票了33[0m' % i)
    else:
        print('33[31m%s没买到票33[0m' % i)

    time.sleep(0.1)
    with open('ticket', 'w')as f:
        json.dump(dic, f)
    lock.release()  # 还钥匙


if __name__ == '__main__':
    # 多进程，开10个进程
    # 查票
    for i in range(10):
        p = Process(target=show, args=(i,))
        p.start()
    # 进程锁
    lock = Lock()

    # 买票
    for i in range(10):
        p = Process(target=buy_titck, args=(i, lock))
        p.start()