多任务——进程

多任务
多任务:操作系统可以同时运行多个任务
并行:任务数小于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()



原文地址:https://www.cnblogs.com/xiaolu915/p/10516974.html