zoukankan      html  css  js  c++  java
  • 第五十七节,线程、进程、协程

    线程

    首先弄清进程和线程之间的区别,这一点是非常重要的。线程与进程的不同之处在于,它们共享状态、内存和资源。对于线程来说,这个简单的区别既是它的优势,又是它的缺点。一方面,线程是轻量级的,并且相互之间易于通信,但另一方面,它们也带来了包括死锁、争用条件和高复杂性在内的各种问题。幸运的是,由于 GIL 和队列模块,与采用其他的语言相比,采用 Python 语言在线程实现的复杂性上要低得多。无论是创建进程或者线程都是为了实现并发操作

    Python进程、线程之间的原理图

    计算机有进程和线程的目的:提高执行效率
    计算机默认有主进程和主线程

    进程:
      优点:同时利用多个CPU,能够同时进行多个操作
      缺点:耗费资源(重新开辟内存空间)
      进程不是越多越好,理论上CPU个数(核数)=进程个数
      计算密集型适用于进程,因为计算之类的需要CPU运算(占用CPU)
    线程:
      优点:共享内存,IO操作时,创造并发操作
      缺点:枪战资源
      线程不是越多越好,具体案例具体分析,请求上下文切换耗时
      IO密集型适用于线程,IO操作打开文件网络通讯类,不需要占用CPU,只是由CPU调度一下(不占用CPU)

    自定义进程和线程:注意python解释器自带了主进程和主线程,比如在代码文件里没有定义线程和进程,程序也能运行就是靠的解释器自带主进程的主线程执行的

      自定义进程:
        由主进程创建,子进程
      自定义线程:
        由主线程创建,子线程

    GIL全局解释器锁:

    GIL全局解释器锁在进程入口,控制着进程数量与CPU的相应

    threading线程模块

    线程是应用程序中工作的最小单元

    threading 模块建立在 _thread 模块之上。thread 模块以低级、原始的方式来处理和控制线程,而 threading 模块通过对 thread 进行二次封装,提供了更方便的 api 来处理线程。

    Thread()创建线程对象【有参】

    使用方法:赋值变量 = 模块名称.Thread(target=事件函数,args=元祖类型事件函数的实际参数)  如函数多个参数,元祖里就是多个元素

    格式:t = threading.Thread(target=show, args=(i,))

    currentThread()获取当前线程【无参】

    使用方法:自定义变量 = threading模块名称.currentThread()

    格式:current_thread = threading.currentThread()

    start()激活线程【无参】

    使用方法:thread对象变量.start()

    格式:t.start()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
    print("默认主线程等待子线程完成任务后,主线程停止")
    # 输出
    # 默认主线程等待子线程完成任务后,主线程停止
    # 线程0
    # 线程5
    # 线程8
    # 线程3
    # 线程6
    # 线程4
    # 线程1
    # 线程7
    # 线程2
    # 线程9

    上述代码创建了10个“前台”线程,然后控制器就交给了CPU,CPU根据指定算法进行调度,分片执行指令。

     自定义线程类

    import threading
    import time
     
     
    class MyThread(threading.Thread):
        def __init__(self,num):
            threading.Thread.__init__(self)
            self.num = num
     
        def run(self):#定义每个线程要运行的函数
     
            print("running on number:%s" %self.num)
     
            time.sleep(3)
     
    if __name__ == '__main__':
     
        t1 = MyThread(1)
        t2 = MyThread(2)
        t1.start()
        t2.start()

    getName()获取线程的名称【无参】

    使用方法:thread对象变量.getName()

    格式:t.getName()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
        print(t.getName()) #获取线程的名称
    print("默认主线程等待子线程完成任务后,主线程停止")
    # 输出
    # Thread-1
    # Thread-2
    # Thread-3
    # Thread-4
    # Thread-5
    # Thread-6
    # Thread-7
    # Thread-8
    # Thread-9
    # Thread-10
    # 默认主线程等待子线程完成任务后,主线程停止
    # 线程2
    # 线程1
    # 线程0
    # 线程9
    # 线程8
    # 线程6
    # 线程3
    # 线程5
    # 线程7
    # 线程4

    setName()设置线程的名称【有参】

    使用方法:thread对象变量.setName("要设置的线程名称")

    格式:t.setName("jhf")

    name获取或设置线程的名称【无参】

    使用方法:thread对象变量.name

    格式:t.name

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.setName("jhf") #设置线程的名称
        print(t.name) #获取或设置线程的名称
        t.start() #激活子线程
    
    print("默认主线程等待子线程完成任务后,主线程停止")
    # 输出
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # jhf
    # 默认主线程等待子线程完成任务后,主线程停止
    # 线程1
    # 线程0
    # 线程2
    # 线程7
    # 线程5
    # 线程3
    # 线程4
    # 线程9
    # 线程8
    # 线程6

    is_alive()判断线程是否为激活状态返回布尔值【无参】

    使用方法:thread对象变量.is_alive()

    格式:t.is_alive()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
    a = t.is_alive() #判断线程是否为激活状态返回布尔值
    print(a) #打印出返回值
    print("默认主线程等待子线程完成任务后,主线程停止")
    # 输出
    # True
    # 默认主线程等待子线程完成任务后,主线程停止
    # 线程3
    # 线程4
    # 线程2
    # 线程5
    # 线程1
    # 线程0
    # 线程6
    # 线程7
    # 线程8
    # 线程9

    isAlive()判断线程是否为激活状态返回布尔值【无参】

    使用方法:thread对象变量.isAlive()

    格式:t.isAlive()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
    a = t.isAlive() #判断线程是否为激活状态返回布尔值
    print(a) #打印出返回值
    print("默认主线程等待子线程完成任务后,主线程停止")
    # 输出
    # True
    # 默认主线程等待子线程完成任务后,主线程停止
    # 线程3
    # 线程4
    # 线程2
    # 线程5
    # 线程1
    # 线程0
    # 线程6
    # 线程7
    # 线程8
    # 线程9

    setDaemon() 设置为后台线程或前台线程,也就是定义主线程是否等待子线程执行完毕后,主线程才停止【有参】
    (默认:False);通过一个布尔值设置线程是否为守护线程,必须在执行start()方法之后才可以使用。如果是后台线程,主线程执行过程中,后台线程也在进行,主线程执行完毕后,后台线程不论成功与否,均停止;如果是前台线程,主线程执行过程中,前台线程也在进行,主线程执行完毕后,等待前台线程也执行完成后,程序停止

    使用方法:thread对象变量.setDaemon(布尔值)

    格式:t.setDaemon(True)

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.setDaemon(True) #设置为后台线程或前台线程,也就是定义主线程是否等待子线程执行完毕后,主线程才停止
        t.start() #激活子线程
    # 输出
    # 主线没等子线程执行完,主线程就停止了,所以没输出信息

    isDaemon()判断是否为守护线程,也就是主线程是否等待子线程执行完成后,才停止主线程,返回布尔值【无参】

    使用方法:thread对象变量.isDaemon()

    格式:t.isDaemon()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
    a = t.isDaemon() #判断是否为守护线程,也就是主线程是否等待子线程执行完成后,才停止主线程返回布尔值
    print(a) #打印布尔值
    # 输出
    # False
    # 线程1
    # 线程3
    # 线程0
    # 线程5
    # 线程2
    # 线程4
    # 线程9
    # 线程6
    # 线程8
    # 线程7

    ident获取线程的标识符。线程标识符是一个非零整数,只有在调用了start()方法之后该属性才有效,否则它只返回None。【无参】

    使用方法:thread对象变量.ident

    格式:t.ident

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(3) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
        a = t.ident #获取线程的标识符。线程标识符是一个非零整数,只有在调用了start()方法之后该属性才有效,否则它只返回None。
        print(a) #打印线程的标识符
    # 输出
    # 10040
    # 13172
    # 12096
    # 4456
    # 10200
    # 844
    # 2200
    # 2440
    # 2968
    # 12756
    # 线程3
    # 线程2
    # 线程1
    # 线程0
    # 线程7
    # 线程9
    # 线程8
    # 线程4
    # 线程5
    # 线程6

    join()逐个执行每个线程,等待一个线程执行完毕后继续往下执行,该方法使得多线程变得无意义【有参可选】

    有参可选,参数为等待时间,秒为单位,如t.join(1) 就是一个线程不在是等待它执行完,而是只等待它1秒后继续下一个线程

    使用方法:thread对象变量.join()

    格式:t.join()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    def show(arg): #定义函数
        time.sleep(1) #睡眠3秒
        print('线程'+str(arg)) #打印线程加循环次数
    for i in range(10): #定义一个10次循环
        t = threading.Thread(target=show, args=(i,)) #用threading模块的Thread类来创建子线程对象
        t.start() #激活子线程
        t.join() #逐个执行每个线程,执行完毕后继续往下执行,该方法使得多线程变得无意义
    # 输出
    # 线程0
    # 线程1
    # 线程2
    # 线程3
    # 线程4
    # 线程5
    # 线程6
    # 线程7
    # 线程8
    # 线程9

    run()线程被cpu调度后自动执行线程对象的run方法

    使用方法:thread对象变量.run()

    格式:t.run()

    线程锁threading.RLock和threading.Lock

     我们使用线程对数据进行操作的时候,如果多个线程同时修改某个数据,可能会出现不可预料的结果,为了保证数据的准确性,引入了锁的概念。

    没有线程锁的情况列如:一个全局变量值为50,创建10条线程让每条线程累计减一,输出的结果是10个40,原因是10条线程同时减一就减去了10,所以打印出来就是10个40了

    未使用锁

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    globals_num = 50 #设置一个变量
    def Func(a): #定义一个函数
        global globals_num #将变量转换成全局变量,函数里可以调用
        globals_num -= 1 #全局变量减1
        time.sleep(1) #睡眠1秒
        print(globals_num,a) #打印全局变量减少后的结果,和函数传进来的值
    for i in range(10): #创建一个10次循环
        t = threading.Thread(target=Func,args=(i,)) #创建线程对象
        t.start() #激活线程
    # 输出  没有线程锁,线程之间抢占了数据资源
    # 40 5
    # 40 3
    # 40 6
    # 40 4
    # 40 0
    # 40 2
    # 40 1
    # 40 9
    # 40 8
    # 40 7
    根据上列情况可以看出,没有线程锁,线程之间抢占了数据资源
    线程锁就是将线程锁定,一个线程执行完毕后释放锁后在执行第二个线程

    RLock()定义线程锁对象

    使用方法:定义对象变量 = threading模块名称.RLock()

    格式:lock = threading.RLock()

    acquire()获得锁,将线程锁定,一个线程执行完毕释放锁后在执行第二个线程

    使用方法:线程锁对象变量.acquire()

    格式:lock.acquire()

    release()释放线程锁

    使用方法:线程锁对象变量.release()

    格式:lock.release()

    使用锁

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading #导入线程模块
    import time #导入时间模块
    globals_num = 50 #设置一个变量
    lock = threading.RLock()
    def Func(a): #定义一个函数
    
        lock.acquire()  # 获得锁,将线程锁定,一个线程执行完毕后在执行第二个线程
    
        global globals_num #将变量转换成全局变量,函数里可以调用
        globals_num -= 1 #全局变量减1
        time.sleep(1) #睡眠1秒
        print(globals_num,a) #打印全局变量减少后的结果,和函数传进来的值
    
        lock.release()  # 释放锁
    
    for i in range(10): #创建一个10次循环
        t = threading.Thread(target=Func,args=(i,)) #创建线程对象
        t.start() #激活线程
    # 输出  将线程锁定,一个线程执行完毕后在执行第二个线程
    # 49 0
    # 48 1
    # 47 2
    # 46 3
    # 45 4
    # 44 5
    # 43 6
    # 42 7
    # 41 8
    # 40 9

    threading.RLock和threading.Lock 的区别

    RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。 如果使用RLock,那么acquire和release必须成对出现,即调用了n次acquire,必须调用n次的release才能真正释放所占用的琐。

    import threading
    lock = threading.Lock()    #Lock对象
    lock.acquire()
    lock.acquire()  #产生了死琐。
    lock.release()
    lock.release()
    import threading
    rLock = threading.RLock()  #RLock对象
    rLock.acquire()
    rLock.acquire()    #在同一线程内,程序不会堵塞。
    rLock.release()
    rLock.release()

    threading.Event事件对象

    Event()创建标识事件对象,全局定义了一个“Flag”,如果“Flag”值为 False,那么当程序执行 event.wait 方法时就会阻塞,如果“Flag”值为True,那么event.wait 方法时便不再阻塞

    python线程的事件用于主线程控制其他线程的执行,事件主要提供了三个方法 set、wait、clear。

    事件处理的机制:全局定义了一个“Flag”,如果“Flag”值为 False,那么当程序执行 event.wait 方法时就会阻塞,如果“Flag”值为True,那么event.wait 方法时便不再阻塞。

    当线程执行的时候,如果flag为False,则线程会阻塞,当flag为True的时候,线程不会阻塞。它提供了本地和远程的并发性。

    Event事件对象的方法有

      wait([timeout]) : 堵塞线程,直到Event对象内部标识位被设为True或超时(如果提供了参数timeout)。
      set() :将标识位设为Ture
      clear() : 将标识位设为False。
      isSet() :判断标识位是否为Ture。

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading
    def do(event):
        print('start')
        event.wait() #堵塞线程,直到Event对象内部标识位被设为True或超时(如果提供了参数timeout)
        print('execute')
    event_obj = threading.Event() #创建标识事件对象
    for i in range(10):
        t = threading.Thread(target=do, args=(event_obj,)) #创建线程对象
        t.start() #激活线程
    event_obj.clear() #将标识设置为False
    inp = input('input:')
    if inp == 'true':
        event_obj.set() #将标识设置为True
    # 输出
    # start
    # start
    # start
    # start
    # start
    # start
    # start
    # start
    # start
    # input:true
    # execute
    # execute
    # execute
    # execute
    # execute
    # execute
    # execute
    # execute
    # execute
    # execute

    threading.BoundedSemaphore信号对象

    是同时允许一定数量的线程更改数据 ,比如厕所有3个坑,那最多只允许3个人上厕所,后面的人只能等里面有人出来了才能再进去。

     BoundedSemaphore()创建信号对象【有参】

    使用方法:定义变量.threading.BoundedSemaphore(最大允许线程数)

    格式:semaphore  = threading.BoundedSemaphore(5)

    BoundedSemaphore信号对象的方法有

      acquire()获取信号
      release()释放信号

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading,time #导入线程模块,和时间模块
    
    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
    def run(n): #创建函数
    
        semaphore.acquire() #获取信号
    
        time.sleep(1)
        print("run the thread: %s" %n)
    
        #semaphore.release() #释放信号
    
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))
        t.start()
    # 输出
    # run the thread: 3
    # run the thread: 2
    # run the thread: 0
    # run the thread: 1
    # run the thread: 4

    threading.Condition条件对象

    使得线程等待,只有满足某条件时,才释放n个线程

    Condition()创建条件对象【无参】

    使用方法:定义变量.threading.Condition()

    格式:con = threading.Condition()

    Condition条件对象的方法有

      acquire()
      wait()
      release()
      notify()

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import threading
    
    con = threading.Condition()
    
    def run(n):
    
        con.acquire()
    
        con.wait()
    
        print("run the thread: %s" %n)
        con.release()
    
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()
    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()

    queue模块

    Queue 就是对队列,它是线程安全的

    举例来说,我们去肯德基吃饭。厨房是给我们做饭的地方,前台负责把厨房做好的饭卖给顾客,顾客则去前台排队领取做好的饭。这里的前台就相当于我们的队列。

    这个模型也叫生产者-消费者模型。

    Queue()创建队列对象【有参】

    使用方法:定义变量 = queue.Queue(对列长度数) 0表示长度无限制

    格式:message = queue.Queue(10)

    Queue对象方法有:

      join()等到队列为空的时候,在执行别的操作【无参】

      qsize()返回队列的大小(不可靠),因为获取后有可能有新数据加入【无参】

      empty()清空队列里的所有数据

      full()检查队列是否为满,当队列满的时候,返回True,否则返回False(不可靠),因为获取后有可能有新数据加入【无参】

      put(放入对列的数据必选, block=True, timeout=None) 向队列里放入数据(生产)【有参】

        将数据放入对列尾部(生产),数据必须存在,可以参数block默认为True,表示当队列满时,会等待队列给出可用位置,为False时为非阻塞,此时如果队列已满,会引发queue.Full 异常。

        可选参数timeout,表示 会阻塞设置的时间,过后,如果队列无法给出放入item的位置,则引发 queue.Full 异常

      get(block=True, timeout=None)移除并返回队列头部的一个值(消费)【有参】

         可选参数block默认为True,表示获取值的时候,如果队列为空,则阻塞,为False时,不阻塞,若此时队列为空,则引发 queue.Empty异常。

        可选参数timeout,表示会阻塞设置的时候,过后,如果队列为空,则引发Empty异常。

      put_nowait(放入对列的数据必选)向队列里放入数据(生产)【有参】,如果队列满时不阻塞,不等待队列给出可用位置,引发 queue.Full 异常

      get_nowait()移除并返回队列头部的一个值(消费)【无参】,如果队列空时不阻塞,引发 queue.Full 异常

      

    对列模型-生产者-消费者

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import queue #导入列队模块
    import threading #导入线程模块
    
    message = queue.Queue(10) #定义列队对象,设置列队长度
    
    def producer(i): #定义生产者函数
        while True:
            message.put("生产") #向队列里放数据
    
    def consumer(i): #定义消费者函数
        while True:
            msg = message.get() #从队列里取数据
            print(msg) #打印出从队列里取出
    
    for i in range(12): #创建12条线程生产,也就是有12条线程向队列里放数据
        t = threading.Thread(target=producer, args=(i,)) #创建线程对象
        t.start() #激活线程
    
    for i in range(10): #创建10条线程消费,也就是有10条线程从列队里取数据
        t = threading.Thread(target=consumer, args=(i,)) #创建线程对象
        t.start() #激活线程
    # 输出
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产
    # 生产

     对列模型-生产者-消费者原理图

    multiprocessing进程模块

    multiprocessing是python的多进程管理包,和threading.Thread类似。直接从侧面用subprocesses替换线程使用GIL的方式,由于这一点,multiprocessing模块可以让程序员在给定的机器上充分的利用CPU。

    在multiprocessing中,通过创建Process对象生成进程,然后调用它的start()方法,

     Process()创建进程对象【有参】

    注意:wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题

     使用方法:定义变量 = multiprocessing.Process(target=要创建进程的函数, args=元祖类型要创建进程函数的参数、多个参数逗号隔开)

    格式:t = multiprocessing.Process(target=f1, args=(133,))

    start()激活进程【无参】

    使用方法:Process对象变量.start()

    格式:t.start()

    创建10条进程

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    def f1(r): #创建函数
        print(r) #打印传值
    if __name__ == "__main__": #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        for i in range(10): #循环10次,创建10条进程
            t = multiprocessing.Process(target=f1, args=(133,)) #创建进程对象
            t.start() #激活进程
    # 输出
    # 133
    # 133
    # 133
    # 133
    # 133
    # 133
    # 133
    # 133
    # 133
    # 133

    daemon主进程是否等待子进程执行完毕后,在停止主进程,daemon=True(主进程不等待子进程)、daemon=False(主进程等待子进程)

    使用方法:Process对象变量.daemon=True或者False

    格式:t.daemon = False

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    def f1(r): #创建函数
        print(r) #打印传值
    if __name__ == "__main__": #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        for i in range(10): #循环10次,创建10条子进程
            t = multiprocessing.Process(target=f1, args=(133,)) #创建进程对象
    
            t.daemon = True #主进程是否等待子进程执行完毕后,在停止主进程
            
            t.start() #激活进程
    # 输出 #daemon = False 主进程没等子进程执行完,主进程就停止了,所以没有打印出信息

    join()逐个执行每个进程,等待一个进程执行完毕后继续往下执行,该方法使得进程程变得无意义【有参可选】

    有参可选,参数为等待时间,秒为单位,如t.join(1) 就是一个进程不在是等待它执行完,而是只等待它1秒后继续下一个进程

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    import time
    def f1(r): #创建函数
        time.sleep(1)
        print(r) #打印传值
    if __name__ == "__main__": #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        for i in range(10): #循环10次,创建10条子进程
            t = multiprocessing.Process(target=f1, args=(133,)) #创建进程对象
            t.start() #激活进程
    
            t.join() #逐个执行每个进程,等待一个进程执行完毕后继续往下执行
    # 输出
    # 133
    # 133
    # 133
    # 133

    进程各自持有一份数据,默认无法共享数据

    所以相当于每一个进程有一份自己的数据,每个进程操作数据时,操作的属于自己的一份数据

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    li = [] #创建空列表
    def f1(i): #创建函数
        li.append(i) #追加列表
        print("列表",li) #打印追加后的列表
    if __name__ == "__main__": #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        for i in range(10): #循环10次,创建10条子进程,进程各自持有一份数据,默认无法共享数据,所以相当于每一个进程有一个f1函数,每个进程在追加列表时追加的属于自己的一份f1函数
            t = multiprocessing.Process(target=f1, args=(i,)) #创建进程对象
            t.start() #激活进程
    # 输出
    # 列表 [0]
    # 列表 [2]
    # 列表 [5]
    # 列表 [3]
    # 列表 [1]
    # 列表 [6]
    # 列表 [8]
    # 列表 [7]
    # 列表 [4]
    # 列表 [9]

     进程原理图

    注意:由于进程之间的数据需要各自持有一份,所以创建进程需要的非常大的开销。

    进程数据共享

    注意:进程与进程之间无法共享数据,要想共享数据就得用特殊方法,在主进程创建特殊数据,然后几个子进程来共享这个主进程的特殊数据

    方法一

    Array()创建数组,数组,定义数组必须要定义数组的长度,数组里必须是统一的数据类型【有参】

    使用方法:Array('指定数组数据类型',列表样式的数组元素)

    指定数组数据类型有:

    'c': ctypes.c_char,  'u': ctypes.c_wchar,
    'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
    'h': ctypes.c_short, 'H': ctypes.c_ushort,
    'i': ctypes.c_int,   'I': ctypes.c_uint,
    'l': ctypes.c_long,  'L': ctypes.c_ulong,
    'f': ctypes.c_float, 'd': ctypes.c_double

    利用Array()数组来多进程共享数据(不推荐使用)

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    temp = multiprocessing.Array('i', [11,22,33,44,]) #创建数组
    
    def Foo(i): #定义函数
        #第一条进程,将100加0等于100,重新赋值给数组里的第0个元素,也就是将数组里的11改成了100
        #第二条进程,将100加1等于101,重新赋值给数组里的第1个元素,也就是将数组里的22改成了101
        temp[i] = 100+i
        for item in temp: #循环数组
            print(i,'----->',item) #循环打印进程线,和数组元素
        print("
    ")
    if __name__ == "__main__": #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        for i in range(2):
            p = multiprocessing.Process(target=Foo,args=(i,)) #创建进程对象
            p.start() #激活进程
    
    # 输出
    # 0 -----> 100
    # 0 -----> 22
    # 0 -----> 33
    # 0 -----> 44
    
    # 1 -----> 11
    # 1 -----> 101
    # 1 -----> 33
    # 1 -----> 44

    方法二

    Manager()创建特殊字典对象【无参】

    使用方法:定义变量 = multiprocessing模块名称.Manager()

    格式:manage = multiprocessing.Manager()

    dict()创建特殊字典【可选参数】

    参数为字段值,一般都不设置,为空即可,注意:这个特殊字典和前面的字典有所区别,但大部分使用方法相同,可以索引,可以values()取值

    使用方法:殊字典对象变量.dict()

    格式:dic = manage.dict()

    利用特殊字典dict()来多进程共享数据【推荐】

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    def Foo(i,dic): #定义函数
        dic[i] = 100+i #100加以进程线,索引方式重新赋值给特殊字典
        print(dic.values()) #打印特殊字典的所有值
    if __name__ == '__main__': #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
    
        manage = multiprocessing.Manager() #创建特殊字典对象
        dic = manage.dict() #创建特殊字典,值为空
    
        for i in range(10): #循环创建10条进程
            p = multiprocessing.Process(target=Foo,args=(i,dic,)) #创建进程对象
            p.start() #激活进程
            p.join() #等待一个进程执行完,在执行第二个进程,否则主进程停止了无法共享数据,因为共享数据时在主进程里
    # 输出
    # [100]
    # [100, 101]
    # [100, 101, 102]
    # [100, 101, 102, 103]
    # [100, 101, 102, 103, 104]
    # [100, 101, 102, 103, 104, 105]
    # [100, 101, 102, 103, 104, 105, 106]
    # [100, 101, 102, 103, 104, 105, 106, 107]
    # [100, 101, 102, 103, 104, 105, 106, 107, 108]
    # [100, 101, 102, 103, 104, 105, 106, 107, 108, 109]

    进程-队列-生产者-消费者【不推荐】严重耗费内存资源

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    
    def f2(i,q): #定义生产者函数
        q.put("h1") #向队列里放数据
    
    def f(i,q): #定义消费者函数
        print(i,q.get()) #向列队里取数据
    
    if __name__ == '__main__': #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        q = multiprocessing.Queue() #定义对列
    
        for i in range(10): #创建10条进程生产
            p = multiprocessing.Process(target=f2, args=(i,q,)) #创建进程对象
            p.start() #激活进程
    
        for i in range(10): #创建10条进程消费
            p = multiprocessing.Process(target=f, args=(i,q,)) #创建进程对象
            p.start() #激活进程
    # 输出
    # 1 h1
    # 0 h1
    # 8 h1
    # 5 h1
    # 3 h1
    # 6 h1
    # 7 h1
    # 2 h1
    # 9 h1
    # 4 h1

    进程锁

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import multiprocessing #导入进程模块
    def Foo(lock,temp,i): #创建函数
        """
        将第0个数加100
        """
        lock.acquire() #获取进程锁
        temp[0] = 100+i #100加上进程线循环次数,重新赋值给进程循环次数对应下标数组里的值
        for item in temp: #循环数组
            print(i,'----->',item) #打印出进程循环次数,和数组
        lock.release() #释放进程锁
        print("
    ")
    if __name__ == '__main__': #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
        lock = multiprocessing.RLock() #创建进程锁对象
        temp = multiprocessing.Array('i', [11, 22, 33, 44]) #创建数组
    
        for i in range(5): #循环创建5条子进程
            p = multiprocessing.Process(target=Foo,args=(lock,temp,i,)) #创建进程对象
            p.start() #激活进程
    # 输出
    # 0 -----> 100
    # 0 -----> 22
    # 0 -----> 33
    # 0 -----> 44
    
    # 1 -----> 101
    # 1 -----> 22
    # 1 -----> 33
    # 1 -----> 44
    
    # 3 -----> 103
    # 3 -----> 22
    # 3 -----> 33
    # 3 -----> 44
    
    # 2 -----> 102
    # 2 -----> 22
    # 2 -----> 33
    # 2 -----> 44
    
    # 4 -----> 104
    # 4 -----> 22
    # 4 -----> 33
    # 4 -----> 44

    进程池

    进程池内部维护一个进程序列,当使用时,则去进程池中获取一个进程,如果进程池序列中没有可供使用的进进程,那么程序就会等待,直到进程池中有可用进程为止。进程池Python有提供

    Pool()创建进程池对象【有参】

    默认进程池里没有进程,只有在向进程池申请进程的时候,进程池才创建进程

    使用方法:定义变量 = multiprocessing模块名称.Pool(定义进程数)

    格式:pool = multiprocessing.Pool(5)

    close()进程池里的进程执行完毕后关闭进程池连接【无参】

    使用方法:进程池对象变量.close()

    格式:pool.close()

    terminate()不等进程池里的进程执行完毕,立即关闭进程池连接

    使用方法:进程池对象变量.terminate()

    格式:pool.terminate()

    join()主进程等待进程池里的子进程全部执行完成后,主进程才停止【可选参数】

    可选参数,不写就是等待直到子进程全部执行完成后,主进程才停止,写了就是只等待指定的时间,时间到了就停止主进程,不管子进程有没有完成

    使用方法:进程池对象变量.join(可选参数秒)

    格式:pool.join()

    向进程池申请进程的方法

    apply()向进程池申请一条进程,进程函数执行完后将进程放回进程池,【有参】

    注意:apply()向进程池申请的进程不是并发的,是一个进程执行完毕后在执行一个进程,以此循环的,

    apply()向进程池申请进程的时候,进程池创建的每一个进程都有一个,进程对象.join()方法,所以进程才是排队执行的,这里我们需要知道一下

    使用方法:进程池对象变量.apply(func=要执行的进程函数名称,args=(执行函数的实际参数、多个参数逗号隔开))

    格式:pool.apply(func=Foo,args=(i,))

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import  multiprocessing #导入进程模块
    import time #导入时间模块
    def Foo(i): #定义进程执行函数
        time.sleep(1)
        print(i+100)
    if __name__ == '__main__': #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
    
        pool = multiprocessing.Pool(5) #定义进程池对象
    
        for i in range(10): #循环向进程池申请10条进程
            pool.apply(func=Foo,args=(i,)) #向进程池申请进程
    # 输出
    # 100
    # 101
    # 102
    # 103
    # 104
    # 105
    # 106
    # 107
    # 108
    # 109

     apply_async()向进程池申请一条进程,进程函数执行完后将进程放回进程池,并且可以设置进程执行函数的回调函数

    注意:apply_async()向进程池申请的进程是并发的,也就是申请了几条进程就是同时执行几条进程的,回调函数的形式参数、接收的进程执行函数的返回值

    apply_async()向进程池申请进程的时候,进程池创建的进程都没有,进程对象.join()方法,所以进程都是并发的,而且进程对象的daemon=True,也就是主进程不会等待子进程执行完毕就终止,所以使用apply_async()向进程池申请进程的时候,进程申请后,要使用close()进程池里的进程执行完毕后关闭进程池连接,join()主进程等待进程池里的子进程全部执行完成后,主进程才停止,否则会报错

    使用方法:进程池对象变量.apply_async(func=要执行的进程函数名称,args=(执行函数的实际参数、多个参数逗号隔开),callback=回调函数名称)

    格式:pool.apply_async(func=Foo,args=(i,),callback=f2)

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import  multiprocessing #导入进程模块
    import time #导入时间模块
    def Foo(i): #定义进程执行函数
        time.sleep(1)
        print(i+100)
        return "返回值,返回给回调函数的,形式参数"
    def f2(a): #执行函数的回调函数,形式参数等于执行函数的返回值
        print(a) #打印进程执行函数返回的值
    if __name__ == '__main__': #wds系统下必须if __name__ == "__main__"才能创建进程,我们调试没关系,以后在Linux系统没这个问题
    
        pool = multiprocessing.Pool(5) #定义进程池对象
    
        for i in range(10): #循环向进程池申请10条进程
            pool.apply_async(func=Foo,args=(i,),callback=f2) #向进程池申请进程,并设置执行函数,和回调函数
    
        pool.close() #进程池里的进程执行完毕后关闭进程池连接
        pool.join()#主进程等待进程池里的子进程全部执行完成后,主进程才停止
    # 输出
    # 100
    # 返回值,返回给回调函数的,形式参数
    # 101
    # 返回值,返回给回调函数的,形式参数
    # 102
    # 返回值,返回给回调函数的,形式参数
    # 103
    # 返回值,返回给回调函数的,形式参数
    # 104
    # 返回值,返回给回调函数的,形式参数
    # 105
    # 返回值,返回给回调函数的,形式参数
    # 106
    # 返回值,返回给回调函数的,形式参数
    # 107
    # 返回值,返回给回调函数的,形式参数
    # 108
    # 返回值,返回给回调函数的,形式参数
    # 109
    # 返回值,返回给回调函数的,形式参数

     apply_async()向进程池申请进程原理图

    自定义线程池

    自定义线程池版本一

    这个版本并不理想,但是简单

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import queue #导入队列模块
    import threading #导入线程模块
    """定义一个类"""
    class ThreadPool(object): #创建类
        def __init__(self, max_num=20): #初始化
            self.queue = queue.Queue(max_num) #定义普通字段等于,长度为20的队列
            for i in range(max_num): #设置20次循环
                self.queue.put(threading.Thread) #循环向队列里,放入20个线程对象名称
    
        def get_thread(self): #定义get_thread方法
            return self.queue.get() #返回在队列里取出线程名称
    
        def add_thread(self): #定义add_thread方法
            self.queue.put(threading.Thread) #向队列里放入一个线程对象名称
    
    """创建一个类对象"""
    pool = ThreadPool(20) #创建类对象,初始化__init__方法
    
    """定义线程执行函数"""
    def func(arg, p): #定义线程执行函数
        print(arg) #打印线程数,也就是第几次循环线程
        import time #导入时间模块
        time.sleep(2) #睡眠2秒
        p.add_thread() #向队列放入一个线程对象名称,创建一个线程对象的时候,就从队列里拿走一个线程对象名称,所有要在放回一个回去
    
    """创建线程"""
    for i in range(30): #定义一个30次循环
        thread = pool.get_thread() #在列队里拿出一个线程名称
        t = thread(target=func, args=(i, pool)) #在队列里拿出一个线程对象名称,创建一个线程对象,传入线程执行函数和参数
        t.start() #激活线程
    
    # 输出
    # 0
    # 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

     自定义线程池版本一原理图

    自定义线程池版本二【推荐使用】

    ThreadPool源码模块,使用方法将ThreadPool源码模块文件,放到工程目录下,导入模块使用

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    """线程池源码"""
    import queue    #导入队列模块
    import threading    #导入线程模块
    import contextlib   #导入上下文管理模块
    
    StopEvent = object()    #设置全局变量,停止标志
    class ThreadPool(object):   #创建类
        """
        ThreadPool()创建线程池类对象,有参:线程最大数量【使用方法:定义线程池对象变量 = ThreadPool(线程最大数量)】
        run()向线程池申请一条线程,有参【使用方法:线程对象.run(线程执行函数,(执行函数参数),回调函数)】
        close()让所有线程执行完毕后,停止线程,无参【使用方法:线程对象.close()】
        terminate()无论是否还有任务,终止线程,有参:是否立即清空队列里的数据,默认yes清空,no不清空【使用方法:线程对象.terminate(yes或no)】
        """
        def __init__(self, max_num, max_task_num = None):
            """
            初始化ThreadPool类数据,创建队列,记录线程最大数量,创建、记录真实创建的线程列表
            创建、记录空闲线程列表
            """
            if max_task_num:    #判断max_task_num如果有值
                self.q = queue.Queue(max_task_num)  #创建队列,队列长度为max_task_num的值
            else:   #如果max_task_num没有值
                self.q = queue.Queue()  #创建队列,队列的长度没有限制
            self.max_num = max_num  #创建普通字段max_num等于,定义ThreadPool类对象的第一个实际参数,也就是最多能创建的线程数
            #self.cancel = False #创建普通字段cancel = False
            self.terminal = False    #创建普通字段terminal = False,以这个标识决定线程是否继续到队列取任务
            self.generate_list = []  #创建generate_list空列表,记录真实创建的线程
            self.free_list = []     #创建free_list空列表,记录空闲线程
    
        def run(self, func, args, callback=None):
            w = (func, args, callback,)  #将传进来的,线程执行函数名称和执行函数参数,以及回调函数名称,组合成一个元组赋值给w变量
            self.q.put(w)   #将元祖放入对列中
            """判断空闲线程列表等于0,也就是空闲列表里没有空闲的线程时,
            并且真实创建的线程列表里的线程数小于总线程数,执行generate_thread方法
            """
            if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
                self.generate_thread()   #执行generate_thread方法
    
        def generate_thread(self):
            t = threading.Thread(target=self.call)   #创建一个线程,线程执行函数为call方法
            t.start()   #激活线程
    
        def call(self):
            """
            循环去获取任务函数并执行任务函数
            """
            current_thread = threading.currentThread()  #获取当前线程
            self.generate_list.append(current_thread)    #将获取到的当前线程,追加到真实创建的线程列表里
            event = self.q.get()     #到队列里取出,run方法放入队列的元组
            while event != StopEvent:    #如果到队列里取到的不等于停止标志,说明是元组,如果是元组开始循环
                """将元组里的3个元素,分别赋值给3个变量,第一个是线程执行函数名称,第二个是线程执行函数参数,第三个是回调函数名称"""
                func, arguments, callback = event
                #success = True #自定义一个执行函数是否执行成功标识,默认True表示成功
                try:
                    result = func(*arguments)    #执行线程执行函数,并接收执行函数参数
                except Exception as e:   #如果执行错误
                    result = e  #如果线程执行函数出错,线程执行函数返回值等于错误信息
                if callback is not None:     #如果回调函数存在
                    try:
                        callback(result)    #执行回调函数,并将执行函数返回结果传值给回调函数的形式参数result
                    except Exception as e:
                        pass
    
                """标记线程空闲了"""
                if self.terminal:   #判断terminal变量是True
                    event = StopEvent   #如果是True就想列队里放入线程停止标志
                else:
                    with self.worker_state(self.free_list, current_thread):   #执行里面代码块前先执行上下文管理函数
                        event = self.q.get()     #到队列里取出,run方法放入队列的元组,没有就等待
            else:
                self.generate_list.remove(current_thread)   #如果在队列里取出的不是元组,而是停止标识,就在真实创建的线程列表里移除当前的线程
    
        def close(self):
            """
            执行完所有的任务后,所有线程停止
            """
            full_size = len(self.generate_list)     #获取真实创建的线程列表里的线程个数
            while full_size:     #循环,真实创建线程列表里,的线程个数对应的次数
                self.q.put(StopEvent)   #每循环一次,向队列里加一个全局变量StopEvent,停止标识
                full_size -= 1  #每循环一次让循环次数减一
    
        def terminate(self, qkdl = "yes"):
            """
            无论是否还有任务,终止线程
            """
            if qkdl == "yes":
                self.terminal = True    #将是否继续到队列取任务的判断变量修改为True,向队列里放停止标识,使其线程停止
                self.q.empty()  #清空队列里的所有数据
                zuiduo = len(self.generate_list)     #检查真实创建线程列表里有多少个线程
                while zuiduo:   #循环真实创建线程列表里线程数,对应次数
                    self.q.put(StopEvent)    #每循环一次向队列里放停止标识
                    zuiduo -= 1     #每循环一次,减少一次循环次数
            else:
                self.terminal = True    #将是否继续到队列取任务的判断变量修改为True,向队列里放停止标识,使其线程停止
                zuiduo = len(self.generate_list)    #检查真实创建线程列表里有多少个线程
                while zuiduo:   #循环真实创建线程列表里线程数,对应次数
                    self.q.put(StopEvent)   #每循环一次向队列里放停止标识
                    zuiduo -= 1     #每循环一次,减少一次循环次数
    
        @contextlib.contextmanager #定义上下文管理装饰器
        def worker_state(self, state_list, worker_thread): #定义上下文管理装饰器函数
            """
            用于记录线程中正在等待的线程数
            """
            state_list.append(worker_thread)    #执行代码块前执行,将当前线程追加到空闲线程列表
            try:
                yield   #遇到yield,跳出装饰器函数,执行代码块后,在回到yield这里向下执行
            finally:
                state_list.remove(worker_thread)    #执行代码块后执行,将当前线程移除空闲线程列表

    ThreadPool自定义线程池版本二模块使用说明

    首先from xxx import ThreadPool 导入模块

    ThreadPool()创建线程池对象【有参】

    使用方法:定义线程池对象变量 = ThreadPool模块名称.ThreadPool(线程池线程最大数量)

    格式:pool = ThreadPool.ThreadPool(5)

    run()到线程池申请一条线程【有参】

    使用方法:线程池对象.run(线程执行函数,(线程执行函数的参数),回调函数)

    格式:pool.run(f1,(i,),f2)

    close()执行完所有的任务后,所有线程停止【无参】

    使用方法:线程池对象.close()

    格式:pool.close()

    terminate()无论是否还有任务,终止线程【可选参数】

    使用方法:线程池对象.terminate()

    参数默认为yes终止线程前清空队列,no为终止线程不清空队列

    格式:pool.terminate()

    ThreadPool自定义线程池版本二使用代码

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    from lib.ska import ThreadPool #导入线程池模块
    import time #导入时间模块
    
    def f2(i): #定义回调函数
        print(i) #打印线程执行函数的返回值,回调函数的形式参数接收的,线程执行函数的返回值
    
    def f1(i): #定义线程执行函数
        time.sleep(1) #睡眠1秒
        print(i) #打印申请线程时传进来的参数
        return "回调" #返回值给回调函数
    
    pool = ThreadPool.ThreadPool(5) #创建线程池对象
    for i in range(100): #循环
        pool.run(f1,(i,),f2) #到线程池申请线程
    pool.close() #执行完所有的任务后,所有线程停止
    #pool.terminate() #无论是否还有任务,终止线程

    自定义线程池版本二原理图

    协程

    协程又叫(微线程),就是在一个线程里可以创建多个协程,由程序员来控制什么时候执行那条协程,协程可以用一条线程,同时执行多个任务,适用于IO密集型场景

    协程存在的意义:对于多线程应用,CPU通过切片的方式来切换线程间的执行,线程切换时需要耗时(保存状态,下次继续)。协程,则只使用一个线程,在一个线程中规定某个代码块执行顺序。
    协程的适用场景:当程序中存在大量不需要CPU的操作时(IO),适用于协程;

    greenlet最基础协程模块 第三方模块

    greenlet()创建协程对象【有参】
    使用方法:自定义变量 = greenlet(协程执行函数)
    格式:gr1 = greenlet(test1)

    switch()执行指定的协程,switch()前面为指定要执行的协程对象名称【无参】
    如果协程执行函数里,遇到switch()时就会跳出当前协程执行函数,并记录当前跳出位置,去执行遇到switch()指定的线程,记录的跳出位置下次进入时,从跳出位置开始
    使用方法:要执行的协程对象变量.switch()
    格式:gr1.switch()

    简单协程代码

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    from greenlet import greenlet #导入协程模块
    
    def test1():    #定义协程执行函数
        print(12)   #打印12
        gr2.switch()    #执行指定的协程,switch()前面为指定要执行的协程对象名称,执行gr2协程
        print(34)   #打印34
        gr2.switch()    #执行指定的协程,switch()前面为指定要执行的协程对象名称,执行gr1协程
    
    def test2():    #定义协程执行函数
        print(56) #打印56
        gr1.switch()    #执行指定的协程,switch()前面为指定要执行的协程对象名称,执行gr1协程
        print(78) #打印78
    
    gr1 = greenlet(test1) #创建协程对象,传入协程执行函数
    gr2 = greenlet(test2) #创建协程对象,传入协程执行函数
    gr1.switch() #执行指定的协程,switch()前面为指定要执行的协程对象名称,执行gr1协程

     简单协程原理图

    gevent协程模块

    gevent协程模块是基于greenlet模块改进的,也是第三方模块

    joinall()创建协程对象【有参】
    参数是列表类型的,创建协程spawn()方法
    使用方法:模块名称.joinall([gevent.spawn(线程执行函数)])
    格式:gevent.joinall([gevent.spawn(foo), gevent.spawn(bar), gevent.spawn(ba),])

    spawn()创建协程【有参】
    参数是协程执行函数名称
    使用方法:gevent模块名称.joinall([gevent.spawn(协程执行函数名称), gevent.spawn(协程执行函数名称), gevent.spawn(协程执行函数名称),])
    格式:gevent.joinall([gevent.spawn(foo), gevent.spawn(bar), gevent.spawn(ba),])

    sleep()跳出协程执行函数,执行协程对象里的,下一个协程,并记录当前跳出位置,再次进入当前协程执行函数时,从当前跳出位置开始
    使用方法:模块名称.sleep(0)
    格式:gevent.sleep(0)

    gevent简单协程代码

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    import gevent  #导入协程模块
    
    def foo():  #定义协程执行函数
        print(12)
        gevent.sleep(0) #执行协程对象里下一条协程,如果已经是最后一条协程,就返回第一条协程执行
        print(34)
    
    def bar():  #定义协程执行函数
        print(56)
        gevent.sleep(0) #执行协程对象里下一条协程,如果已经是最后一条协程,就返回第一条协程执行
        print(78)
    
    def ba():  #定义协程执行函数
        print(910)
        gevent.sleep(0) #执行协程对象里下一条协程,如果已经是最后一条协程,就返回第一条协程执行
        print(1112)
    
    gevent.joinall([    #定义协程对象
        gevent.spawn(foo),  #创建协程
        gevent.spawn(bar),  #创建协程
        gevent.spawn(ba),   #创建协程
    ])
    
    # 输出
    # 12
    # 56
    # 910
    # 34
    # 78
    # 1112

     gevent简单协程原理图

    遇到IO操作自动切换:

    #!/usr/bin/env python
    # -*- coding:utf8 -*-
    from gevent import monkey; monkey.patch_all() #导入模块目录里的,gevent目录,里的monkey模块的patch_all()方法
    import gevent   #导入协程模块
    import requests #模拟浏览器请求模块
    
    def f(url): #创建协程执行函数
        print('GET: %s' % url)  #打印字符串格式化GET:+函数参数url
        resp = requests.get(url)    #将url发送http请求
        data = resp.text    #获取http字符串代码
        print('%d 请求返回 %s.' % (len(data), url)) #打印字符串格式化,http字符串代码字符串数和url地址
    
    """
    相当于三条协程同时发url请求,那条协程先完成请求就先获取那条协程的返回数据
    也就是,协程在发IO请求时不会等待发送的请求返回数据完成,就自动切换下一条线程开始发下一条请求了,所有协程请求发完后,那条请求先返回数据完成,就先获取那条请求的数据
    """
    gevent.joinall([    #创建协程对象
            gevent.spawn(f, 'https://www.python.org/'), #创建协程,传入执行函数和执行函数参数
            gevent.spawn(f, 'https://www.yahoo.com/'),  #创建协程,传入执行函数和执行函数参数
            gevent.spawn(f, 'https://github.com/'), #创建协程,传入执行函数和执行函数参数
    ])

     遇到IO操作自动切换原理图

  • 相关阅读:
    Centos7 安装RabbitMQ 3.6.1
    面向对象编程(类的绑定方法与非绑定方法)
    面向对象编程(封装、封装的意义、封装与扩展性、@property)
    函数式编程(__slots__)
    面向对象编程(实例属性、类属性)
    面向对象编程(获取对象信息)
    面向对象编程(继承、多态)
    函数式编程(访问限制)
    面向对象编程(类与实例)
    面向对象编程(基础简介)
  • 原文地址:https://www.cnblogs.com/adc8868/p/5924941.html
Copyright © 2011-2022 走看看