当前位置:   article > 正文

Python 进程、线程、协程(multiprocessing、threading)详解_python multiprocessing threading

python multiprocessing threading

在Python中多任务的实现方式为:
1、多进程
2、多线程
3、协程
4、多进程+多线程

一、多任务原理

多任务:操作系统可以同时完成多个任务
单核CPU多任务原理:单个cpu核心轮流的执行各个程序,但是切换的速度非常快导致表面上看是同时在执行
多核CPU多任务原理:真正的实现多任务,但是任务的数量会远远超过CPU核心的数量,因此操作系统会将各个任务轮流调度到各个核心上执行
并发:任务看上去一起执行,实际上任务书多余CPU核心数
并行:任务真正的一起执行,实际任务小于CPU核心数

二、进程、线程的概念

(1)进程:一个执行中的程序,操作系统管理其上所有进程的执行(比如QQ、微信等就相当于一个进程)

(2)线程:可称为轻量级进程;任何进程都会默认启动一个线程,即为主线程,主线程可以启动新的线程,称其为子线程。一个进程的各个线程与主线程共享一个数据空间,相比于独立的进程而言,线程间共享数据和通信更加容易(比如QQ的多个聊天窗口就相当于多个线程)

进程、线程优缺点比较

优点缺点
进程稳定性高:一个子进程崩溃了,不会影响主进程创建进程的开销大、操作系统运行的进程数有限
线程比多进程快,但是也快不到哪去任何线程的崩溃都会影响到整个进程

三、Python实现多进程 - - multiprocessing
multiprocessing 是一个用于产生进程的包,在 multiprocessing 中,通过创建一个 Process 对象然后调用它的 start() 方法来生成进程

示例一:用Process创建子进程

#父子进程的先后执行顺序
from  multiprocessing import Process
import time
def run():
    print('子进程启动')
    time.sleep(3)
    print('子进程结束')
if __name__ == '__main__':
    print('父进程启动')
    p = Process(target=run)
    #父进程结束不影响子进程
    p.start()
    #jion()方法让子进程结束在执行父进程,阻塞父进程,等待所有子进程执行完毕
    p.join()
    print('父进程结束')
"""
运行结果:
父进程启动
子进程启动
子进程结束
父进程结束
"""
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

示例二:通过进程池Pool创建大量子进程

from multiprocessing import Pool
import time,os,random


def runing(name):

    #os.getpid 可以获得进程号,打印子进程的名称和进程号
    print('子进程 %d 启动-%s'%(name,os.getpid()))
    #进程开始的时间
    start = time.time()
    #随机选择进程休息时间
    time.sleep(random.choice([1,2,3]))
    #进程结束时间
    end = time.time()
    #打印进程的名称、进程号和运行时间
    print('子进程 %d 结束-%s-耗时%.2f'%(name,os.getpid(),end-start))
if __name__ =='__main__':
    print('父进程启动')
    #创建多个进程
    #pool默认大小是cpu核心数量
    #创建两个进程,会同时运行两个进程
    pp = Pool(2)
    for i in range(8):
        
        #创建进程,放入进程池同时管理
        pp.apply_async(runing,args=(i,))
        
    #在调用join之前必须先调用close,调用close之后就不能在继续添加新的进程了
    pp.close()
    #等待子进程结束
    pp.join()
    print('父进程结束')
"""
运行结果:
父进程启动
子进程 0 启动-15996
子进程 1 启动-16016
子进程 0 结束-15996-耗时2.00
子进程 2 启动-15996
子进程 1 结束-16016-耗时2.00
子进程 3 启动-16016
子进程 2 结束-15996-耗时1.00
子进程 4 启动-15996
子进程 4 结束-15996-耗时1.00
子进程 5 启动-15996
子进程 3 结束-16016-耗时2.00
子进程 6 启动-16016
子进程 6 结束-16016-耗时2.00
子进程 7 启动-16016
子进程 5 结束-15996-耗时3.00
子进程 7 结束-16016-耗时2.00
父进程结束
"""
  • 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
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53

示例三:全局变量在多个进程中不能共享

from multiprocessing import Process
import  os
num = 100
def run():
    print('子进程启动')
    global num
    num += 1
    print(num)
    print('子进程结束')
if __name__ =='__main__':
    print('父进程启动')
    p = Process(target=run)
    p.start()
    #在子进程中修改全局变量对父进程中的全局变量没有影响
    #在创建子进程时对全局变量做了一个备份,父进程与子进程中的num是完全不同的两个变量
    p.join()
    print('父进程结束%s'%num)
"""
运行结果:
父进程启动
子进程启动
101
子进程结束
父进程结束100
"""
  • 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

示例四:进程间的通信

进程间的通信可以通过消息队列(queue)来实现
通过消息队列实现两个进程间通信,一个进程从终端输入(put)数据,通过消息队列发送,另一个进程通过消息队列接收(get)数据

from multiprocessing import  Process,Queue
import time,os

def write(q):
    print("启动写子进程%s"%(os.getpid()))
    for chr in ["A","B","C","D"]:
        #put方法将信息写入消息队列
        q.put(chr)
        time.sleep(1)
    print("结束子进程%s"%(os.getpid()))

def read(q):
    print("启动读子进程%s"%(os.getpid()))
    while True:
        #get方法将信息读取
        value = q.get(True)
        print("value = "+ value)
    print("结束读子进程%s"%(os.getpid()))

if __name__ =="__main__":
    #父进程建立队列,并传递给子进程
    #建立消息队列的对象实现进程间的通信
    q = Queue()
    #进程的创建
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    #进程的启动
    pw.start()
    pr.start()
    #等待子进程的执行
    pw.join()
    #pr进程是个死循环,无法等待其结束,只能强行结束
    pr.terminate()
    print("父进程结束")
"""
运行结果:
启动写子进程7668
启动读子进程15268
value = A
value = B
value = C
value = D
结束子进程7668
父进程结束
"""
  • 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
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45

四、Python实现多线程 - - threading
通过threading模块创建线程

示例一:启动一个线程
threading.Thread()会创建一个线程对象
current_thread() 方法返回当前对应调用者的控制线程的 Thread 对象。如果调用者的控制线程不是利用 threading 创建,会返回一个功能受限的虚拟线程对象。

mport threading,time

def run():
    print("子线程(%s)开始"%(threading.current_thread().name))
    #实现线程的功能
    time.sleep(2)
    print("打印")
    time.sleep(2)
    print("子线程(%s)结束"%(threading.current_thread().name))

if __name__ == "__main__":
    #任何进程默认就会启动一个线程,称为主线程,主线程可以启动新的线程
    #current_threed():返回当前线程实例
    print("主线程(%s)启动"%(threading.current_thread().name))
    #创建子线程
    t = threading.Thread(target=run,name = "rooThread")
    #子线程启动
    t.start()
    #等待线程结束
    t.join()
    print("主线程(%s)结束"%(threading.current_thread().name))
"""
运行结果:
主线程(MainThread)启动
子线程(rooThread)开始
打印
子线程(rooThread)结束
主线程(MainThread)结束
"""
  • 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
  • 28
  • 29

示例二:定时线程 - - threading.Time()
threading.Time() 会创建一个定时器对象,会等待到一定的时间再执行线程

import threading
def run():
    print("i love you")
#延时执行线程 5s
t = threading.Timer(5,run)
t.start()
t.join()
print("父进程结束")
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

示例三:信号量控制线程数量 - - threading.Semaphore( )
一个信号量管理一个内部计数器,该计数器因 acquire() 方法(获取一个信号量)的调用而递减,因 release() (释放一个信号量)方法的调用而递增。 计数器的值永远不会小于零;当 acquire() 方法发现计数器为零时,将会阻塞,直到其它线程调用 release() 方法。

import threading,time
sem = threading.Semaphore(2)
def run():
	#通过with自动的获取和释放信号量
    with sem:
        for i in range(10):
            print("%s-%d"%(threading.current_thread().name,i))
            time.sleep(1)

if __name__ =="__main__":
    for i in range(5):
        threading.Thread(target=run).start()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

示例四:线程间的通信 - - threading.Event
threading.Event 对象是线程之间通信的最简单机制之一:一个线程发出事件信号,而其他线程等待该信号
wait():阻塞线程直到内部变量为true。如果调用时内部标志为true,将立即返回。否则将阻塞线程,直到调用 set() 方法将标志设置为true或者发生可选的超时
set() :将内部标志设置为true。所有正在等待这个事件的线程将被唤醒。当标志为true时,调用 wait() 方法的线程不会被被阻塞。
clear()将内部标志设置为false。之后调用 wait() 方法的线程将会被阻塞,直到调用 set() 方法将内部标志再次设置为true。

import threading,time
def func():
    event = threading.Event()
    def run():
        for i in range(5):
            #阻塞,等待事件触发,即等待内部变量变为True
            event.wait()
            #重置,将内部变量设置为False
            event.clear()
            print("执行 %d 次"%i)
    t = threading.Thread(target=run).start()
    return event

e = func()
#触发事件
for i in range(5):
    time.sleep(2)
    #将内部变量设置为True
    e.set()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

示例五:线程间的数据共享 - - threading.Lock
threading.Lock 对象会阻塞随后尝试获得锁的线程,直到它被释放;任何线程都可以释放它。
acquire()可以阻塞或非阻塞地获得锁。
release()释放一个锁

import threading
#锁对象
lock = threading.Lock()
num = 0
def run(n):
    global num
    for i in range(100000):
        #确保这段代码只能由一个线程从头到尾的执行
        #阻止了多线程的并发执行,包含锁的某段代码只能以单线程执行,所以效率大大降低
        #由于可以存在多个锁,不同线程持有不同的锁,并试图获取其他的锁,可能造成死锁,导致多个线程挂起,只能操作系统强制结束
        try:
            #上锁,其他线程无法修改num的值,每次只允许一个线程对num的值进行修改
            lock.acquire()
            num -= n
            num += n
        #修改完一定要释放锁,其他线程才能修改num的值
        finally:
            lock.release()
  
if __name__ =="__main__":
    #创建另个子线程
    t1 = threading.Thread(target=run,args=(6,))
    t2 = threading.Thread(target=run,args=(9,))
    #子线程启动
    t1.start()
    t2.start()
    #等待子线程完成
    t1.join()
    t2.join()
    print("num:",num)
  • 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
  • 28
  • 29
  • 30

示例六:凑够一定数量再执行 (栅栏对象) - - threading.Barrier
threading.Barrie用于应对固定数量的线程需要彼此相互等待的情况;线程调用 wait() 方法后将阻塞,直到所有线程都调用了 wait() 方法。此时所有线程将被同时释放。

import threading,time
#阻塞的线程数量达到三个就执行
bar = threading.Barrier(3)

def run():
    print("%s-start"%(threading.current_thread().name))
    time.sleep(2)
    bar.wait()
    print("%s-end"%(threading.current_thread().name))

if __name__ =="__main__":
    for i in range(6):
        threading.Thread(target=run).start()
        time.sleep(2)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

示例七:生产者与消费者模型

import threading,time,queue,random

#生产者
def product(id,q):
    while True:
        #随机产生数据
        num = random.randint(0,10000)
        #将数据放入到消息队列中
        q.put(num)
        #显示数据已经被生成
        print("生产者%d产生了%d数据放入队列"%(id,num))
        time.sleep(3)
    #任务完成
    q.task_done()

#消费者
def customer(id,q):
    while True:
        #获得消息队列的数据
        item = q.get()
        if item is None:
            break
        #显示数据已经被消费
        print("消费者%d消费%d数据"%(id,item))
        time.sleep(2)
    #re任务完成
    q.task_done()
if __name__ =="__main__":
    #消息队列
    q = queue.Queue()
    #创建两个线程
    #生产者线程
    for i in range(4):
        threading.Thread(target=product,args=(i,q)).start()
    #消费者线程
    for i in range(3):
        threading.Thread(target=customer, args=(i, q)).start()
  • 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
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

五、Python 协程
协程:可以用单线程实现多任务,协程其实是更小切分;进程是系统调度,协程是用户态调度。因为是单线程执行多任务所以协程不需要不需要加锁
在Python中通过yield生成器来实现协程

协程优点:
1、无需上下文切换的开销,避免无意义调度,提升性能
2、无需加锁及同步开销
3、方便切换控制流,简化编程模型
4、高并发、高扩展、低成本
协程缺点:
1、无法利用多核资源
2、进行阻塞操作会阻塞掉整个程序

示例一:协程实现生产者与消费者

#消费者
def consumer():
    a = ""
    print('c1') # 只有第一次会执行(启动生成器), 之后再调用生成器就会从yield处执行
    while True:
        #首先返回一个空字符串,然后被阻塞,等待生产者发出信号
        # 再次执行时从这里的yield继续执行, 将把produce传入的参数 n 赋给局部变量 n . 下轮循环再次遇到yield就会就将 a 返回给produce函数
        n = yield a 
        # 由于生成器在启动的时候遇到上面的yield就返回了, 所以第一次不会执行这条语句. 之后每次都会被执行
        print('c %d'%n)  
        if not n:
            return
        print('[CONSUMER] Consuming %s' % n)
        #在下一次执行到yield a 时,将值返回给生产者,即是200 ok
        a = '200 OK'  
        #b = 'fake 200 OK' # 返回的值与a无关
#生产者
def produce(c):
    #第一次发送None 与消费者建立关系
    c.send(None)
    #执行标记
    print('p1')
    n = 0
    while n < 5:
        n = n + 1
        #生产的数据
        print('[PRODUCER] Producing %s' % n)
        #生产的数据发送给消费者
        r = c.send(n) # 获取生成器consumer中由yield语句返回的下一个值,即200 ok
        print('[PRODUCER] Consumer return: %s' % r)
    c.close()
c = consumer()      # 并不会启动生成器, 只是将c变为一个生成器
produce(c)
  • 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
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

图解:
在这里插入图片描述

声明:本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:【wpsshop博客】
推荐阅读
相关标签
  

闽ICP备14008679号